diff --git a/AGENTS.md b/AGENTS.md index a518fd49..e897c762 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -165,8 +165,8 @@ step-by-step checklist. - **fleet/**: The Fleet Manager (`conductor stop` run-record scope, `conductor fleet`) — see `docs/fleet.md` for the user-facing guide. Fixes the bug where a plain `conductor run` (no `--web-bg`) was invisible to `conductor stop`/discovery: every run path now writes a `run_id`-keyed JSON record (not the legacy port-keyed `.pid` file `cli/pid.py` used to write) describing its mode/PID/workflow/port, so foreground, `--web-bg`, and `--web` runs are all discoverable the same way. - `records.py` - `RunRecord` (nine fields, no more — a tenth `tty` field was considered and rejected as POSIX-only with `pid` already sufficient; `mode` is a `RunMode = Literal["fg", "fg-web", "bg"]` so the single write site is checked by `ty`, and an *unrecognised* mode read from disk normalises to `"bg"` rather than raising — a raise reaches `_read_and_prune` as `corrupt`, which deletes **without checking liveness**, so a newer Conductor's mode would make an older one delete a live run's record) + `write_run_record` / `read_run_records` / `read_run_record` / `remove_run_record` / `remove_run_record_for_current_process`. `read_run_records()` filters to processes that pass `cli.pid.is_process_alive` and tolerates legacy port-keyed `.pid` files (surfaced as `mode="bg"` records) alongside corrupt/partial JSON. - `resume.py` - The fleet layer's first checkpoint consumer (issue #460): `correlate_checkpoints(entries)` joins History's `HistoryEntry` rows against `engine/checkpoint.py::CheckpointManager.list_checkpoints`'s on-disk checkpoints for the History screen's Resume action. The join key is the **event log path** (normalized via `os.path.realpath`, since both sides ultimately derive from `tempfile.gettempdir()`), with `run_id` as a documented fallback that is refused whenever that id is ambiguous across the scanned entries (a nested `conductor` invocation inherits `CONDUCTOR_RUN_ID`, so two logs can share one id). Gating a row's Resume availability is **checkpoint existence + the checkpoint's recorded `workflow_path` existing on disk — never `outcome`**: an `unknown` row from a crash is exactly the case a periodic checkpoint exists for, and a `completed` row can correlate to a stale checkpoint too (re-executing already-finished work is accepted, mitigated only by surfacing the checkpoint's provenance in the UI, not by hiding the key). Kept out of `fleet/history.py` deliberately — that module's docstring is emphatic that History is derived from event logs alone, and `build_history_entries` is patched by name in several existing tests. - - `summary.py` - Derives a `RunSummary` (status/current-step/elapsed/tokens/cost/gate) from a **bounded tail** read of a run's JSONL event log (cheap enough to call once per row on the Runs screen's ~2s poll) plus a separate bounded **full**-log read (capped at 8 MiB from the start of the file) for the run-detail screen's per-agent history — an unusually large log's trailing history can fall outside that cap and be omitted, the same accepted trade-off the bounded tail read makes at the other end. Status is derived from explicit event markers (`gate_presented`/`gate_resolved`/`workflow_completed`/`workflow_failed`), never inferred from timing alone — the design's own measurement found that unreliable (228 false positives, 0 true positives). `gate_resolvable` (Fleet Manager D4) is computed once here: `True` whenever the record has a dashboard port (`fg-web`/`bg`), `False` for `mode == "fg"`. - - `history.py` - Enumerates every retained run directly from `$TMPDIR/conductor/*.events.jsonl` files (not run records, which are already gone by the time a run is history) for the History screen, regardless of outcome. Classifies each log by its terminal event (`workflow_completed`/`workflow_failed`); a log with neither is `"unknown"`, **never** `"running"` — the same non-inference constraint `summary.py` follows, sharpened here since there is no run record to fall back on at all. The returned list is bounded by `[fleet.retention].keep_last` and, independently, by a fixed 200-entry display cap (so a `keep_last` configured below 1 — "unbounded" for the pruning sweep — still cannot grow the History screen without limit). Each log is read through a single-pass generator (`_read_full_log` yields one parsed event at a time to `_scan_history_events`) with no byte cap, unlike `summary.py`'s bounded readers — deliberately, because History reads a completed log exactly once and a truncated read would misclassify a finished run as `"unknown"` with incomplete totals; `_scan_history_events` must therefore stay a single forward pass over its `Iterable` argument (issue #436). + - `summary.py` - Derives a `RunSummary` (status/current-step/elapsed/tokens/cost/gate) from a **streamed, uncapped** read of a run's JSONL event log (issue #485) — `stream_event_log` is bounded only by the longest single line, not by file size or event count, replacing three separate bounded windows (a 512 KiB tail, a 512 KiB head recovery read, and an 8 MiB full-log cap) that a long or resumed run had already outgrown in production (a real 9.72 MB / 20,361-line log lost its current step, its token/cost totals, and — via the head window never firing for a *resumed* run's second `workflow_started` — reported the wrong topology). The Runs screen's ~2s poll passes `keep_types=_SUMMARY_EVENT_TYPES` so the reader skips an uninteresting line via a cheap regex prefilter without JSON-parsing it (12.5 ms measured against that same 9.72 MB log, vs. 65 ms unfiltered); the run-detail/step-detail screens read unfiltered since they need every event type. `_scan_events` is **generation-aware**: a resumed run always writes a second root `workflow_started` into the same log (the engine's own re-emit, or the dashboard-seeding path's synthesized copy), and reaching one resets `status`/`gate`/open steps and overwrites `topology`/`workflow_name`/`cwd`/`inputs` with the new generation's own — but does **not** reset token/cost totals, which accumulate across every generation (a resumed run's lifetime usage, not just its latest attempt's). Status is derived from explicit event markers (`gate_presented`/`gate_resolved`/`workflow_completed`/`workflow_failed`), never inferred from timing alone — the design's own measurement found that unreliable (228 false positives, 0 true positives). `gate_resolvable` (Fleet Manager D4) is computed once here: `True` whenever the record has a dashboard port (`fg-web`/`bg`), `False` for `mode == "fg"`. + - `history.py` - Enumerates every retained run directly from `$TMPDIR/conductor/*.events.jsonl` files (not run records, which are already gone by the time a run is history) for the History screen, regardless of outcome. Classifies each log by its terminal event (`workflow_completed`/`workflow_failed`); a log with neither is `"unknown"`, **never** `"running"` — the same non-inference constraint `summary.py` follows, sharpened here since there is no run record to fall back on at all. The returned list is bounded by `[fleet.retention].keep_last` and, independently, by a fixed 200-entry display cap (so a `keep_last` configured below 1 — "unbounded" for the pruning sweep — still cannot grow the History screen without limit). `_read_full_log` (a single-pass generator feeding `_scan_history_events`) now delegates its actual line-reading to `summary.stream_event_log` (issue #485), which generalized the uncapped-streaming approach History pioneered first; it re-derives its own corrupt-vs-empty distinction from `path.stat().st_size` rather than a raw non-blank-line count, since the shared reader has no reason to track that for a live run's cheap repeated poll. `_scan_history_events` takes the **latest** root `workflow_started`'s timestamp as `started_at` (issue #485, Q2) — mirroring `summary.py`'s generation-reset — so a resumed run's `duration_seconds` fallback (`ended_at - started_at`, used absent an engine-reported `elapsed`) measures the current attempt, not the idle gap since the original one; `_scan_history_events` must therefore stay a single forward pass over its `Iterable` argument (issue #436). - `launch.py` - Resolves a file path or registry reference (reusing `registry/resolver.py::resolve_ref` + `registry/cache.py::resolve_and_fetch`, the same pair `conductor show` uses) and calls `cli.bg_runner.launch_background()` directly for the New Run screen — never re-implements detached process spawning, per the design's explicit warning that doing so would make a TUI-launched run die with the TUI. `launch_resume(checkpoint_path, ...)` (issue #460) is the same pattern for the History screen's Resume action: it calls `cli.bg_runner.launch_background_resume(workflow_path=None, checkpoint_path=...)` directly rather than re-implementing detached spawning, since a `HistoryEntry` carries no workflow path of its own and the checkpoint is the only route to one. Both `resolve_workflow` and `launch_workflow`/`launch_resume` take the Fleet Manager's launch directory as an explicit argument (issue #477) -- `base_dir` on the former, `cwd` on the latter two, forwarded to `cli.bg_runner.launch_background()`/`launch_background_resume()` -- rather than ever reading the process cwd, so this module stays a pure function of its arguments; `fleet/tui/app.py::FleetApp.launch_dir` is the one place that directory is actually decided. `resolve_workflow` joins a relative *file* reference onto `base_dir` (a registry reference and an absolute reference both ignore it) and always returns an absolute `ResolvedWorkflow.path` -- `Path(os.path.abspath(...))`, not `.resolve()`, matching `_resolve_agent_working_dir`'s "normpath, not resolve" convention so a symlinked project directory stays the alias the user typed. Absolutising is not optional: `launch_background` puts `str(workflow_path)` straight into a detached child's argv, and a relative path there resolves against whatever cwd that child happens to inherit, not the directory it was typed against. - `retention.py` - `prune_event_logs(keep_last, dry_run)` bounds `$TMPDIR/conductor/*.events.jsonl` (Fleet Manager D3), mirroring `CheckpointManager.rotate_periodic_checkpoints`'s `keep_last` vocabulary. Never deletes the `checkpoints/` subdirectory or an event log a live/resuming run still references; a retained/live log's `.bg.stderr.log`/`.bg.stdout.log` companions are kept or pruned alongside it. `maybe_prune_event_logs()` is the opportunistic-startup-sweep wrapper `cli/run.py` calls, gated by `[fleet.retention].enabled` and never raising. - `tui/` - The Textual app (`app.py`, `Screen` push/pop stack) and its screens (`screens/runs.py` home; `screens/run_detail.py`; `screens/step_detail.py`; `screens/providers.py`; `screens/registries.py`; `screens/new_run.py`; `screens/history.py`; `screens/splash.py`), plus `actions.py` (shared stop/kill/gate-respond logic, reusing `cli/app.py::stop_records` and `cli/gate.py::_gate_respond_impl` rather than duplicating either), `theme.py` (the single status glyph/label/colour vocabulary — screens must not define local glyph maps), `dag.py` (step-status chips), `anim.py` (pure frame→glyph functions; animation is off by explicit `CONDUCTOR_FLEET_NO_ANIM`, on by explicit `CONDUCTOR_FLEET_ANIM`, off by default on a detected RDP session (`SESSIONNAME` starting `RDP-Tcp`) via `is_remote_session()`, and on otherwise — `CONDUCTOR_FLEET_NO_ANIM` always wins. **SSH is deliberately not detected**: it ships the ANSI byte stream for the local terminal to render (a few hundred bytes per frame), where RDP renders remotely and diffs/encodes/ships changed *pixel regions*, so only the latter is costly — and the real trigger is a *slow* link, for which there is no signal, only "SSH at all", which is usually fast. `CONDUCTOR_FLEET_NO_ANIM` is the remedy there and for VNC/Citrix/xrdp; a negative test pins this so it is not re-added as an oversight; any path that disables animation — explicit `CONDUCTOR_FLEET_NO_ANIM` or detection — also sets Textual's own `App.animation_level = "none"` in `app.py::on_mount`, issue #462), `art.py`, `widgets.py`, and `notify.py` (terminal bell / OSC 9 notifications, debounced to fire once per status transition). Optional: imported only by `conductor fleet`'s bare invocation, gated behind the `tui` extra (floor `textual>=8.0` — `widgets.py::BlockFooter` is written against 8.x `Footer` group rendering). The install command for that extra is resolved per install context by `install_hint.py`, never hardcoded. **No blocking I/O on the event loop** (issue #437): every screen that reads the filesystem *on a timer, or on a path that can be slow*, does it in a worker (`@work async def` + `await asyncio.to_thread()`), with the render half running on the event loop after the `await` — the pattern `screens/new_run.py::action_resolve`/`action_launch`, `screens/step_detail.py::load_step` and `screens/registries.py::load_workflows` established first, now also `screens/runs.py::_refresh_worker`/`_open_dashboard_worker`, `screens/run_detail.py::_refresh_worker`, `screens/history.py::load_history`, and (not a screen, but the same reasoning) `actions.py::kill_runs`. Two deliberate exceptions, both documented at their definition: `registries.py::load_registries` reads one local `registries.toml` inline, and `providers.py::load_providers` is a `@work` method awaiting a natively async `gather()` — note that `gather()` still does its offline SDK-availability imports synchronously, so it is *not* yet an instance of this pattern. The **two polled** screens (`runs.py`, `run_detail.py`) additionally keep their `refresh_runs()`/`refresh_detail()` entry point as a **synchronous dispatcher** (still callable from `on_mount`, `set_interval`, and action handlers) guarded by a plain `_refreshing` boolean, so a tick arriving mid-scan is **dropped** rather than started alongside the in-flight one — deliberately not `@work(exclusive=True)`, which would cancel the in-flight worker ("newest wins") instead of skipping the newer tick. Load-once screens (`history.py::load_history`) need no guard and are `@work` methods directly. The flag is set in the *dispatcher*, not as the worker's first line, because a `@work` body doesn't start until Textual schedules it; and it is released in a `finally`, because without that one transient error stops the screen refreshing for the rest of the session, silently and indistinguishably from a calm fleet. An **explicit** refresh (`_kill_and_refresh`, `action_resolve_gate`'s `finally`) passes `explicit=True` and is *coalesced* via `_refresh_pending` rather than dropped: those callers promise the table reflects what they just did, and the scan they collide with started before it. Every screen shows `theme.py::loading_text()` in `compose()` until its first result lands. **A failed load is surfaced, never degraded into an empty state** — rendering "No runs" or "No run history yet" for an unreadable directory is a positive claim of absence that reads as success, so `runs.py` distinguishes an empty fleet from one whose summaries all failed (`RunScan.failed`/`seen_run_ids`) and `history.py` follows `step_detail.py`/`registries.py`'s red-error-line convention. `RunScan.seen_run_ids` carries *every* record read, not just the rows that rendered, because the notifier is pruned against it — pruning against the rendered rows made a run whose summary briefly failed look brand-new and re-fire its gate notification. Test caution: a pilot test that presses a key opening a modal (kill confirm, gate options), or that deliberately suspends a worker on a `threading.Event`, and then needs a second keypress to resolve it, must use a plain `await pilot.pause()` between the two, never `tests/test_fleet/conftest.py::settle` — `settle` awaits `app.workers.wait_for_complete()`, and the suspended `@work` method won't finish until that second keypress, so awaiting it first deadlocks the test. `history.py` also correlates checkpoints in the same `load_history` worker as a second thread hop (issue #460, `conductor.fleet.resume.correlate_checkpoints`), gating its `r`/Resume binding via `check_action` the same way `runs.py` gates `g`/Gate -- hidden outright when the highlighted row has no correlated checkpoint, refreshed on `on_data_table_row_highlighted` so the footer never shows yesterday's answer as the cursor moves. **The frame timer (~10fps, `runs.py::_tick`) may only repaint what actually moves** (issue #462): the animated table cells and the preview pane's `#run-preview-score` widget. Everything else in the preview pane (the gate section, the `Progress N/M` header), and the footer's `refresh_bindings()`, belong to the ~2s data poll (`_update_gate_detail`) and to selection changes — that is the invariant a future edit is most likely to break by re-adding a convenient `_update_gate_detail()` call to `_tick`. `app.py::FleetApp.launch_dir` is an app-level, **process-lifetime-only** Textual `reactive[Path]` (issue #477) -- no `config.toml` key, no state file, reset to the process's cwd on the next `conductor fleet` -- read by `screens/new_run.py` (the base a relative workflow reference resolves against, and the `cwd` a launched run's detached child inherits) and mutated only through `FleetApp.set_launch_dir`, matching this module's `push_*`/`return_to_runs` one-named-mutation-site convention. `actions.py::DirectoryPickerModal` (opened via the shared `change_launch_directory`, bound to `d` on Runs and `ctrl+d` -- `priority=True`, since `Input` binds it first -- on New Run) is the only way to change it; a bad path is rejected in place (a red message line, modal stays on the stack) rather than dismissed. `widgets.py::BlockFooter` passes `show_command_palette=False` to `Footer.__init__` (a real 8.x kwarg) so the Runs footer's docked `^p palette` key -- not `ctrl+p` itself, which still opens it -- is hidden, reclaiming the columns `d Dir` needed once `Providers`/`Registries` were also shortened to `Prov`/`Regs`. diff --git a/src/conductor/cli/fleet.py b/src/conductor/cli/fleet.py index 010b2534..8152678f 100644 --- a/src/conductor/cli/fleet.py +++ b/src/conductor/cli/fleet.py @@ -132,7 +132,7 @@ def list_runs() -> None: # stays coarse-grained ("running") rather than deriving the # richer gate/status vocabulary `derive_run_summary` (E6) # computes for the TUI's Runs screen, which also requires a - # bounded event-log tail read per row; use `conductor fleet`'s + # streamed event-log scan per row; use `conductor fleet`'s # TUI (or `--web`) for the finer-grained status. "running", str(record.pid), diff --git a/src/conductor/fleet/history.py b/src/conductor/fleet/history.py index 6878f674..8b7da566 100644 --- a/src/conductor/fleet/history.py +++ b/src/conductor/fleet/history.py @@ -33,15 +33,14 @@ from __future__ import annotations -import json import logging -import math import re from collections.abc import Iterable, Iterator from dataclasses import dataclass from pathlib import Path from typing import Any, Literal +from conductor.fleet import summary from conductor.fleet.retention import event_log_root from conductor.run_id import RUN_ID_PATTERN_SOURCE @@ -110,11 +109,15 @@ class HistoryEntry: to still be alive (see the module docstring).""" started_at: float | None - """Unix timestamp of the ``workflow_started`` event, or ``None`` if - the log has no such event at all (an unrecognized/garbled file, or - one that hasn't recorded it yet) -- the full log is always read - (:func:`_read_full_log`), so this is never unknown merely because - the event was far from the end of the file.""" + """Unix timestamp of the **latest** ``workflow_started`` event, or + ``None`` if the log has no such event at all (an unrecognized/garbled + file, or one that hasn't recorded it yet) -- the full log is always + read (:func:`_read_full_log`), so this is never unknown merely because + the event was far from the end of the file. On a resumed run this is + the current attempt's start time, not the original one (issue #485, + Q2): since this field isn't itself a displayed column, the effect is + on :attr:`duration_seconds`'s fallback, which stops spanning the idle + gap between a resume's generations.""" ended_at: float | None """Unix timestamp of the terminal event, or ``None`` if there is none.""" @@ -162,18 +165,13 @@ def _parse_filename(path: Path) -> tuple[str, str | None]: def _finite_float(value: Any) -> float | None: """Return ``value`` as a ``float`` iff it is a finite ``int``/``float``. - ``NaN``/``Infinity``/``-Infinity`` are valid JSON values (Python's - ``json`` module accepts them by default) but are not legitimate - timestamps, durations, token counts, or costs -- letting one through - would silently poison a sum or crash the History screen's duration - formatting (``int(nan)``/``int(inf)`` both raise) downstream (E14 - review round 1). Rejected the same way a wrong-shaped value already - is: silently ignored, not raised. + Delegates to :func:`conductor.fleet.summary._finite_float`, which + enforces the identical rule (``NaN``/``Infinity`` are valid JSON but + not legitimate timestamps/durations/token counts/costs) over the same + engine-written event payloads -- hosted there since this module + already imports :mod:`conductor.fleet.summary`. """ - if not isinstance(value, int | float): - return None - value = float(value) - return value if math.isfinite(value) else None + return summary._finite_float(value) @dataclass @@ -204,18 +202,26 @@ def _scan_history_events(events: Iterable[dict[str, Any]]) -> _ScanResult: :func:`_read_full_log` stream rather than materialise a list. A future edit must not break this by, say, iterating ``events`` twice. - A **resumed** run appends a fresh ``workflow_started`` after an - earlier terminal event without a dashboard attached (the engine only - suppresses that re-emit when seeding a dashboard on resume -- - ``cli/run.py``), so a log can legitimately contain - ``workflow_started`` -> ... -> ``workflow_failed`` -> a *second* - ``workflow_started`` -> more activity. Any ``workflow_started`` past - the first is therefore treated as the start of a new root execution - generation: the previously recorded terminal ``outcome``/``ended_at``/ + A **resumed** run always appends a fresh ``workflow_started`` after an + earlier terminal event, whether or not a dashboard is attached -- + ``cli/run.py`` suppresses only the *engine's own re-emit* (so a live + dashboard does not double-count nested-workflow depth), then writes + the same payload directly to the event-log subscriber, bypassing the + emitter, so the persisted JSONL always records the generation boundary + either way. So a log can legitimately contain ``workflow_started`` -> + ... -> ``workflow_failed`` -> a *second* ``workflow_started`` -> more + activity. Any ``workflow_started`` past the first is therefore treated + as the start of a new root execution generation: the previously + recorded terminal ``outcome``/``ended_at``/ ``reported_elapsed`` are reset back to their "no terminal event yet" defaults, so an in-progress resumed attempt reads as ``"unknown"`` (never the stale prior attempt's outcome) until *its own* terminal - event is seen (E14 review round 1). + event is seen (E14 review round 1). ``started_at`` is likewise taken + from the **latest** ``workflow_started`` (issue #485, Q2): a resumed + run's :attr:`HistoryEntry.duration_seconds` fallback (``ended_at - + started_at``, used when no engine-reported ``elapsed`` is available -- + e.g. a resumed-then-failed run) should measure the current attempt, + not span the idle gap since the very first one. """ result = _ScanResult() seen_workflow_started = False @@ -243,8 +249,8 @@ def _scan_history_events(events: Iterable[dict[str, Any]]) -> _ScanResult: result.reported_elapsed = None else: seen_workflow_started = True - if ts is not None: - result.started_at = ts + if ts is not None: + result.started_at = ts elif etype == "workflow_completed": result.outcome = "completed" @@ -289,89 +295,71 @@ def _read_full_log(path: Path) -> Iterator[dict[str, Any]]: """Stream-parse every line of a retained event log, oldest first, yielding one parsed event dict at a time. - Unlike ``fleet.summary``'s bounded tail/head readers -- built for a - *live* run's cheap, repeated ~2s poll, or a bounded detail view -- - History only ever reads a log once, after the run is already done, so - there is no reason to accept a bounded reader's truncation trade-off - here. A byte-capped read can silently omit an early token/cost event - or discard an oversized terminal event that falls outside the window, - presenting a genuinely completed run as ``"unknown"`` with an - incomplete total (E14 review round 1). This function is a generator: - neither the raw bytes nor the parsed events are ever fully - materialised into a list, so memory is proportional to the largest - single line, not to the log's size (retained state beyond that one - line/dict is just two booleans). Contrast - ``conductor.fleet.summary.read_event_log_full``, which caps its read - at 8 MiB and therefore trades coverage for a bounded (but nonzero) - memory footprint; streaming gives History unbounded coverage and - bounds memory against event *count*, so it needs no equivalent - count-based cap -- but a single oversized line is still read and - parsed whole (``for raw_line in f`` reads up to the next newline), so - this is not an unconditional memory bound. The two readers' differing - bounds are a deliberate consequence of their differing consumers (a - live-run detail view vs. a done-run, read-once history), not an - oversight. - - The generator holds the file handle open (inside its ``with``) until - it is exhausted, closed, or garbage collected -- that ``with`` block, - not any particular caller, owns the handle's lifetime. The sole - consumer in production, :func:`_build_entry`, drains it to completion - via :func:`_scan_history_events`; tests may consume it directly and - abandon it early, which is exactly why the handle's release does not - depend on caller behavior. + Delegates the actual line-reading and JSON-parsing to + :func:`conductor.fleet.summary.stream_event_log` (issue #485), which + made this exact choice -- an uncapped, streamed read bounded by the + longest single line rather than by file size or event count -- for + History first and generalized it for the Runs/run-detail screens' + former bounded tail/head/full-log readers to share. History still + reads a log once, after the run is already done, so there is no + reason to accept a bounded reader's truncation trade-off: a byte-capped + read can silently omit an early token/cost event or discard an + oversized terminal event outside its window, presenting a genuinely + completed run as ``"unknown"`` with an incomplete total (E14 review + round 1). + + The generator holds the file handle open (inside ``stream_event_log``'s + own ``with``) until it is exhausted, closed, or garbage collected -- + that ``with`` block, not any particular caller, owns the handle's + lifetime. The sole consumer in production, :func:`_build_entry`, drains + it to completion via :func:`_scan_history_events`; tests may consume it + directly and abandon it early, which is exactly why the handle's + release does not depend on caller behavior. Because this is a generator, none of its side effects happen when ``_read_full_log(path)`` is called -- they happen while the returned - iterator is being **consumed**. In particular, ``open()`` and any - ``OSError`` it raises (permission denied, the file vanishing - mid-scan, or any other read failure) surface on the first - ``next()``, not at call time. Unlike ``read_event_log_tail``'s - never-raise contract, such a failure is deliberately **not** - swallowed: it propagates so :func:`build_history_entries`'s - per-file guard can tell "genuinely no events" apart from "couldn't - read this file at all" and skip the latter, rather than presenting a - fabricated ``"unknown"`` entry for a log it never actually read - (E14-T4 / E14 review round 1). + iterator is being **consumed**. In particular, ``open()`` (and any + ``OSError`` it raises: permission denied, the file vanishing mid-scan, + or any other read failure) surfaces on the first ``next()``, not at + call time -- the delegated-to ``stream_event_log`` is always the + *first* thing this generator's body does, before any size check, so a + failure opening the file is never masked by an empty-file fast path. + Unlike ``fleet.summary``'s former bounded readers' never-raise + contract, such a failure is deliberately **not** swallowed: it + propagates so :func:`build_history_entries`'s per-file guard can tell + "genuinely no events" apart from "couldn't read this file at all" and + skip the latter, rather than presenting a fabricated ``"unknown"`` + entry for a log it never actually read (E14-T4 / E14 review round 1). A malformed individual line (bad JSON, a truncated write caught - mid-flush) is tolerated the same way the bounded readers already do - -- skipped rather than aborting the whole read, with a single - aggregate warning logged if any lines were skipped (partial - corruption still produces an entry, but not silently). But a - **non-empty** file (at least one non-blank line) that yields **zero** - parseable events is corrupt, not "legitimately empty" -- E14-T4 - requires a corrupt log to be skipped, not shown as an ordinary - ``"unknown"`` entry (E14 review round 2). Raises - :class:`_CorruptEventLogError` in that case, once the stream is - exhausted; a genuinely empty file (no non-blank lines at all) still - yields nothing and raises nothing, which :func:`_build_entry` - legitimately turns into an ``"unknown"`` entry. + mid-flush) is tolerated by the shared reader the same way it always + has been here -- skipped rather than aborting the whole read. But a + **non-empty** file that yields **zero** parseable events is corrupt, + not "legitimately empty" -- E14-T4 requires a corrupt log to be + skipped, not shown as an ordinary ``"unknown"`` entry (E14 review + round 2). Raises :class:`_CorruptEventLogError` in that case, once the + stream is exhausted; a genuinely empty (zero-byte) file still yields + nothing and raises nothing, which :func:`_build_entry` legitimately + turns into an ``"unknown"`` entry. Distinguishing the two now checks + ``path.stat().st_size`` **after** the read completes with nothing + yielded (rather than "at least one non-blank raw line", this + function's own former signal) -- deliberately after, not before, + because checking size first would let a permission error on a + zero-byte file skip the read (and the ``OSError`` it should raise) + entirely. The shared reader has no reason to track skipped-line + counts for a live run's cheap, repeated poll, so this function no + longer can either. A file containing only blank/whitespace lines -- + untested, and not known to occur in practice, since every write here + is a single JSON object per line -- would now read as "corrupt" rather + than "empty"; a zero-byte file (the only shape a genuinely fresh or + truncated-at-creation log actually takes) is unaffected. """ - saw_nonblank_line = False yielded_any = False - skipped_lines = 0 - with open(path, "rb") as f: - for raw_line in f: - line = raw_line.strip() - if not line: - continue - saw_nonblank_line = True - try: - obj = json.loads(line.decode("utf-8")) - except (UnicodeDecodeError, ValueError, RecursionError): - skipped_lines += 1 - continue - if isinstance(obj, dict): - yielded_any = True - yield obj - if saw_nonblank_line and not yielded_any: + for evt in summary.stream_event_log(path): + yielded_any = True + yield evt + if not yielded_any and path.stat().st_size > 0: raise _CorruptEventLogError(f"{path}: non-empty log with no parseable events") - if skipped_lines: - logger.warning( - "%s: skipped %d unparseable line(s); this run's totals may be incomplete", - path, - skipped_lines, - ) def _build_entry(path: Path) -> HistoryEntry: diff --git a/src/conductor/fleet/summary.py b/src/conductor/fleet/summary.py index d9d95be1..6695c566 100644 --- a/src/conductor/fleet/summary.py +++ b/src/conductor/fleet/summary.py @@ -1,15 +1,29 @@ -"""``RunSummary`` derivation from a run record plus its event-log tail. +"""``RunSummary`` derivation from a run record plus its event log. Fleet Manager E6 (see ``docs/projects/fleet-manager/fleet-manager.design.md``, *Implementation* → ``summary.py``): given a live :class:`~conductor.fleet.records.RunRecord` (as returned by :func:`conductor.fleet.records.read_run_records`), derive a :class:`RunSummary` — the status vocabulary, current step, elapsed-on-step, -token/cost totals, and any open human gate — from a **bounded tail** of the -run's JSONL event log rather than the whole file, so this can be called once -per row on a ~2-second poll loop (E7's Runs screen) without the cost of a -whole-file load growing with the run's lifetime (contrast -``web/replay.py::_load_events``, which loads the entire file and is the wrong -tool for this). +token/cost totals, and any open human gate — from the run's JSONL event log, +so this can be called once per row on a ~2-second poll loop (E7's Runs +screen) without silently dropping state a long or resumed run has already +outgrown a bounded read window (issue #485). + +This module used to bound its reads to a 512 KiB tail (list screen), a +512 KiB head (topology recovery for a run whose ``workflow_started`` had +aged out of that tail), and an 8 MiB cap (run-detail/step-detail screens). +Every one of those windows was sized against logs that reality then +outgrew: a run long enough to be interesting silently lost its current +step, its token/cost totals, and (on a resumed run) reported a stale +*prior* generation's terminal status. :func:`stream_event_log` replaces +all three with one **streaming, uncapped** reader — the same choice +:mod:`conductor.fleet.history`'s ``_read_full_log`` already made for the +same reason (an accepted-then-exceeded cap silently truncating live data +is worse than an unbounded read that costs proportionally more CPU). +Memory is bounded by the longest single line, not by the file's size or +event count; a 12.5 ms scan of a real 9.72 MB / 20,361-line log (with the +prefilter below) is comfortably inside the Runs screen's worker-thread +poll budget. **Liveness is not re-derived here.** The design's own measurement found inferring "is this run still running" from the event stream alone unreliable @@ -29,8 +43,11 @@ import json import logging +import math +import re import time from collections import deque +from collections.abc import Iterable, Iterator from dataclasses import dataclass, field from datetime import datetime from pathlib import Path @@ -44,34 +61,29 @@ AgentDetailStatus = Literal["pending", "running", "at-gate", "completed", "failed"] -# Bounded read windows. Neither grows with the file's size, satisfying the -# "bounded, not whole-file" requirement even for a very long-running -# workflow's log. 512 KB comfortably covers a typical run's entire event -# history (events are small JSON lines), so token/cost totals and current-step -# tracking are correct in the common case; for an unusually long run whose -# log has grown past this window, older completed-agent totals age out of the -# tail and are undercounted — an accepted, documented limitation consistent -# with this project's other "known data gaps" (see the module docstring's -# liveness note and D5 in the plan). -_DEFAULT_TAIL_BYTES = 512 * 1024 - -# Bound for the run-detail screen's *full*-log read (E9-T3). Unlike the tail -# reader above, this reads from the start of the file so every agent's -# history (not just the most recent window) is available for the per-agent -# rows the detail screen renders. 8 MB is comfortably larger than any -# realistic single-run event log (the design's own directory-wide -# measurement was 12 MB across 1522 files) while still bounding memory use -# against a pathological log; a log that exceeds this bound has its -# *trailing* history (including, in the worst case, the run's own -# `workflow_completed`/`workflow_failed`) silently truncated rather than -# read in full -- an accepted, documented limitation, not a crash. -_DEFAULT_FULL_LOG_MAX_BYTES = 8 * 1024 * 1024 - _STEP_ACTIVITY_LIMIT = 200 """How many activity lines the step drill-down keeps. A long agentic loop emits hundreds of tool calls; this is a drill-down, not a log viewer.""" +def _finite_float(value: Any) -> float | None: + """Return ``value`` as a ``float`` iff it is a finite ``int``/``float``. + + ``NaN``/``Infinity``/``-Infinity`` are valid JSON values (Python's + ``json`` module accepts them by default) but are not legitimate token + counts or costs -- letting one through would silently poison a sum or + crash downstream formatting (``int(nan)``/``int(inf)`` both raise). + Rejected the same way a wrong-shaped value already is: silently + ignored, not raised. Shared with :mod:`conductor.fleet.history`, which + already imports this module and enforces the identical rule over the + same engine-written event payloads. + """ + if not isinstance(value, int | float): + return None + value = float(value) + return value if math.isfinite(value) else None + + @dataclass(frozen=True) class GateInfo: """The open human gate carried onto a :class:`RunSummary`. @@ -156,19 +168,25 @@ def elapsed_seconds(self, now: float | None = None) -> float | None: @dataclass(frozen=True) class RunDetail: """Full per-agent detail for the run-detail screen (E9), derived from a - bounded **full** log read (:func:`read_event_log_full`) rather than the - tail window :class:`RunSummary` uses -- the list screen's ~2s poll stays - on the cheaper tail path; only opening the detail screen pays for a - fuller read, and only once per open (not on every poll tick).""" + streamed, uncapped read of the run's event log (:func:`stream_event_log`) + -- the same reader :class:`RunSummary` uses, over every event rather + than a tail window. The distinction from :class:`RunSummary` is not + frequency but filtering: :func:`derive_run_summary` passes + ``keep_types=_SUMMARY_EVENT_TYPES`` (12.5 ms measured against a real + 9.72 MB / 20,361-line log) while this class's unfiltered scan costs + roughly 5x that (65 ms on the same log) because it also needs + per-agent event types the Runs screen's aggregate totals do not. The + scan runs on every ~2s poll tick for as long as the run-detail screen + is open, not once per open -- kept off the event loop in a worker for + that reason.""" run_id: str workflow_name: str topology: RunTopology | None - """``None`` when the log is missing/unreadable, empty, or its - ``workflow_started`` event fell outside the (bounded) read window -- - the detail screen renders a placeholder in this case rather than an - empty table (E9-T5).""" + """``None`` when the log is missing/unreadable, empty, or has not + (yet) written a ``workflow_started`` event -- the detail screen + renders a placeholder in this case rather than an empty table (E9-T5).""" agents: list[AgentDetail] """One row per :attr:`topology`'s agent, in the same (declared) order. @@ -196,7 +214,7 @@ class RunSummary: current_step: str | None """Name of the agent, parallel group, or for_each group most recently - started without a matching completion event seen in the tail. ``None`` + started without a matching completion event seen in the log. ``None`` when nothing is open (e.g. a log with no events yet, or a terminal status).""" @@ -205,16 +223,23 @@ class RunSummary: """Unix timestamp of the current step's start event, or ``None``.""" total_tokens: int - """Sum of ``tokens`` across every ``agent_completed`` event seen in the - tail. Per D5, this is **completed-agent tokens only** — there is no + """Sum of ``tokens`` across every ``agent_completed`` event in the log. + Per D5, this is **completed-agent tokens only** — there is no mid-flight usage event, so the agent currently running never - contributes here until it finishes.""" + contributes here until it finishes. A **lifetime total across every + generation** in the log (issue #485): a resumed run's totals include + whatever the prior, terminated generation(s) already accumulated, not + just the current one -- only the *status*/*current step*/*gate* reset + at a resume boundary, not the usage totals (:attr:`total_tokens`, + :attr:`total_cost_usd`, and :attr:`unpriced_agent_count` alike -- all + three accumulate through the same event-handling code path).""" total_cost_usd: float | None """Sum of ``cost_usd`` across priced ``agent_completed`` events in the - tail, or ``None`` if none were priced. Mirrors - ``WorkflowUsage.total_cost_usd``: never a confident total when - :attr:`has_unpriced` is true — see that attribute.""" + log (every generation -- see :attr:`total_tokens`), or ``None`` if none + were priced. Mirrors ``WorkflowUsage.total_cost_usd``: never a + confident total when :attr:`has_unpriced` is true — see that + attribute.""" unpriced_agent_count: int """Count of completed agents that consumed tokens but had no cost data @@ -236,21 +261,24 @@ class RunSummary: from the record's ``port`` — not re-derived per-screen.""" topology: RunTopology | None - """Run topology from ``workflow_started``, if that event happened to - fall within the read window (E6-T5). ``workflow_started`` is always - the first line of the log, so this is populated for a just-started run; - for an older run whose log has grown past the tail window, this is - ``None`` until a dedicated full-log read is added (E9-T3) for the - run-detail screen. A head read now recovers it for a long run, whose - log has always outgrown the tail window by definition.""" + """Run topology from the log's **latest** root ``workflow_started`` + event (E6-T5). ``workflow_started`` is always the first line of a + fresh log, and a resume always writes a second one into the same file + (the engine's own re-emit, or the dashboard-seeding path's synthesized + copy) -- so on a resumed run this is the *current* generation's + topology, not a stale earlier one (issue #485). ``None`` only when the + log has no root ``workflow_started`` at all (missing/unreadable log, + or one that hasn't written it yet).""" cwd: str | None = None """Directory conductor was launched from (``workflow_started``'s ``system`` block). Not on the run record, and what distinguishes two - runs of the same workflow started from different checkouts.""" + runs of the same workflow started from different checkouts. Like + :attr:`topology`, taken from the latest generation.""" inputs: dict[str, Any] | None = None - """The values this run was launched with, when the log records them.""" + """The values this run was launched with, when the log records them. + Like :attr:`topology`, taken from the latest generation.""" @property def has_unpriced(self) -> bool: @@ -287,143 +315,97 @@ def _parse_iso_timestamp(value: str) -> float | None: # --------------------------------------------------------------------------- -# Bounded JSONL reading (E6-T1) +# Streaming JSONL reading (issue #485) # --------------------------------------------------------------------------- - -def _parse_jsonl_bytes(raw: bytes) -> list[dict[str, Any]]: - """Parse ``raw`` as newline-delimited JSON objects, tolerating bad lines. - - Each line is decoded and parsed independently, so a single malformed - line (a truncated write caught mid-flush, invalid UTF-8, or a line cut - off by a bounded read's window boundary) is silently skipped rather - than aborting the whole parse — this is what lets the tail/head readers - tolerate a file being appended to concurrently. - """ - events: list[dict[str, Any]] = [] - for raw_line in raw.split(b"\n"): - line = raw_line.strip() - if not line: - continue - try: - obj = json.loads(line.decode("utf-8")) - except (UnicodeDecodeError, ValueError): - continue - if isinstance(obj, dict): - events.append(obj) - return events - - -_HEAD_BYTES = 512 * 1024 -"""How much of a log's *start* is read for topology. - -``workflow_started`` is the first event a run writes, so on any log longer -than the tail window it is the one event guaranteed to be outside it -- which -is why a long-running workflow's step list silently disappeared while a young -one still showed it. - -Sized to hold that one line, which is far larger than "one event" suggests: -it carries the whole workflow definition (every step, route, parallel and -for-each group, plus provider and metadata blocks), and a real 23-step -workflow measured **77 KB**. A window smaller than the line truncates it -mid-JSON, and a truncated line is discarded like any other malformed one -- -so an under-sized window does not degrade the result, it removes it -entirely, which is exactly the bug this constant exists to fix. -""" - - -def read_event_log_head(path: Path, *, head_bytes: int = _HEAD_BYTES) -> list[dict[str, Any]]: - """Read the first ``head_bytes`` of a JSONL event log, parsed into events. - - The mirror of :func:`read_event_log_tail`, and bounded the same way: a - trailing line cut off by the byte limit fails to parse and is skipped - like any other malformed line. - """ - try: - with open(path, "rb") as f: - chunk = f.read(head_bytes) - except OSError: - logger.debug("Could not read head of event log %s", path, exc_info=True) - return [] - return _parse_jsonl_bytes(chunk) - - -def read_event_log_tail( - path: Path, *, tail_bytes: int = _DEFAULT_TAIL_BYTES -) -> list[dict[str, Any]]: - """Read the last ``tail_bytes`` of a JSONL event log, parsed into event dicts. - - Seeks from the end of the file rather than loading it whole (E6-T1): - when the file is larger than ``tail_bytes``, seeks to - ``size - tail_bytes`` and discards the (likely partial) leading line - before parsing the rest, so a line straddling the seek boundary never - surfaces as a corrupt/garbled event. A trailing line cut short by a - concurrent in-progress write is tolerated the same way every other - malformed line is — it fails to parse and is silently skipped (see - :func:`_parse_jsonl_bytes`). - - Never raises: a missing file, a permission error, or any other - ``OSError`` while reading yields an empty event list rather than - propagating — a diagnostic read must not be able to crash a poll loop. - - Args: - path: Path to the JSONL event log. - tail_bytes: Maximum number of bytes to read from the end of the - file, regardless of the file's actual size. - - Returns: - Parsed event dicts, oldest first, from (at most) the last - ``tail_bytes`` of the file. - """ - try: - with open(path, "rb") as f: - f.seek(0, 2) - size = f.tell() - if size > tail_bytes: - f.seek(size - tail_bytes) - f.readline() # Discard the partial line at the seek boundary. - else: - f.seek(0) - raw = f.read() - except OSError: - return [] - return _parse_jsonl_bytes(raw) - - -def read_event_log_full( - path: Path, *, max_bytes: int = _DEFAULT_FULL_LOG_MAX_BYTES -) -> list[dict[str, Any]]: - """Read up to ``max_bytes`` from the *start* of a JSONL event log (E9-T3). - - Unlike :func:`read_event_log_tail` (which the Runs list screen's ~2s - poll uses so per-row reads stay cheap and bounded regardless of the - run's age), this is the **full**-log read used only by the run-detail - screen: it needs every agent's complete history -- including agents - that finished and aged out of a tail window -- not just the most - recent events. It is still bounded (not a literal whole-file read) so - a pathologically large log can't exhaust memory when a user opens the - detail screen; see :data:`_DEFAULT_FULL_LOG_MAX_BYTES` for the bound - and its trade-off. - - Never raises: mirrors :func:`read_event_log_tail`'s contract -- a - missing file, a permission error, or any other ``OSError`` yields an - empty event list rather than propagating. +# Whitespace-tolerant, anchored match of a JSONL event line's leading +# `"type"` key -- lets `stream_event_log`'s prefilter skip an uninteresting +# line without JSON-parsing it. Anchored (`.match`, not `.search`) so a +# `"type"` key appearing later in the object (e.g. inside `data`) can never +# be mistaken for the event's own type. Whitespace-tolerant because +# production log lines are written with `json.dumps(..., separators=(",", +# ":"))` (`engine/event_log.py`) -- no spaces at all -- while test fixtures +# typically use plain `json.dumps`, which inserts a space after each colon; +# a regex anchored to one exact spacing would make the fast path silently +# dead in tests while live in production. +_TYPE_PREFIX_RE = re.compile(rb'\{\s*"type"\s*:\s*"([A-Za-z0-9_]+)"') + + +def stream_event_log( + path: Path, *, keep_types: frozenset[str] | None = None +) -> Iterator[dict[str, Any]]: + """Stream-parse a JSONL event log, oldest first, yielding one parsed + event dict at a time. + + Replaces this module's former bounded tail/head/full-log readers + (issue #485): none of the three windows they used was actually safe + against a run outliving it, and every one of them did, in production, + silently and repeatedly. Memory here is bounded by the longest single + line in the file, not by the file's size or event count -- retained + state beyond that one line/dict is nothing. This mirrors + :mod:`conductor.fleet.history`'s ``_read_full_log``, which made the + same choice first, for the same reason: a byte-capped read can drop an + early token/cost event or discard an oversized terminal event outside + its window, silently corrupting the very state callers rely on. + + With ``keep_types``, a whitespace-tolerant anchored regex + (:data:`_TYPE_PREFIX_RE`) reads the ``type`` key off the raw bytes and + skips a line whose type is not in the set *without* JSON-decoding it -- + this is what keeps a per-row scan on the Runs screen's ~2s poll cheap + even against a very large log (12.5 ms measured against a real 9.72 MB + / 20,361-line log, versus 65 ms unfiltered). A line whose ``type`` key + is not first (so the regex fails to match) is always parsed rather + than skipped -- the prefilter is only ever an optimization, never a + second opinion about what a line means, so it can never silently drop + an event the writer happens to serialize with a different key order. + + This function is a generator: it holds the file handle open (inside + its own ``with``) only while being consumed, and none of its side + effects -- including ``open()`` and any ``OSError`` it raises (a + missing file, a permission error, or any other read failure) -- happen + until the returned iterator's first ``next()``. Unlike this module's + former bounded readers, this does **not** swallow ``OSError``; it + propagates, so a caller that wants the old "never raise" behavior must + wrap the *consumption* (e.g. inside :func:`_scan_events`'s ``for`` + loop), not just the call to this function -- see :func:`derive_run_summary`. + + A malformed individual line (bad JSON, a truncated write caught + mid-flush, invalid UTF-8) is tolerated the same way a concurrent + in-progress write always has been here: skipped rather than aborting + the whole read. Args: path: Path to the JSONL event log. - max_bytes: Maximum number of bytes to read from the start of the - file, regardless of the file's actual size. - - Returns: - Parsed event dicts, oldest first, from (at most) the first - ``max_bytes`` of the file. + keep_types: A best-effort prefilter, not an exact one: a line + whose leading ``type`` key matches :data:`_TYPE_PREFIX_RE` and + is not in this set is skipped without being JSON-parsed. A + line the regex cannot read (e.g. ``type`` is not the first + key) is always parsed and yielded regardless of this set -- + callers must still branch on ``type`` rather than assume the + filter was exact. ``None`` parses every line. + + Yields: + Parsed event dicts, oldest first. + + Raises: + OSError: On the returned iterator's first ``next()`` (not at call + time) if ``path`` cannot be opened or read -- see above. """ - try: - with open(path, "rb") as f: - raw = f.read(max_bytes) - except OSError: - return [] - return _parse_jsonl_bytes(raw) + with open(path, "rb") as f: + for raw_line in f: + line = raw_line.strip() + if not line: + continue + if keep_types is not None: + match = _TYPE_PREFIX_RE.match(line) + if match is not None and match.group(1).decode("ascii") not in keep_types: + continue + try: + obj = json.loads(line.decode("utf-8")) + except (UnicodeDecodeError, ValueError, RecursionError): + continue + if isinstance(obj, dict): + yield obj # --------------------------------------------------------------------------- @@ -481,6 +463,37 @@ def read_event_log_full( } ) +# Every event type `_scan_events` actually branches on -- the `keep_types` +# prefilter passed to `stream_event_log` for the Runs screen's per-row scan +# (issue #485). Derived from the two frozensets above (not restated) so +# adding a new close/failure event type to either one keeps this prefilter +# correct automatically, without a second edit anyone could forget. This +# set and `_scan_events` move together: a type the scanner branches on but +# this set omits would be silently dropped from every derived summary -- +# the exact failure class issue #485 is about -- so `TestStreamEventLog`'s +# prefilter-equivalence test compares a filtered scan against an unfiltered +# one over a log exercising every branch, to catch that drift here rather +# than in production. +_SUMMARY_EVENT_TYPES: frozenset[str] = ( + _AGENT_CLOSE_EVENT_TYPES + | _AGENT_FAILED_EVENT_TYPES + | frozenset( + { + "workflow_started", + "workflow_completed", + "workflow_failed", + "agent_started", + "parallel_agent_started", + "parallel_started", + "parallel_completed", + "for_each_started", + "for_each_completed", + "gate_presented", + "agent_paused", + } + ) +) + @dataclass class _ScanResult: @@ -521,13 +534,32 @@ def _extract_topology(data: dict[str, Any]) -> RunTopology: return RunTopology(entry_point=data.get("entry_point"), agents=agents) -def _scan_events(events: list[dict[str, Any]]) -> _ScanResult: +def _scan_events(events: Iterable[dict[str, Any]]) -> _ScanResult: """Single pass over event dicts deriving status, gate, current step, and totals. Events are assumed oldest-first (the natural order of a JSONL log and - of :func:`read_event_log_tail`'s output), so later events in the list - override earlier ones for state that only has one current value - (``status``, ``gate``). + of :func:`stream_event_log`'s output), so later events override earlier + ones for state that only has one current value (``status``, ``gate``, + ``topology``, ``workflow_name``, ``cwd``, ``inputs``). + + Accepts any one-shot iterable and makes exactly one forward pass over + it -- no indexing, no ``len()``, no re-iteration (mirrors + ``fleet.history._scan_history_events``'s identical issue-#436 + constraint) -- which is what lets :func:`derive_run_summary` stream + straight from :func:`stream_event_log` instead of materializing a list. + + **Generation-aware (issue #485):** every *root* ``workflow_started`` + marks the start of a new execution generation of this run -- a resumed + run always writes a second one into the same log, whether or not a + dashboard was attached (see the module docstring). Reaching one resets + the per-generation state a fresh run starts with -- ``status`` back to + ``"running"``, any open gate cleared, every open step closed -- and + *overwrites* (not merges) ``topology``/``workflow_name``/``cwd``/ + ``inputs`` with this generation's own, so a resumed run's current step + and topology reflect what is actually running now, not a stale earlier + generation. Token/cost totals are the one exception: they are **not** + reset here, and keep accumulating across every generation in the log -- + a resumed run's lifetime usage, not just its latest attempt's. """ result = _ScanResult() @@ -540,36 +572,43 @@ def _scan_events(events: list[dict[str, Any]]) -> _ScanResult: # originate from a nested sub-workflow engine, not the root run. # A nested agent can share a root agent's name, so scanning these # would corrupt the root agent's status/timing/usage -- skip - # anything outside the root context. + # anything outside the root context. This is also what keeps a + # nested sub-workflow's own `workflow_started` from being mistaken + # for a resume boundary. if data.get("subworkflow_path"): continue ts = evt.get("timestamp") - if etype == "workflow_started" and result.workflow_name is None: + if etype == "workflow_started": # The workflow's *declared* name (`workflow.name`), which is not # the file stem the run record carries: a repo that stores each # workflow as `/workflow.yaml` makes every run show up as # "workflow". History reads this same declared name out of the # log filename, which is why the two screens disagreed. declared = data.get("name") - if isinstance(declared, str) and declared: - result.workflow_name = declared + result.workflow_name = declared if isinstance(declared, str) and declared else None # Where conductor was launched from, and what it was launched # with -- neither is on the run record, and both are what tells # two runs of the same workflow apart. system = data.get("system") - if isinstance(system, dict): - cwd = system.get("cwd") - if isinstance(cwd, str) and cwd: - result.cwd = cwd + cwd = system.get("cwd") if isinstance(system, dict) else None + result.cwd = cwd if isinstance(cwd, str) and cwd else None inputs = data.get("inputs") - if isinstance(inputs, dict): - result.inputs = inputs + result.inputs = inputs if isinstance(inputs, dict) else None - if etype == "workflow_started" and result.topology is None: result.topology = _extract_topology(data) + # A new generation starts clean: nothing from a dead prior + # attempt (its terminal status, an unresolved gate, a step + # that never got its own closing event before the process + # exited) may leak into how this generation reads. Totals are + # deliberately untouched -- see the docstring. + result.status = "running" + result.gate = None + result.open_steps = [] + continue + elif etype in ("agent_started", "parallel_agent_started"): # A plain agent opens via `agent_started`; a parallel-group # member instead opens via its own `parallel_agent_started` @@ -608,13 +647,13 @@ def _scan_events(events: list[dict[str, Any]]) -> _ScanResult: if result.status == "at-gate": result.status = "running" elif etype in ("agent_completed", "parallel_agent_completed"): - tokens = data.get("tokens") - if isinstance(tokens, int | float): + tokens = _finite_float(data.get("tokens")) + if tokens is not None: result.total_tokens += int(tokens) - cost = data.get("cost_usd") - if isinstance(cost, int | float): - result.total_cost_usd = (result.total_cost_usd or 0.0) + float(cost) - elif isinstance(tokens, int | float) and tokens > 0: + cost = _finite_float(data.get("cost_usd")) + if cost is not None: + result.total_cost_usd = (result.total_cost_usd or 0.0) + cost + elif tokens is not None and tokens > 0: result.unpriced_agent_count += 1 elif etype in _AGENT_FAILED_EVENT_TYPES: @@ -680,9 +719,9 @@ def _scan_events(events: list[dict[str, Any]]) -> _ScanResult: def _scan_agent_details( - events: list[dict[str, Any]], topology: RunTopology | None -) -> tuple[list[AgentDetail], str | None]: - """Single pass over **full**-log events building per-agent detail rows (E9-T3). + events: Iterable[dict[str, Any]], +) -> tuple[RunTopology | None, str | None, list[AgentDetail], str | None]: + """Single pass over the full event stream building per-agent detail rows (E9-T3). Unlike :func:`_scan_events` (which only tracks aggregate totals for the Runs list), this tracks each agent's own completion payload @@ -690,21 +729,39 @@ def _scan_agent_details( ``agent_name`` -- the per-agent history the detail screen needs that the list screen's aggregate totals do not carry. + Topology (and the run's declared workflow name) extraction is folded + into this same pass -- required once the log is read as a one-shot + stream (issue #485): :func:`derive_run_detail` used to iterate the + event list twice, once to find the topology and once here, which a + one-shot iterator cannot support. The **latest** root + ``workflow_started`` wins (last generation wins), so a resumed run's + detail screen reflects the current generation's topology, not a stale + earlier one. That ``workflow_started`` branch also resets ``open_steps``, + ``gated`` and ``started_at_by_name`` -- the same resume-boundary reset + :func:`_scan_events` does -- so a generation killed mid-step (its open + steps and any unresolved gate never getting a closing event) cannot + leak into how the next generation reads. Per-agent status/usage + tracking is otherwise unchanged: an agent's cumulative tokens/cost keep + accumulating across every restart they already did (a loop-back or a + resume look the same to this loop -- a fresh ``agent_started`` always + means "running now", regardless of why the process is executing it + again), consistent with :func:`_scan_events`'s Q1 + totals-accumulate-across-generations rule. + Args: - events: Full-log event dicts, oldest first (see - :func:`read_event_log_full`). - topology: The run's topology (agent order/definitions), or ``None`` - if unavailable -- in which case there is nothing to build rows - for. + events: Event dicts, oldest first (see :func:`stream_event_log`). Returns: - A ``(agents, current_step)`` tuple: one :class:`AgentDetail` per - ``topology.agents`` entry (same order), and the name of the - currently open step (agent, parallel group, or for_each group), or - ``None`` if nothing is open. + A ``(topology, workflow_name, agents, current_step)`` tuple: the + run's topology from its latest ``workflow_started`` (``None`` if + the log never had one), that event's declared workflow name + (``None`` if undeclared or no such event), one :class:`AgentDetail` + per ``topology.agents`` entry (same order, empty when ``topology`` + is ``None``), and the name of the currently open step (agent, + parallel group, or for_each group), or ``None`` if nothing is open. """ - if topology is None: - return [], None + topology: RunTopology | None = None + workflow_name: str | None = None open_steps: list[tuple[str, str, float]] = [] # agent_name -> (status, started_at, reported_elapsed) for the latest @@ -737,7 +794,23 @@ def _scan_agent_details( continue ts = evt.get("timestamp") - if etype in ("agent_started", "parallel_agent_started"): + if etype == "workflow_started": + topology = _extract_topology(data) + declared = data.get("name") + workflow_name = declared if isinstance(declared, str) and declared else None + # A new generation starts clean: an open step or unresolved + # gate from a dead prior attempt (which got no closing event + # before the process exited) must not leak into this + # generation's reading -- the same reset `_scan_events` does at + # its own `workflow_started` branch. `closed`, + # `cumulative_tokens` and `cumulative_cost` are deliberately + # left untouched -- see the docstring. + open_steps.clear() + gated.clear() + started_at_by_name.clear() + continue + + elif etype in ("agent_started", "parallel_agent_started"): # A plain agent opens via `agent_started`; a parallel-group # member instead opens via its own `parallel_agent_started` -- # it never gets a plain `agent_started` at all. @@ -767,12 +840,12 @@ def _scan_agent_details( gated.discard(name) elapsed = data.get("elapsed") if etype in ("agent_completed", "parallel_agent_completed"): - tokens = data.get("tokens") - cost = data.get("cost_usd") - if isinstance(tokens, int | float): + tokens = _finite_float(data.get("tokens")) + cost = _finite_float(data.get("cost_usd")) + if tokens is not None: cumulative_tokens[name] = cumulative_tokens.get(name, 0) + int(tokens) - if isinstance(cost, int | float): - cumulative_cost[name] = (cumulative_cost.get(name) or 0.0) + float(cost) + if cost is not None: + cumulative_cost[name] = (cumulative_cost.get(name) or 0.0) + cost closed[name] = ( "completed", started_at_by_name.get(name), @@ -845,6 +918,9 @@ def _scan_agent_details( open_agent_names = {name for (kind, name, _ts) in open_steps if kind == "agent"} current_step = open_steps[-1][1] if open_steps else None + if topology is None: + return None, workflow_name, [], None + agent_details: list[AgentDetail] = [] for ta in topology.agents: cum_tokens = cumulative_tokens.get(ta.name) @@ -898,7 +974,7 @@ def _scan_agent_details( ) ) - return agent_details, current_step + return topology, workflow_name, agent_details, current_step # --------------------------------------------------------------------------- @@ -906,49 +982,37 @@ def _scan_agent_details( # --------------------------------------------------------------------------- -def derive_run_summary( - record: RunRecord, - *, - tail_bytes: int = _DEFAULT_TAIL_BYTES, -) -> RunSummary: - """Derive a :class:`RunSummary` for ``record`` from its event log's tail. +def derive_run_summary(record: RunRecord) -> RunSummary: + """Derive a :class:`RunSummary` for ``record`` from its event log. ``record`` is assumed to already be known-live (e.g. sourced from :func:`conductor.fleet.records.read_run_records`) — this function does not re-check liveness; see the module docstring for why. When - ``record.event_log_path`` is empty or unreadable, this still returns a - usable summary (``status="running"``, no current step, zero totals) - rather than raising, so a run whose log hasn't been created yet (or - whose path is stale) never crashes the caller's poll loop. + ``record.event_log_path`` is empty, unreadable, or any other + ``OSError`` occurs while streaming it, this still returns a usable + summary (``status="running"``, no current step, zero totals) rather + than raising, so a run whose log hasn't been created yet (or whose + path is stale) never crashes the caller's poll loop -- the + :func:`stream_event_log` generator raises on its first ``next()``, not + at construction, so the ``try``/``except`` here wraps the *consumption* + (:func:`_scan_events`'s draining loop), not just the call. Args: record: The run record to summarize. - tail_bytes: Passed through to :func:`read_event_log_tail`. Returns: A :class:`RunSummary` reflecting the run's state as of the last - event visible within the tail window. + event in the log (see :attr:`RunSummary.total_tokens` for how a + resumed run's multiple generations are combined). """ - events: list[dict[str, Any]] = [] - log_path: Path | None = None + scan = _ScanResult() if record.event_log_path: log_path = Path(record.event_log_path) - events = read_event_log_tail(log_path, tail_bytes=tail_bytes) - - scan = _scan_events(events) - # `workflow_started` is the log's first event, so on a run long enough to - # outgrow the tail window it is always outside it -- the topology (and - # with it the preview's step list) disappeared exactly when a run got - # interesting enough to want it. Read lazily: for any log under the tail - # window -- the common case, and every young run -- the tail already - # contained `workflow_started`, and this runs once per row on the Runs - # screen's ~2s poll, on the UI thread. - if log_path is not None and (scan.topology is None or scan.workflow_name is None): - head_scan = _scan_events(read_event_log_head(log_path)) - scan.topology = scan.topology or head_scan.topology - scan.workflow_name = scan.workflow_name or head_scan.workflow_name - scan.cwd = scan.cwd or head_scan.cwd - scan.inputs = scan.inputs if scan.inputs is not None else head_scan.inputs + try: + scan = _scan_events(stream_event_log(log_path, keep_types=_SUMMARY_EVENT_TYPES)) + except OSError: + logger.debug("Could not read event log %s", log_path, exc_info=True) + scan = _ScanResult() if scan.open_steps: current_step_type, current_step, current_step_started_at = scan.open_steps[-1] @@ -1007,39 +1071,63 @@ class StepDetail: """The run's declared name, not the workflow file's stem.""" -def derive_step_detail( - record: RunRecord, agent_name: str, *, max_bytes: int = _DEFAULT_FULL_LOG_MAX_BYTES -) -> StepDetail: - """Extract one step's input/output/activity from a run's event log. +def _scan_step_events( + events: Iterable[dict[str, Any]], agent_name: str +) -> tuple[str, str | None, Any | None, deque[ActivityLine], str | None]: + """Single pass over the full event stream extracting one step's detail. - Reads the **full** (bounded) log rather than the tail, for the same - reason :func:`derive_run_detail` does: a step's prompt is emitted once, - when it started, which on a long run is far outside the tail window. + Returns ``(status, prompt, output, activity, workflow_name)`` only + after the stream drains -- callers must assign the whole tuple in one + statement inside their own ``try``/``except OSError`` so a mid-stream + read failure (propagating out of the ``events`` iterator) can never + leave a caller holding a partial scan: an exception here means the + assignment in :func:`derive_step_detail` never happens at all, and its + pristine defaults are returned instead. - Activity is bounded to the most recent :data:`_STEP_ACTIVITY_LIMIT` - entries -- a long agentic loop emits hundreds of tool calls, and this is - a drill-down, not a log viewer. - """ - events: list[dict[str, Any]] = [] - if record.event_log_path: - events = read_event_log_full(Path(record.event_log_path), max_bytes=max_bytes) + Args: + events: Event dicts, oldest first (see :func:`stream_event_log`). + agent_name: The step whose prompt/output/activity to extract. + Returns: + The five-tuple described above. ``status`` defaults to + ``"pending"``; ``prompt``/``output``/``workflow_name`` default to + ``None``; ``activity`` defaults to an empty, capacity-bounded + deque. + """ prompt: str | None = None output: Any | None = None status = "pending" activity: deque[ActivityLine] = deque(maxlen=_STEP_ACTIVITY_LIMIT) + # The run's declared workflow name, captured from the same pass (last + # generation wins) rather than a second scan over the same log -- see + # `_scan_agent_details`'s identical reasoning. + workflow_name: str | None = None for evt in events: data = evt.get("data") if not isinstance(data, dict) or data.get("subworkflow_path"): continue + etype = evt.get("type") + + if etype == "workflow_started": + declared = data.get("name") + workflow_name = declared if isinstance(declared, str) and declared else None + # A new generation starts clean: an agent that was mid-flight + # when a prior generation died (and never got a closing event) + # cannot still be "running" across a resume boundary -- but a + # genuine `completed`/`failed` status is real history and is + # left alone. + if status == "running": + status = "pending" + continue + if data.get("agent_name") != agent_name: continue - etype = evt.get("type") if etype in ("agent_started", "parallel_agent_started"): status = "running" # A re-run (loop-back) supersedes the previous attempt's result. + prompt = None output = None activity.clear() elif etype == "agent_prompt_rendered": @@ -1067,88 +1155,93 @@ def derive_step_detail( elif etype == "agent_tool_complete": activity.append(ActivityLine("tool_result", str(data.get("tool_name") or "tool"))) + return status, prompt, output, activity, workflow_name + + +def derive_step_detail(record: RunRecord, agent_name: str) -> StepDetail: + """Extract one step's input/output/activity from a run's event log. + + Streams the whole (uncapped) log -- a step's prompt is emitted once, + when it started, which on a long run the old bounded tail window + always missed. When ``record.event_log_path`` is empty, or any + ``OSError`` occurs while streaming it, this still returns a usable + (``status="pending"``, no prompt/output/activity) :class:`StepDetail` + rather than raising, mirroring :func:`derive_run_summary` / + :func:`derive_run_detail`'s never-raise contract. The scan itself lives + in :func:`_scan_step_events` and is assigned here in one statement so a + mid-stream ``OSError`` can never leave this function holding (and + returning) a partial scan as if it were authoritative. + + Activity is bounded to the most recent :data:`_STEP_ACTIVITY_LIMIT` + entries -- a long agentic loop emits hundreds of tool calls, and this is + a drill-down, not a log viewer. + + Args: + record: The run record whose event log to read. + agent_name: The step (agent, parallel-group member, or non-LLM + step type) to extract input/output/activity for. + + Returns: + A :class:`StepDetail` for ``agent_name``. + """ + prompt: str | None = None + output: Any | None = None + status = "pending" + activity: deque[ActivityLine] = deque(maxlen=_STEP_ACTIVITY_LIMIT) + workflow_name: str | None = None + + if record.event_log_path: + log_path = Path(record.event_log_path) + try: + status, prompt, output, activity, workflow_name = _scan_step_events( + stream_event_log(log_path), agent_name + ) + except OSError: + logger.debug("Could not read event log %s", log_path, exc_info=True) + return StepDetail( agent_name=agent_name, status=status, prompt=prompt, output=output, activity=list(activity), - workflow_name=_declared_workflow_name(events) or record.workflow_name, + workflow_name=workflow_name or record.workflow_name, ) -def _declared_workflow_name(events: list[dict[str, Any]]) -> str | None: - """Return the root run's *declared* workflow name, if the log has it. - - The run record carries the workflow file's stem, so a repo that stores - each workflow as ``/workflow.yaml`` labels every run "workflow". - The declared name is in the log, and it is what the Runs and History - screens already show -- the drill-downs read it here so all four agree. - - Args: - events: Parsed event dicts from the run's log. - - Returns: - The declared name, or ``None`` when no root ``workflow_started`` - carries one. - """ - for evt in events: - if evt.get("type") != "workflow_started": - continue - data = evt.get("data") - if not isinstance(data, dict) or data.get("subworkflow_path"): - continue - declared = data.get("name") - if isinstance(declared, str) and declared: - return declared - return None - - -def derive_run_detail( - record: RunRecord, *, max_bytes: int = _DEFAULT_FULL_LOG_MAX_BYTES -) -> RunDetail: - """Derive a :class:`RunDetail` for ``record`` from its event log's **full** - (bounded) contents -- the run-detail screen's data source (E9-T3), - distinct from :func:`derive_run_summary`'s tail-based read the polled - Runs list uses. - - When ``record.event_log_path`` is empty or unreadable, or the log has - no (or no yet-visible) ``workflow_started`` event, this returns a - :class:`RunDetail` with ``topology=None`` and an empty ``agents`` list - rather than raising -- the detail screen renders a placeholder for this - case (E9-T5) instead of crashing or showing an empty table. +def derive_run_detail(record: RunRecord) -> RunDetail: + """Derive a :class:`RunDetail` for ``record`` from its event log's full + (streamed) contents -- the run-detail screen's data source (E9-T3), + distinct from :func:`derive_run_summary`'s aggregate-only scan the + polled Runs list uses. - Args: - record: The run record to derive detail for. - max_bytes: Passed through to :func:`read_event_log_full`. + When ``record.event_log_path`` is empty, unreadable, or the log has no + ``workflow_started`` event, this returns a :class:`RunDetail` with + ``topology=None`` and an empty ``agents`` list rather than raising -- + the detail screen renders a placeholder for this case (E9-T5) instead + of crashing or showing an empty table. Returns: A :class:`RunDetail` with one row per topology agent (in the topology's own order) and the name of the currently open step. """ - events: list[dict[str, Any]] = [] - if record.event_log_path: - events = read_event_log_full(Path(record.event_log_path), max_bytes=max_bytes) - topology: RunTopology | None = None - for evt in events: - if evt.get("type") == "workflow_started": - data = evt.get("data") - if not isinstance(data, dict): - continue - # Skip a nested sub-workflow's own workflow_started (stamped - # with subworkflow_path) -- only the root run's topology - # belongs on this screen. - if data.get("subworkflow_path"): - continue - topology = _extract_topology(data) - break + workflow_name: str | None = None + agents: list[AgentDetail] = [] + current_step: str | None = None - agents, current_step = _scan_agent_details(events, topology) + if record.event_log_path: + log_path = Path(record.event_log_path) + try: + topology, workflow_name, agents, current_step = _scan_agent_details( + stream_event_log(log_path, keep_types=_SUMMARY_EVENT_TYPES) + ) + except OSError: + logger.debug("Could not read event log %s", log_path, exc_info=True) return RunDetail( run_id=record.run_id, - workflow_name=_declared_workflow_name(events) or record.workflow_name, + workflow_name=workflow_name or record.workflow_name, topology=topology, agents=agents, current_step=current_step, diff --git a/src/conductor/fleet/tui/dag.py b/src/conductor/fleet/tui/dag.py index dc92a377..e0e03d13 100644 --- a/src/conductor/fleet/tui/dag.py +++ b/src/conductor/fleet/tui/dag.py @@ -155,10 +155,11 @@ def _append_more(out: Text, remaining: int, line_width: int, width: int) -> None def step_statuses(agent_names: list[str], current_step: str | None) -> dict[str, str]: """Infer per-step statuses from position alone. - The Runs screen's bounded tail read knows the current step but not the - per-step history the run-detail screen derives from a full log read. - Position is a sound stand-in *for a linear run*: everything before the - current step has been passed through, everything after has not. + The Runs screen's streamed, prefiltered scan knows the current step + but not the per-step history the run-detail screen derives from its + own streamed, unfiltered scan of the same log. Position is a sound + stand-in *for a linear run*: everything before the current step has + been passed through, everything after has not. It is deliberately only a stand-in. A workflow that loops back, or one whose current step is unknown, gets no invented history -- callers with diff --git a/src/conductor/fleet/tui/screens/run_detail.py b/src/conductor/fleet/tui/screens/run_detail.py index 76bd210e..a00c182b 100644 --- a/src/conductor/fleet/tui/screens/run_detail.py +++ b/src/conductor/fleet/tui/screens/run_detail.py @@ -9,15 +9,16 @@ no DAG rendering, no agent messages, no tool output — only status, elapsed, tokens, and cost per agent, with the current step highlighted (E9-T2)). -Unlike the Runs screen (which stays on the bounded tail-window read so its -~2s poll never grows with a run's age), this screen uses the bounded -**full**-log read (:func:`conductor.fleet.summary.derive_run_detail`, -E9-T3) so every agent's complete history is available, not just whichever -agents happen to still be inside the tail window. +Unlike the Runs screen (which prefilters its streamed scan to a small set +of event types so its ~2s poll stays cheap regardless of a run's age), +this screen uses the unfiltered streamed read +(:func:`conductor.fleet.summary.derive_run_detail`, E9-T3) so every +agent's complete history is available, not just whichever event types the +Runs screen's aggregate totals need. Both derivations run in a single worker thread (:func:`asyncio.to_thread`), -not on the event loop, so the ~2s poll's larger read (a full-log scan on -top of the Runs screen's bounded tail read) never blocks keypresses or the +not on the event loop, so the ~2s poll's larger read (an unfiltered scan on +top of the Runs screen's prefiltered one) never blocks keypresses or the footer repaint (issue #437); a tick that overruns the poll interval is dropped rather than queued. """ diff --git a/src/conductor/fleet/tui/screens/runs.py b/src/conductor/fleet/tui/screens/runs.py index 197589aa..ef292ff1 100644 --- a/src/conductor/fleet/tui/screens/runs.py +++ b/src/conductor/fleet/tui/screens/runs.py @@ -8,7 +8,8 @@ Refreshed on a ~2s poll timer (:data:`RunsScreen.POLL_INTERVAL_SECONDS`) via Textual's ``set_interval`` — a full rescan of the run-record directory plus -a bounded event-log tail seek per live run (:mod:`conductor.fleet.summary`). +a streamed, prefiltered event-log scan per live run +(:mod:`conductor.fleet.summary`). Per the design's *Refresh model*, there is deliberately no file watcher. That scan runs in a worker thread (:func:`asyncio.to_thread`), not on the diff --git a/src/conductor/fleet/tui/screens/step_detail.py b/src/conductor/fleet/tui/screens/step_detail.py index e414cce2..15e2234c 100644 --- a/src/conductor/fleet/tui/screens/step_detail.py +++ b/src/conductor/fleet/tui/screens/step_detail.py @@ -222,10 +222,10 @@ def action_reload(self) -> None: async def load_step(self) -> None: """Read this step's detail from the log, off the event loop. - ``derive_step_detail`` does a bounded **full**-log read (a step's - prompt is emitted once, at its start, so the tail window misses it on - any long run) -- large enough to be worth a thread rather than - blocking the UI while it parses. + ``derive_step_detail`` streams the whole (uncapped) log -- a + step's prompt is emitted once, at its start, which a bounded tail + window would miss on any long run -- large enough to be worth a + thread rather than blocking the UI while it parses. """ status = self.query_one("#step-status", Static) input_content = self.query_one("#input-content", Static) diff --git a/tests/test_fleet/test_history.py b/tests/test_fleet/test_history.py index 31be1a9e..b1a1bca2 100644 --- a/tests/test_fleet/test_history.py +++ b/tests/test_fleet/test_history.py @@ -30,6 +30,7 @@ import pytest import conductor.fleet.history as history_module +import conductor.fleet.summary as summary_module from conductor.fleet.history import ( HistoryEntry, _CorruptEventLogError, @@ -673,6 +674,87 @@ def test_a_resumed_attempts_own_terminal_event_is_used(self, temp_root: Path) -> assert entries[0].outcome == "completed" assert entries[0].duration_seconds == 55.0 + def test_started_at_is_the_latest_generations_not_the_first(self, temp_root: Path) -> None: + """Issue #485, Q2: `started_at` is not itself a displayed column, + but it feeds `duration_seconds`'s fallback, so a resumed run's + entry must not span the idle gap between the original attempt + and its resume.""" + _write_log( + temp_root, + lines=[ + _event("workflow_started", {"name": "my-workflow"}, ts=1000.0), + _event("workflow_failed", {"error_type": "ValueError"}, ts=1010.0), + _event("workflow_started", {"name": "my-workflow"}, ts=2000.0), + _event("workflow_completed", {"elapsed": 55.0}, ts=2055.0), + ], + ) + + entries = build_history_entries() + + assert len(entries) == 1 + assert entries[0].started_at == 2000.0 + + def test_resumed_then_failed_duration_measures_the_latest_generation_only( + self, temp_root: Path + ) -> None: + """A resumed-then-failed run has no engine-reported `elapsed` + (`workflow_failed` carries none), so `duration_seconds` falls back + to `ended_at - started_at` -- which must not span the idle gap + between the original attempt (far in the past) and the resume + that actually failed (issue #485, Q2).""" + _write_log( + temp_root, + lines=[ + _event("workflow_started", {"name": "my-workflow"}, ts=1000.0), + _event("workflow_failed", {"error_type": "ValueError"}, ts=1010.0), + # A long idle gap before the resume -- if `started_at` + # were still the first generation's, duration_seconds + # would (wrongly) include this gap. + _event("workflow_started", {"name": "my-workflow"}, ts=100_000.0), + _event("workflow_failed", {"error_type": "RuntimeError"}, ts=100_005.0), + ], + ) + + entries = build_history_entries() + + assert len(entries) == 1 + assert entries[0].outcome == "failed" + assert entries[0].started_at == 100_000.0 + assert entries[0].ended_at == 100_005.0 + assert entries[0].duration_seconds == pytest.approx(5.0) + + def test_a_nested_subworkflow_start_is_not_mistaken_for_a_resume(self, temp_root: Path) -> None: + """A nested sub-workflow's own `workflow_started` (`subworkflow_path` + stamped) must not be treated as a root resume boundary -- it would + otherwise wrongly reset `started_at` (Q2) and, before this guard + moved `started_at`'s assignment out of the `else` branch, could + overwrite it where first-wins previously made that impossible.""" + _write_log( + temp_root, + lines=[ + _event("workflow_started", {"name": "my-workflow"}, ts=1000.0), + _event("agent_started", {"agent_name": "a"}, ts=1001.0), + _event( + "workflow_started", + {"name": "sub", "subworkflow_path": ["sub"]}, + ts=1500.0, + ), + _event( + "workflow_completed", + {"subworkflow_path": ["sub"]}, + ts=1800.0, + ), + _event("workflow_failed", {"error_type": "ValueError"}, ts=2000.0), + ], + ) + + entries = build_history_entries() + + assert len(entries) == 1 + assert entries[0].outcome == "failed" + assert entries[0].started_at == 1000.0 + assert entries[0].ended_at == 2000.0 + # --------------------------------------------------------------------------- # HistoryEntry shape @@ -718,7 +800,10 @@ def test_read_full_log_does_not_return_a_list(self, temp_root: Path) -> None: def test_read_full_log_parses_lazily(self, temp_root: Path) -> None: """Direct, deterministic proof that memory no longer scales with log length: taking a single `next()` off the generator parses - exactly one line, not the whole file.""" + exactly one line, not the whole file. `_read_full_log` now + delegates the actual parsing to + `conductor.fleet.summary.stream_event_log` (issue #485), so the + call is observed there rather than on `history_module.json`.""" path = _write_log( temp_root, lines=[ @@ -729,7 +814,7 @@ def test_read_full_log_parses_lazily(self, temp_root: Path) -> None: ) with ( - patch.object(history_module.json, "loads", wraps=json.loads) as loads, + patch.object(summary_module.json, "loads", wraps=json.loads) as loads, contextlib.closing(_read_full_log(path)) as gen, ): assert inspect.isgenerator(gen) diff --git a/tests/test_fleet/test_summary.py b/tests/test_fleet/test_summary.py index eb2a3f8e..832d793b 100644 --- a/tests/test_fleet/test_summary.py +++ b/tests/test_fleet/test_summary.py @@ -2,9 +2,10 @@ RunDetail derivation). Covers: -- Bounded JSONL tail reading: whole small file, seek-and-discard-partial-line - for a file larger than the tail window, tolerance of a mid-line truncation, - an empty file, and a missing file. +- Streaming JSONL reading (issue #485): whole small file, tolerance of a + mid-line truncation, an empty file, a missing file, and the + ``keep_types`` prefilter (including its "not first" fallback and its + equivalence with an unfiltered scan). - Every status in the design's vocabulary (`running`, `at-gate`, `paused`, `completed`, `failed`), including a gate opened then resolved returning to `running`. @@ -16,10 +17,13 @@ - Topology extraction from `workflow_started`. - The gate payload carried onto the summary, and `gate_resolvable` per D4 (true for `fg-web`/`bg`, false for `fg`). -- (E9-T3) The bounded *full*-log reader and per-agent `RunDetail` derivation - used only by the run-detail screen: per-agent pending/running/completed/ - failed status, elapsed/tokens/cost per agent, and graceful degradation - when the log or its `workflow_started` event is missing. +- A resumed run's generation-aware reset: status/gate/current-step reset at + each root `workflow_started`, while token/cost totals accumulate across + every generation (issue #485, Q1). +- Per-agent `RunDetail` derivation used only by the run-detail screen: + per-agent pending/running/completed/failed status, elapsed/tokens/cost per + agent, and graceful degradation when the log or its `workflow_started` + event is missing. """ from __future__ import annotations @@ -30,14 +34,20 @@ from pathlib import Path from typing import Any +import pytest + +from conductor.fleet import summary as summary_module from conductor.fleet.records import RunRecord from conductor.fleet.summary import ( + _SUMMARY_EVENT_TYPES, RunDetail, RunSummary, + _scan_agent_details, + _scan_events, derive_run_detail, derive_run_summary, - read_event_log_full, - read_event_log_tail, + derive_step_detail, + stream_event_log, ) # --------------------------------------------------------------------------- @@ -77,8 +87,8 @@ def _make_record(tmp_path: Path, **overrides: object) -> RunRecord: # --------------------------------------------------------------------------- -class TestReadEventLogTail: - def test_reads_all_events_when_file_smaller_than_window(self, tmp_path: Path) -> None: +class TestStreamEventLog: + def test_reads_all_events_oldest_first(self, tmp_path: Path) -> None: path = tmp_path / "small.events.jsonl" _write_jsonl( path, @@ -88,45 +98,27 @@ def test_reads_all_events_when_file_smaller_than_window(self, tmp_path: Path) -> ], ) - events = read_event_log_tail(path, tail_bytes=1024) + events = list(stream_event_log(path)) assert len(events) == 2 assert events[0]["type"] == "agent_started" assert events[1]["type"] == "agent_completed" - def test_bounded_read_discards_older_events(self, tmp_path: Path) -> None: - """A file much larger than the tail window only yields the newest events -- - confirming the read is bounded, not whole-file.""" + def test_reads_a_log_far_larger_than_the_old_bounded_windows(self, tmp_path: Path) -> None: + """No cap at all (issue #485): every event in a log much larger + than the old 512 KiB tail/head windows or the 8 MiB full-log cap + is still read, including the very first line.""" path = tmp_path / "large.events.jsonl" - lines = [_event("agent_started", {"agent_name": f"agent-{i}"}) for i in range(500)] + lines = [_event("workflow_started", {"name": "wf"})] + lines += [_event("agent_started", {"agent_name": f"agent-{i}"}) for i in range(20000)] _write_jsonl(path, lines) + assert path.stat().st_size > 512 * 1024 - events = read_event_log_tail(path, tail_bytes=200) - - assert 0 < len(events) < 500 - # Only the newest (highest-index) agents survive in the tail window. - names = [e["data"]["agent_name"] for e in events] - assert names[-1] == "agent-499" - assert "agent-0" not in names - - def test_discards_partial_leading_line_at_seek_boundary(self, tmp_path: Path) -> None: - """The line straddling the seek point must never surface as a garbled event.""" - path = tmp_path / "boundary.events.jsonl" - lines = [_event("agent_started", {"agent_name": f"agent-{i}"}) for i in range(50)] - _write_jsonl(path, lines) + events = list(stream_event_log(path)) - # Pick a range of tail_bytes that lands mid-line at various boundaries. - saw_at_least_one_event = False - for tail_bytes in range(10, 200, 7): - events = read_event_log_tail(path, tail_bytes=tail_bytes) - if events: - saw_at_least_one_event = True - for e in events: - assert e["type"] == "agent_started" - assert e["data"]["agent_name"].startswith("agent-") - # Confirms the loop above actually exercised parsed events, not just - # a vacuously-true "no garbled events" over consistently-empty results. - assert saw_at_least_one_event + assert len(events) == 20001 + assert events[0]["type"] == "workflow_started" + assert events[-1]["data"]["agent_name"] == "agent-19999" def test_tolerates_truncated_mid_line(self, tmp_path: Path) -> None: path = tmp_path / "truncated.events.jsonl" @@ -135,7 +127,7 @@ def test_tolerates_truncated_mid_line(self, tmp_path: Path) -> None: good2 = _event("workflow_completed", {}) path.write_text(f"{good1}\n{bad}\n{good2}\n") - events = read_event_log_tail(path, tail_bytes=4096) + events = list(stream_event_log(path)) types = [e["type"] for e in events] assert types == ["agent_started", "workflow_completed"] @@ -144,16 +136,174 @@ def test_empty_file_yields_no_events(self, tmp_path: Path) -> None: path = tmp_path / "empty.events.jsonl" path.write_text("") - events = read_event_log_tail(path) - - assert events == [] + assert list(stream_event_log(path)) == [] - def test_missing_file_yields_no_events(self, tmp_path: Path) -> None: + def test_missing_file_raises_oserror_on_first_next(self, tmp_path: Path) -> None: + """Unlike the bounded readers this replaces, a read failure is not + swallowed -- it propagates on the generator's first `next()`, not + at construction (issue #485). `derive_run_summary` is the caller + that must wrap the *consumption*, not just this call, to keep its + own never-raise contract -- see `TestStatusVocabulary`'s + `test_missing_event_log_path_never_raises` and the sibling test + for a genuinely missing (but declared) path below.""" path = tmp_path / "does-not-exist.events.jsonl" - events = read_event_log_tail(path) + gen = stream_event_log(path) - assert events == [] + with pytest.raises(OSError): + next(gen) + + def test_missing_log_path_still_yields_a_usable_summary(self, tmp_path: Path) -> None: + """`stream_event_log` raises for a missing file, but + `derive_run_summary` must not -- it wraps the consumption.""" + record = _make_record( + tmp_path, event_log_path=str(tmp_path / "does-not-exist.events.jsonl") + ) + + summary = derive_run_summary(record) + + assert summary.status == "running" + assert summary.current_step is None + assert summary.total_tokens == 0 + + def test_keep_types_skips_uninteresting_lines_without_parsing_them( + self, tmp_path: Path + ) -> None: + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event("agent_started", {"agent_name": "a"}), + _event("agent_message", {"agent_name": "a", "content": "thinking..."}), + _event("agent_completed", {"agent_name": "a", "tokens": 5}), + ], + ) + + events = list(stream_event_log(path, keep_types=frozenset({"agent_started"}))) + + assert [e["type"] for e in events] == ["agent_started"] + + def test_keep_types_none_parses_every_line(self, tmp_path: Path) -> None: + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event("agent_started", {"agent_name": "a"}), + _event("agent_message", {"agent_name": "a", "content": "thinking..."}), + ], + ) + + events = list(stream_event_log(path, keep_types=None)) + + assert [e["type"] for e in events] == ["agent_started", "agent_message"] + + def test_a_type_key_not_first_is_parsed_regardless_of_keep_types(self, tmp_path: Path) -> None: + """The prefilter is only ever an optimization: a line whose `type` + key isn't first fails the anchored regex, so it falls through to + being parsed (and yielded) unconditionally, regardless of + `keep_types` -- it can never silently drop an event just because + the writer happened to serialize it with a different key order. + The accepted trade-off is the mirror image: such a line is *not* + re-checked against `keep_types` after parsing either, so it can + also survive a filter that would otherwise have excluded it.""" + path = tmp_path / "run.events.jsonl" + reordered = json.dumps( + {"timestamp": 1.0, "type": "agent_started", "data": {"agent_name": "a"}} + ) + path.write_text(reordered + "\n") + + kept = list(stream_event_log(path, keep_types=frozenset({"agent_started"}))) + not_excluded = list(stream_event_log(path, keep_types=frozenset({"workflow_completed"}))) + + assert [e["type"] for e in kept] == ["agent_started"] + assert [e["type"] for e in not_excluded] == ["agent_started"] + + def test_prefilter_matches_an_unfiltered_scan(self, tmp_path: Path) -> None: + """The strongest guard against prefilter drift (issue #485): build + a log exercising every branch `_scan_events` handles, then assert + a `keep_types=_SUMMARY_EVENT_TYPES` scan produces the identical + `_ScanResult` as an unfiltered one. A future event type handled by + the scanner but missing from `_SUMMARY_EVENT_TYPES` fails here + instead of silently vanishing from the Runs screen. A + self-policing assertion pins the fixture itself to + `_SUMMARY_EVENT_TYPES` so forgetting to add a fixture line for a + newly-added type (the same act of forgetting that would defeat + this test) fails loudly instead of the two scans trivially + agreeing on a type neither ever saw.""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event( + "workflow_started", + { + "name": "wf", + "entry_point": "a", + "agents": [{"name": "a", "type": "agent"}], + "system": {"cwd": "/tmp/proj"}, + "inputs": {"question": "hi"}, + }, + ), + _event("agent_started", {"agent_name": "a"}, ts=1.0), + _event( + "gate_presented", + {"agent_name": "a", "prompt": "OK?", "options": ["yes"]}, + ts=2.0, + ), + _event("gate_resolved", {"agent_name": "a"}, ts=3.0), + _event( + "agent_completed", + {"agent_name": "a", "tokens": 10, "cost_usd": 0.01}, + ts=4.0, + ), + _event("agent_started", {"agent_name": "b"}, ts=5.0), + _event("agent_paused", {"agent_name": "b"}, ts=6.0), + _event("agent_started", {"agent_name": "b"}, ts=7.0), + _event("parallel_started", {"group_name": "fanout"}, ts=8.0), + _event( + "parallel_agent_started", + {"group_name": "fanout", "agent_name": "c"}, + ts=9.0, + ), + _event( + "parallel_agent_completed", + {"group_name": "fanout", "agent_name": "c", "tokens": 3}, + ts=10.0, + ), + _event("parallel_agent_failed", {"group_name": "fanout", "agent_name": "d"}), + _event("parallel_completed", {"group_name": "fanout"}), + _event("for_each_started", {"group_name": "triage"}, ts=11.0), + _event("for_each_completed", {"group_name": "triage"}), + _event("script_completed", {"agent_name": "e"}), + _event("wait_completed", {"agent_name": "f"}), + _event("set_completed", {"agent_name": "g"}), + _event("subworkflow_completed", {"agent_name": "h"}), + _event("questions_completed", {"agent_name": "i"}), + _event("script_failed", {"agent_name": "j"}), + _event("wait_failed", {"agent_name": "k"}), + _event("set_failed", {"agent_name": "l"}), + _event("subworkflow_failed", {"agent_name": "m"}), + _event("agent_failed", {"agent_name": "n"}), + _event( + "workflow_started", + {"name": "wf", "agents": [{"name": "z", "type": "agent"}]}, + ts=12.0, + ), + _event("agent_started", {"agent_name": "z"}, ts=13.0), + _event("workflow_failed", {"agent_name": "z"}, ts=14.0), + _event("workflow_completed", {}), + ], + ) + + fixture_types = { + json.loads(line)["type"] for line in path.read_text().splitlines() if line.strip() + } + assert fixture_types >= _SUMMARY_EVENT_TYPES + + filtered = _scan_events(stream_event_log(path, keep_types=_SUMMARY_EVENT_TYPES)) + unfiltered = _scan_events(stream_event_log(path)) + + assert filtered == unfiltered # --------------------------------------------------------------------------- @@ -580,6 +730,33 @@ def test_no_completed_agents_yields_zero_tokens_none_cost(self, tmp_path: Path) assert summary.total_cost_usd is None assert summary.has_unpriced is False + def test_nan_or_infinite_tokens_and_cost_are_ignored_not_summed(self, tmp_path: Path) -> None: + """`NaN`/`Infinity` are valid JSON (Python's `json` module accepts + them by default) but are not legitimate token counts or costs -- + this must not crash the poll loop or silently poison the running + total (`int(nan)` raises `ValueError`, `int(inf)` raises + `OverflowError`).""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event( + "agent_completed", + {"agent_name": "good", "tokens": 100, "cost_usd": 0.01}, + ), + _event( + "agent_completed", + {"agent_name": "bad", "tokens": float("nan"), "cost_usd": float("inf")}, + ), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) + + summary = derive_run_summary(record) + + assert summary.total_tokens == 100 + assert summary.total_cost_usd == pytest.approx(0.01) + # --------------------------------------------------------------------------- # Topology extraction (E6-T5) @@ -749,8 +926,8 @@ def _workflow_started_event(agent_names: list[str], *, ts: float | None = None) ) -class TestReadEventLogFull: - def test_reads_all_events_when_file_smaller_than_bound(self, tmp_path: Path) -> None: +class TestDeriveRunDetailNoCap: + def test_reads_all_events_regardless_of_size(self, tmp_path: Path) -> None: path = tmp_path / "small.events.jsonl" _write_jsonl( path, @@ -760,51 +937,101 @@ def test_reads_all_events_when_file_smaller_than_bound(self, tmp_path: Path) -> _event("agent_completed", {"agent_name": "a", "tokens": 5, "cost_usd": 0.01}), ], ) + record = _make_record(tmp_path, event_log_path=str(path)) - events = read_event_log_full(path) + detail = derive_run_detail(record) - assert len(events) == 3 - assert events[0]["type"] == "workflow_started" - assert events[-1]["type"] == "agent_completed" + assert detail.topology is not None + assert detail.agents[0].tokens == 5 + assert detail.agents[0].cost_usd == 0.01 - def test_bounded_read_truncates_a_pathologically_large_log(self, tmp_path: Path) -> None: - """A log far larger than max_bytes is truncated (from the start - forward, unlike the tail reader) rather than loaded in full.""" + def test_trailing_events_survive_a_log_larger_than_the_old_8mib_cap( + self, tmp_path: Path + ) -> None: + """A log far larger than the old 8 MiB full-log cap (issue #485) + must still surface its trailing state -- including, in the worst + case, the run's own terminal event -- rather than being silently + truncated from the start forward.""" path = tmp_path / "huge.events.jsonl" lines = [_workflow_started_event(["a"])] - lines += [_event("agent_started", {"agent_name": f"agent-{i}"}) for i in range(5000)] + # Padding well past the old 8 MiB bound. + lines += [ + _event("agent_message", {"agent_name": "a", "content": "x" * 900}) for _ in range(9500) + ] + lines += [ + _event("agent_started", {"agent_name": "a"}, ts=1000.0), + _event( + "agent_completed", + {"agent_name": "a", "elapsed": 5.0, "tokens": 42, "cost_usd": 0.02}, + ts=1005.0, + ), + ] _write_jsonl(path, lines) - full_size = path.stat().st_size + assert path.stat().st_size > 8 * 1024 * 1024 + record = _make_record(tmp_path, event_log_path=str(path)) - events = read_event_log_full(path, max_bytes=200) + detail = derive_run_detail(record) - assert 0 < len(events) < len(lines) - # First event (workflow_started) still present -- confirms this - # reads from the start, the opposite end from read_event_log_tail. - assert events[0]["type"] == "workflow_started" - assert full_size > 200 + agent = detail.agents[0] + assert agent.status == "completed" + assert agent.tokens == 42 + assert agent.cost_usd == 0.02 - def test_empty_file_yields_no_events(self, tmp_path: Path) -> None: + def test_empty_file_yields_empty_detail(self, tmp_path: Path) -> None: path = tmp_path / "empty.events.jsonl" path.write_text("") + record = _make_record(tmp_path, event_log_path=str(path)) - assert read_event_log_full(path) == [] + detail = derive_run_detail(record) - def test_missing_file_yields_no_events(self, tmp_path: Path) -> None: - path = tmp_path / "does-not-exist.events.jsonl" + assert detail.topology is None + assert detail.agents == [] - assert read_event_log_full(path) == [] + def test_missing_file_yields_empty_detail_never_raises(self, tmp_path: Path) -> None: + record = _make_record( + tmp_path, event_log_path=str(tmp_path / "does-not-exist.events.jsonl") + ) + + detail = derive_run_detail(record) + + assert detail.topology is None + assert detail.agents == [] def test_tolerates_truncated_mid_line(self, tmp_path: Path) -> None: path = tmp_path / "truncated.events.jsonl" good = _workflow_started_event(["a"]) bad = '{"type": "agent_started", "data": {"agent_nam' path.write_text(f"{good}\n{bad}\n") + record = _make_record(tmp_path, event_log_path=str(path)) + + detail = derive_run_detail(record) - events = read_event_log_full(path) + assert detail.topology is not None + assert [a.name for a in detail.agents] == ["a"] + assert detail.agents[0].status == "pending" + + def test_topology_comes_from_the_last_workflow_started(self, tmp_path: Path) -> None: + """A resumed run's detail screen must reflect the current + generation's topology, not a stale earlier one (issue #485) -- + `_scan_agent_details` can no longer make two passes to find the + *first* `workflow_started` once the log is a one-shot stream, and + "last generation wins" is also the correct answer.""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _workflow_started_event(["first-gen-agent"]), + _event("agent_started", {"agent_name": "first-gen-agent"}), + _event("workflow_failed", {"error_type": "ValueError"}), + _workflow_started_event(["second-gen-agent"]), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) - assert len(events) == 1 - assert events[0]["type"] == "workflow_started" + detail = derive_run_detail(record) + + assert detail.topology is not None + assert [a.name for a in detail.agents] == ["second-gen-agent"] # --------------------------------------------------------------------------- @@ -1329,6 +1556,59 @@ def test_restarted_agent_reflects_latest_attempt(self, tmp_path: Path) -> None: assert reviewer.elapsed_seconds(now=1010.0 + 4.0) == 4.0 +class TestScanAgentDetailsPrefilterEquivalence: + def test_scan_agent_details_prefilter_matches_an_unfiltered_scan(self, tmp_path: Path) -> None: + """Mirrors `TestStreamEventLog.test_prefilter_matches_an_unfiltered_scan` + for `_scan_agent_details`, which now also reads with + `keep_types=_SUMMARY_EVENT_TYPES` (issue #485 review): every branch + `_scan_agent_details` handles is a strict subset of + `_SUMMARY_EVENT_TYPES`, so a filtered and an unfiltered scan must + agree exactly.""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event( + "workflow_started", + {"name": "wf", "agents": [{"name": "a"}, {"name": "b"}, {"name": "c"}]}, + ts=0.0, + ), + _event("agent_started", {"agent_name": "a"}, ts=1.0), + _event("gate_presented", {"agent_name": "a", "prompt": "OK?"}, ts=2.0), + _event("gate_resolved", {"agent_name": "a"}, ts=3.0), + _event( + "agent_completed", + {"agent_name": "a", "tokens": 10, "cost_usd": 0.01, "elapsed": 5.0}, + ts=4.0, + ), + _event("agent_started", {"agent_name": "b"}, ts=5.0), + _event("agent_failed", {"agent_name": "b", "elapsed": 1.0}, ts=6.0), + _event("parallel_started", {"group_name": "fanout"}, ts=7.0), + _event( + "parallel_agent_started", + {"group_name": "fanout", "agent_name": "c"}, + ts=8.0, + ), + _event( + "parallel_agent_completed", + {"group_name": "fanout", "agent_name": "c", "tokens": 3, "elapsed": 1.0}, + ts=9.0, + ), + _event("parallel_completed", {"group_name": "fanout"}), + _event("for_each_started", {"group_name": "triage"}, ts=10.0), + _event("for_each_completed", {"group_name": "triage"}), + _event("script_completed", {"agent_name": "d"}), + _event("script_failed", {"agent_name": "e"}), + _event("workflow_failed", {"agent_name": "a"}), + ], + ) + + filtered = _scan_agent_details(stream_event_log(path, keep_types=_SUMMARY_EVENT_TYPES)) + unfiltered = _scan_agent_details(stream_event_log(path)) + + assert filtered == unfiltered + + class TestDeriveRunDetailGracefulDegradation: def test_missing_event_log_path_yields_empty_detail(self, tmp_path: Path) -> None: record = _make_record(tmp_path, event_log_path="") @@ -1539,24 +1819,363 @@ def test_falls_back_to_the_record_when_undeclared(self, tmp_path: Path) -> None: class TestTopologySurvivesALongLog: - """`workflow_started` is the log's first event, so on any run long enough - to outgrow the tail window it is the one event guaranteed to be outside - it -- the step list disappeared exactly when a run got interesting.""" + """`workflow_started` is the log's first event; before issue #485, on + any run long enough to outgrow the (now-removed) tail window it was + the one event guaranteed to be outside it, so the step list + disappeared exactly when a run got interesting enough to want it. + There is no window to outgrow now -- `stream_event_log` is uncapped.""" - def test_topology_read_from_the_head_when_the_tail_misses_it(self, tmp_path: Path) -> None: + def test_topology_survives_a_log_far_larger_than_the_old_tail_window( + self, tmp_path: Path + ) -> None: path = tmp_path / "run.events.jsonl" lines = [ _event( "workflow_started", - {"workflow_name": "wf", "agents": [{"name": "first", "type": "agent"}]}, + {"name": "wf", "agents": [{"name": "first", "type": "agent"}]}, ) ] - # Push it well outside a deliberately tiny tail window. + # Push the log well past the old (now-removed) 512 KiB tail window. lines += [ - _event("agent_message", {"agent_name": "a", "text": "x" * 200}) for _ in range(80) + _event("agent_message", {"agent_name": "a", "content": "x" * 2000}) for _ in range(500) ] _write_jsonl(path, lines) + assert path.stat().st_size > 512 * 1024 + + summary = derive_run_summary(_make_record(tmp_path)) - summary = derive_run_summary(_make_record(tmp_path), tail_bytes=2048) assert summary.topology is not None assert [a.name for a in summary.topology.agents] == ["first"] + + +# --------------------------------------------------------------------------- +# Issue #485 regression: current step / tokens / cost survive a log whose +# most recent agent_started sits far beyond the old 512 KiB tail window. +# --------------------------------------------------------------------------- + + +class TestIssue485CurrentStepTokensCostSurviveALongLog: + def test_current_step_tokens_cost_and_topology_all_populated(self, tmp_path: Path) -> None: + """The screenshot, reduced: a run whose most recent `agent_started` + is padded well beyond the old bounded tail window used to report + `current_step=None`, `total_tokens=0` -- and no topology, since the + run was long enough for the old head-recovery path to matter too. + None of that is bounded any more.""" + path = tmp_path / "run.events.jsonl" + lines = [ + _event( + "workflow_started", + {"name": "implement", "agents": [{"name": "epic_reviewer", "type": "agent"}]}, + ), + _event( + "agent_completed", + {"agent_name": "earlier", "tokens": 537_391_756, "cost_usd": 175.14}, + ), + ] + # Pad the log with realistic in-between activity -- tool calls and + # message chunks -- until it is comfortably past the old 512 KiB + # tail window, mirroring the real log that produced the issue. + padding = [ + _event("agent_tool_start", {"agent_name": "epic_reviewer", "tool_name": "read"}), + _event("agent_message", {"agent_name": "epic_reviewer", "content": "working..." * 50}), + ] + while sum(len(line) for line in lines) < 600 * 1024: + lines.extend(padding) + lines.append(_event("agent_started", {"agent_name": "epic_reviewer"})) + _write_jsonl(path, lines) + assert path.stat().st_size > 512 * 1024 + record = _make_record(tmp_path, event_log_path=str(path)) + + summary = derive_run_summary(record) + + assert summary.current_step == "epic_reviewer" + assert summary.total_tokens == 537_391_756 + assert summary.total_cost_usd == pytest.approx(175.14) + assert summary.topology is not None + assert [a.name for a in summary.topology.agents] == ["epic_reviewer"] + assert summary.status == "running" + + +# --------------------------------------------------------------------------- +# Resumed runs: generation-aware reset (issue #485, Q1/Q2) +# --------------------------------------------------------------------------- + + +class TestResumedRunGenerations: + def test_status_resets_but_totals_accumulate_across_generations(self, tmp_path: Path) -> None: + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event( + "workflow_started", + { + "name": "wf", + "agents": [{"name": "a", "type": "agent"}], + "system": {"cwd": "/tmp/first"}, + "inputs": {"question": "first"}, + }, + ts=1000.0, + ), + _event("agent_started", {"agent_name": "a"}, ts=1001.0), + _event( + "agent_completed", + {"agent_name": "a", "tokens": 100, "cost_usd": 0.01}, + ts=1002.0, + ), + _event("workflow_failed", {"error_type": "ValueError"}, ts=1003.0), + _event( + "workflow_started", + { + "name": "wf", + "agents": [{"name": "b", "type": "agent"}], + "system": {"cwd": "/tmp/second"}, + "inputs": {"question": "second"}, + }, + ts=2000.0, + ), + _event("agent_started", {"agent_name": "b"}, ts=2001.0), + _event( + "agent_completed", + {"agent_name": "c", "tokens": 50, "cost_usd": 0.02}, + ts=2002.0, + ), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) + + summary = derive_run_summary(record) + + # Status/current-step reflect the *current* (resumed) generation -- + # not the dead one's "failed" outcome (issue #485). + assert summary.status == "running" + assert summary.current_step == "b" + # Totals are a lifetime sum across every generation (Q1) -- the + # first generation's usage is not lost on resume, AND the second + # generation's own usage is added on top of it (a bug that + # preserved the first generation's total but stopped accumulating + # thereafter would satisfy a `total_tokens == 100`-only assertion). + assert summary.total_tokens == 150 + assert summary.total_cost_usd == pytest.approx(0.03) + assert summary.unpriced_agent_count == 0 + # Topology/cwd/inputs come from the *second* workflow_started + # (differing agent lists prove which one won). + assert summary.topology is not None + assert [a.name for a in summary.topology.agents] == ["b"] + assert summary.cwd == "/tmp/second" + assert summary.inputs == {"question": "second"} + + def test_an_unresolved_gate_from_a_dead_generation_does_not_survive_a_resume( + self, tmp_path: Path + ) -> None: + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event("workflow_started", {"name": "wf"}, ts=1000.0), + _event("agent_started", {"agent_name": "ask"}, ts=1001.0), + _event("gate_presented", {"agent_name": "ask", "prompt": "OK?"}, ts=1002.0), + # The process died with the gate still open -- no + # gate_resolved, no workflow_failed even. A resume still + # writes a fresh workflow_started. + _event("workflow_started", {"name": "wf"}, ts=2000.0), + _event("agent_started", {"agent_name": "next"}, ts=2001.0), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) + + summary = derive_run_summary(record) + + assert summary.gate is None + assert summary.status == "running" + assert summary.current_step == "next" + + def test_open_steps_from_a_dead_generation_do_not_survive_a_resume( + self, tmp_path: Path + ) -> None: + """An open parallel group from a dead generation must not leak + into the resumed generation's current-step tracking.""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _event("workflow_started", {"name": "wf"}, ts=1000.0), + _event("parallel_started", {"group_name": "fanout"}, ts=1001.0), + _event("workflow_failed", {"error_type": "ValueError"}, ts=1002.0), + _event("workflow_started", {"name": "wf"}, ts=2000.0), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) + + summary = derive_run_summary(record) + + assert summary.current_step is None + assert summary.status == "running" + + def test_run_detail_does_not_report_a_dead_generations_open_gate(self, tmp_path: Path) -> None: + """`derive_run_detail` (the run-detail screen) must reset the same + way `derive_run_summary` (the Runs screen) already does at a resume + boundary -- otherwise the two screens disagree about the same run: + Runs correctly says "running, nothing open" while run-detail keeps + reporting the dead generation's step as still at-gate, with an + elapsed clock spanning the idle gap (issue #485).""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _workflow_started_event(["a"]), + _event("agent_started", {"agent_name": "a"}, ts=101.0), + _event("gate_presented", {"agent_name": "a", "prompt": "OK?"}, ts=102.0), + # Generation 1 dies with the gate still open -- no + # gate_resolved, no workflow_failed. A resume still writes a + # fresh workflow_started. + _workflow_started_event(["a"], ts=5000.0), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) + + summary = derive_run_summary(record) + detail = derive_run_detail(record) + + assert summary.status == "running" + assert summary.current_step is None + assert summary.gate is None + + assert detail.current_step is None + assert [a.name for a in detail.agents] == ["a"] + row = detail.agents[0] + assert row.status != "at-gate" + assert row.status != "running" + + def test_a_nested_subworkflow_start_is_not_a_resume_boundary(self, tmp_path: Path) -> None: + """A nested sub-workflow's own `workflow_started` (`subworkflow_path` + stamped) must not be mistaken for a root resume boundary -- both + `_scan_events` (the Runs screen) and `_scan_agent_details` (the + run-detail screen) guard against this, and this PR widened what + their failure would cost: `_scan_events` now resets + status/gate/open_steps at every `workflow_started` it sees, and + `_scan_agent_details` now also resets topology, so an unguarded + nested start would wipe out the root generation's real state.""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _workflow_started_event(["a"]), + _event("agent_started", {"agent_name": "a"}, ts=1001.0), + _event("gate_presented", {"agent_name": "a", "prompt": "OK?"}, ts=1002.0), + # A nested sub-workflow starts (and completes) while the + # root is still sitting at its gate. + _event( + "workflow_started", + {"name": "sub", "agents": [{"name": "z"}], "subworkflow_path": ["sub"]}, + ts=1003.0, + ), + _event( + "workflow_completed", + {"subworkflow_path": ["sub"]}, + ts=1004.0, + ), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) + + summary = derive_run_summary(record) + detail = derive_run_detail(record) + + assert summary.status == "at-gate" + assert summary.gate is not None + assert summary.current_step == "a" + + assert detail.current_step == "a" + assert [a.name for a in detail.agents] == ["a"] + + +# --------------------------------------------------------------------------- +# Single-pass consumption (mirrors fleet.history's identical issue-#436 test) +# --------------------------------------------------------------------------- + + +class TestScanEventsAcceptsAOneShotIterator: + def test_scan_events_consumes_a_one_shot_generator_exactly_once(self) -> None: + consumed = 0 + + def _events() -> Any: + nonlocal consumed + for evt in ( + {"type": "workflow_started", "timestamp": 1000.0, "data": {"name": "wf"}}, + { + "type": "agent_started", + "timestamp": 1001.0, + "data": {"agent_name": "a"}, + }, + { + "type": "agent_completed", + "timestamp": 1002.0, + "data": {"agent_name": "a", "tokens": 10, "cost_usd": 0.01}, + }, + ): + consumed += 1 + yield evt + + scan = _scan_events(_events()) + + assert consumed == 3 + assert scan.workflow_name == "wf" + assert scan.total_tokens == 10 + assert scan.total_cost_usd == pytest.approx(0.01) + + +# --------------------------------------------------------------------------- +# derive_step_detail: a mid-stream OSError must not surface a partial scan +# --------------------------------------------------------------------------- + + +class TestDeriveStepDetailMidStreamReadFailure: + def test_an_os_error_after_a_completed_step_does_not_report_partial_state( + self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch + ) -> None: + """On `main`, `derive_step_detail` read the whole log in one shot + (`read_event_log_full`), so an I/O failure was structurally + all-or-nothing. Streaming introduced a `try` around the + consumption loop whose locals (`status`, `output`, `activity`) + survive an exception raised mid-iteration -- a step that actually + completed with output must not render as `status="running"`, + `output=None` just because the stream failed partway through + something unrelated afterwards (issue #485 regression).""" + path = tmp_path / "run.events.jsonl" + _write_jsonl( + path, + [ + _workflow_started_event(["a"]), + _event("agent_started", {"agent_name": "a"}, ts=1001.0), + _event( + "agent_completed", + {"agent_name": "a", "output": {"answer": "done"}}, + ts=1002.0, + ), + ], + ) + record = _make_record(tmp_path, event_log_path=str(path)) + + real_stream = summary_module.stream_event_log + + def _flaky_stream(*args: Any, **kwargs: Any) -> Any: + events = list(real_stream(*args, **kwargs)) + + def _generator() -> Any: + yield from events + raise OSError("simulated mid-stream I/O failure") + + return _generator() + + monkeypatch.setattr(summary_module, "stream_event_log", _flaky_stream) + + detail = derive_step_detail(record, "a") + + # The pristine, pre-scan defaults -- never a scan that completed + # (correctly deriving status="completed"/output={"answer": "done"}) + # but was discarded partway through being assigned. + assert detail.status == "pending" + assert detail.prompt is None + assert detail.output is None + assert detail.activity == []