diff --git a/.opencode/telemetry/server.py b/.opencode/telemetry/server.py index 501f3760a..4bbdab72c 100644 --- a/.opencode/telemetry/server.py +++ b/.opencode/telemetry/server.py @@ -869,6 +869,17 @@ def _api_archived_sessions() -> dict[str, Any]: "wallclock_seconds": data.get("wallclock_seconds"), "message_count": len(data.get("messages") or []), "turn_count": len(per_turn) if isinstance(per_turn, list) else 0, + # v2 (2026-05-10) fields: present for subagent archives, + # ``None`` for top-level wrapper archives. The UI can use + # ``parent_session_id`` to group every session from one + # dispatcher cycle into a single tree row, and + # ``subagent_depth`` to nest the table view. Old (v1) + # archives lack these keys; ``.get`` returns None so the + # row schema stays uniform. + "schema_version": data.get("schema_version"), + "parent_session_id": data.get("parent_session_id"), + "subagent_title": data.get("subagent_title"), + "subagent_depth": data.get("subagent_depth"), } ) # Newest archive first. Use ``modified_at`` since it is ISO-formatted diff --git a/CHANGELOG.md b/CHANGELOG.md index b7e732e5d..8fc5363be 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -5,6 +5,96 @@ The format follows [Keep a Changelog](https://keepachangelog.com/en/1.1.0/). ## [Unreleased] +### Added + +- **Subagent session archive walker (2026-05-10).** Before this + change the dispatcher archived only the top-level wrapper + session — the entire ``task``-tool subagent chain + (``tier-dispatcher`` → ``estimator-implementation`` / + ``tier-qwen-med`` → ``task-implementor`` → ``git-isolator-util``) + was opaque the moment the dispatcher's ``DELETE /session/{id}`` + fired. The post-mortem analysis of an implementer run was + therefore limited to whatever live-API polling we'd done DURING + the run, which is unreliable and only catches the agents that + happened to be active when we polled. + + The wrapper's ``run_session_blocking`` ``finally`` block now + walks ``GET /session`` BFS-from-the-root after the root archive + is written and BEFORE the root ``DELETE``. Every descendant + session — at any depth — has its full ``/session/{id}/message`` + stream fetched and written to a per-descendant archive file. + The schema is the same as the wrapper archive (so the existing + telemetry-console endpoints render them) with three new + optional fields: + + - ``parent_session_id`` — wrapper's ``session_id`` for direct + children; the immediate parent's id for deeper nodes. + - ``subagent_title`` — OpenCode's full session title (e.g. + ``"Dispatch implementation to tier-dispatcher (@tier-dispatcher subagent)"``). + - ``subagent_depth`` — 1-based distance from the wrapper. + + Filenames embed a ``sub`` infix + (``__sub01______.json``) so a + ``ls -1 .dispatcher-logs/sessions`` listing groups every session + from one dispatcher cycle and reads top-down in BFS order. The + schema version is bumped to **v2**; v1 readers see ``None`` for + the three new fields and remain forward-compatible (no breaking + change for the telemetry UI). The redaction list passed by the + dispatcher (the Forgejo PAT) propagates into every subagent + archive too, so a ``git clone https://${PAT}@...`` command in + ``git-isolator-util``'s bash history is masked the same way the + wrapper's prompt is. + + Concretely: + + - ``tools/_opencode_worker.py`` gains + ``_walk_subagent_descendants(server_url, root_session_id)`` + (BFS over ``GET /session`` keyed on ``parentID``, returns one + dict per descendant with an added ``_subagent_depth`` field), + ``_extract_subagent_agent_name(title)`` (regex on the + OpenCode title convention with a ``unknown-subagent`` + fallback), ``_ms_to_iso(ms)`` (epoch-ms → ISO-8601 helper), + and ``_archive_subagent_tree(...)`` (best-effort + walk + fetch + write driver; never raises, logs and + continues on per-descendant transport failure). + - ``_archive_session`` and ``_build_archive_payload`` gain + optional ``parent_session_id`` / ``subagent_title`` / + ``subagent_depth`` kwargs. When set, the filename includes + the ``sub`` infix; when unset, the legacy filename + pattern is preserved byte-for-byte. + - ``run_session_blocking``'s ``finally`` block invokes + ``_archive_subagent_tree`` after the root archive write and + before the root ``DELETE``. Both calls are wrapped in + ``try/except BaseException`` so a subagent-walk failure can + never mask the worker outcome or stop the + dispatcher from cleaning up. + - ``.opencode/telemetry/server.py``'s ``_api_archived_sessions`` + listing endpoint now surfaces the three v2 fields in the row + payload so a future UI render can nest subagents under their + wrapper. The existing row shape is preserved (additive only). + - **17 new tests** in ``test_opencode_worker_observability.py``: + ``TestSubagentTitleExtraction`` (5) covers the title parser, + ``TestWalkSubagentDescendants`` (6) covers BFS order, depth + annotation, transport-error / malformed-payload paths, and + cycle safety, ``TestArchiveSubagentTree`` (5) covers + end-to-end orchestration including a redaction-propagation + test that asserts a PAT inside a subagent's + ``bash`` tool input is replaced with ````, and + ``TestEndToEndSubagentArchive`` (1) drives the full + ``run_session_blocking`` lifecycle with a wired subagent + descendant and asserts both archives land on disk. + Total auto_agents suite: **1061 passed, 3 skipped** (up from + 1044). + + Why this matters: the prior post-mortem analysis recommendations + (default-flip ``IMPLEMENTER_DISPATCHER_PREFETCH=1`` etc.) all + rested on live-poll snapshots of two specific subagents I + happened to capture mid-run. With the walker in place, every + cycle leaves a complete trace on disk that a human can read + bottom-up months later, and the telemetry console can render + the wrapper → tier-dispatcher → task-implementor → + git-isolator-util tree as one navigation context. + ### Changed - **Tighten `implementation-worker.md` skill allowlist and Step 0 diff --git a/tests/auto_agents/test_opencode_worker_observability.py b/tests/auto_agents/test_opencode_worker_observability.py index 660304117..4597a61fb 100644 --- a/tests/auto_agents/test_opencode_worker_observability.py +++ b/tests/auto_agents/test_opencode_worker_observability.py @@ -115,6 +115,15 @@ def _wire_lifecycle_with_statuses( httpserver.expect_request( f"/session/{session_id}/message", method="GET" ).respond_with_json(msg_payload) + # Subagent descendant walk (added 2026-05-10): _walk_subagent_descendants + # issues GET /session in the run_session_blocking finally block to find + # any subagent sessions that need archiving. By default the test wires + # this to return ONLY the root session (no descendants), so existing + # tests are subagent-archive-clean. Tests that exercise the subagent + # path should ``httpserver.clear()`` and re-wire with descendants. + httpserver.expect_request("/session", method="GET").respond_with_json( + [_session_obj(session_id)] + ) httpserver.expect_request( f"/session/{session_id}", method="DELETE" ).respond_with_data("", status=200) @@ -566,7 +575,7 @@ def test_archive_completed_session_writes_payload_with_schema( assert len(files) == 1, f"expected one archive file, got {files}" payload = json.loads(files[0].read_text()) - assert payload["schema_version"] == 1 + assert payload["schema_version"] == 2 assert payload["session_id"] == "s-obs-1" assert payload["agent"] == "test-agent" assert payload["tag"] == "AUTO-OBS" @@ -581,6 +590,10 @@ def test_archive_completed_session_writes_payload_with_schema( assert payload["state_history"][0]["state"] in {"busy", "idle", "unknown"} assert isinstance(payload["messages"], list) assert payload["messages"][-1]["info"]["role"] == "assistant" + # v2 subagent-marker fields default to None on a top-level archive. + assert payload["parent_session_id"] is None + assert payload["subagent_title"] is None + assert payload["subagent_depth"] is None def test_archive_transport_error_path_still_writes_payload( @@ -614,6 +627,13 @@ def test_archive_transport_error_path_still_writes_payload( } ] ) + # Subagent walker fires in finally{} on every terminal path including + # transport-error. Wire it to return ONLY the root (no descendants) + # so the warning-on-transport-error path isn't muddied with subagent + # walk noise. + httpserver.expect_request("/session", method="GET").respond_with_json( + [_session_obj()] + ) httpserver.expect_request( "/session/s-obs-1", method="DELETE" ).respond_with_data("") @@ -675,6 +695,10 @@ def test_archive_timeout_path_still_writes_payload( } ] ) + # Subagent walker fires on timeout too — return only the root. + httpserver.expect_request("/session", method="GET").respond_with_json( + [_session_obj()] + ) httpserver.expect_request( "/session/s-obs-1", method="DELETE" ).respond_with_data("") @@ -1072,3 +1096,499 @@ class TestRunSessionRedactsArchive: for rec in caplog.records ) assert short in body + + +# ─── Subagent archive walker (2026-05-10) ─────────────────────────────────── + + +class TestSubagentTitleExtraction: + """``_extract_subagent_agent_name`` parses the ``(@ subagent)`` + suffix OpenCode appends to subagent session titles. The archive + walker relies on this to populate the ``agent`` field of the + subagent archive filename, so the parsing has to be tolerant of + the OpenCode title conventions we've actually seen on live runs. + """ + + def test_extracts_basic_agent_name(self, mod): + title = "Dispatch implementation to tier-dispatcher (@tier-dispatcher subagent)" + assert mod._extract_subagent_agent_name(title) == "tier-dispatcher" + + def test_handles_dotted_agent_name(self, mod): + title = "Foo (@some.dotted.agent subagent)" + assert mod._extract_subagent_agent_name(title) == "some.dotted.agent" + + def test_handles_trailing_whitespace(self, mod): + title = "Foo (@bar subagent) " + assert mod._extract_subagent_agent_name(title) == "bar" + + def test_returns_sentinel_for_non_subagent_title(self, mod): + # Top-level sessions don't carry the ``(@... subagent)`` suffix; + # the walker is only called with descendants, but defensive + # handling means a misclassified record still archives. + assert mod._extract_subagent_agent_name("[AUTO-IMP] foo") == "unknown-subagent" + + def test_returns_sentinel_for_empty(self, mod): + assert mod._extract_subagent_agent_name("") == "unknown-subagent" + assert mod._extract_subagent_agent_name(None) == "unknown-subagent" + + +class TestWalkSubagentDescendants: + """``_walk_subagent_descendants`` walks the ``/session`` payload + BFS from a root session id and returns every descendant. Coverage: + (a) BFS order, (b) depth annotation, (c) transport error path, + (d) malformed payload path, (e) cycle-safety (just in case). + """ + + def _wire_sessions(self, httpserver, sessions: list[dict]): + httpserver.expect_request("/session", method="GET").respond_with_json( + sessions + ) + + def test_returns_empty_for_no_descendants(self, mod, httpserver): + self._wire_sessions( + httpserver, + [{"id": "root", "parentID": "", "title": "[AUTO] root"}], + ) + out = mod._walk_subagent_descendants(_server_url(httpserver), "root") + assert out == [] + + def test_collects_two_level_chain_in_bfs_order(self, mod, httpserver): + # root -> child1 -> grandchild; root -> child2. + # BFS order: child1, child2, grandchild. + self._wire_sessions( + httpserver, + [ + {"id": "root", "parentID": "", "title": "root", "time": {"created": 1}}, + { + "id": "child1", + "parentID": "root", + "title": "Call A (@a subagent)", + "time": {"created": 10}, + }, + { + "id": "child2", + "parentID": "root", + "title": "Call B (@b subagent)", + "time": {"created": 20}, + }, + { + "id": "grandchild", + "parentID": "child1", + "title": "Inner (@inner subagent)", + "time": {"created": 30}, + }, + ], + ) + out = mod._walk_subagent_descendants(_server_url(httpserver), "root") + ids = [s["id"] for s in out] + assert ids == ["child1", "child2", "grandchild"] + depths = [s["_subagent_depth"] for s in out] + assert depths == [1, 1, 2] + + def test_orders_siblings_by_creation_time(self, mod, httpserver): + # Two siblings created out-of-order in the /session response; + # BFS within a depth level should still be creation-time + # ordered so the trace reads left-to-right by clock. + self._wire_sessions( + httpserver, + [ + {"id": "root", "parentID": "", "title": "root"}, + { + "id": "later", + "parentID": "root", + "title": "Later (@x subagent)", + "time": {"created": 999}, + }, + { + "id": "earlier", + "parentID": "root", + "title": "Earlier (@x subagent)", + "time": {"created": 1}, + }, + ], + ) + out = mod._walk_subagent_descendants(_server_url(httpserver), "root") + assert [s["id"] for s in out] == ["earlier", "later"] + + def test_returns_empty_on_transport_error(self, mod, httpserver): + # No /session route wired; httpserver returns 500 for the + # unmatched request and _request raises HTTPError which the + # walker catches. + out = mod._walk_subagent_descendants(_server_url(httpserver), "root") + assert out == [] + + def test_returns_empty_on_non_list_payload(self, mod, httpserver): + httpserver.expect_request("/session", method="GET").respond_with_json( + {"unexpected": "shape"} + ) + out = mod._walk_subagent_descendants(_server_url(httpserver), "root") + assert out == [] + + def test_handles_cycle_without_infinite_loop(self, mod, httpserver): + # Defensive: OpenCode should never report a cycle, but the + # walker should terminate even if it ever does. Each id is + # only visited once thanks to the ``seen`` set. + self._wire_sessions( + httpserver, + [ + {"id": "root", "parentID": "loop-end", "title": "root"}, + { + "id": "loop-mid", + "parentID": "root", + "title": "Mid (@x subagent)", + "time": {"created": 1}, + }, + { + "id": "loop-end", + "parentID": "loop-mid", + "title": "End (@x subagent)", + "time": {"created": 2}, + }, + ], + ) + out = mod._walk_subagent_descendants(_server_url(httpserver), "root") + # Only ``loop-mid`` and ``loop-end`` get visited; the back-edge + # to ``root`` is blocked by the seen-set so we don't recurse. + ids = [s["id"] for s in out] + assert ids == ["loop-mid", "loop-end"] + + +class TestArchiveSubagentTree: + """``_archive_subagent_tree`` orchestrates: walk → fetch each + descendant's messages → write archive per descendant. Coverage: + (a) writes one file per descendant, (b) filename infix carries + ``sub``, (c) payload v2 fields populated, (d) parent's tag + + started_at inherited, (e) per-descendant fetch failure logs and + continues. + + These tests call ``_archive_subagent_tree`` directly (i.e. they + bypass ``run_session_blocking`` and therefore bypass + ``_resolve_archive_dir``). Because of that we have to create the + archive directory ourselves — the autouse fixture only sets the + env var; ``_resolve_archive_dir`` is what calls ``mkdir`` on it. + """ + + @pytest.fixture(autouse=True) + def _ensure_archive_dir_exists(self, archive_dir): + archive_dir.mkdir(parents=True, exist_ok=True) + + def _wire(self, httpserver, descendants: list[dict], messages_for: dict[str, list]): + # Root session is always returned by /session GET — the walker + # uses it as a no-op (excludes root from descendants). + all_sessions = [{"id": "root", "parentID": "", "title": "root"}] + descendants + httpserver.expect_request("/session", method="GET").respond_with_json( + all_sessions + ) + for sid, msgs in messages_for.items(): + httpserver.expect_request( + f"/session/{sid}/message", method="GET" + ).respond_with_json(msgs) + + def test_writes_one_archive_per_descendant( + self, mod, httpserver, archive_dir + ): + descendants = [ + { + "id": "child1", + "parentID": "root", + "title": "Dispatch (@tier-dispatcher subagent)", + "time": {"created": 1_000_000_000_000, "updated": 1_000_000_005_000}, + }, + { + "id": "child2", + "parentID": "root", + "title": "Estimate (@estimator-implementation subagent)", + "time": {"created": 1_000_000_001_000, "updated": 1_000_000_003_500}, + }, + ] + messages_for = { + "child1": [ + { + "info": { + "id": "m1", + "role": "assistant", + "time": {"created": 1, "completed": 2}, + "tokens": {"input": 10, "output": 5}, + }, + "parts": [{"type": "text", "text": "child1 said hi"}], + } + ], + "child2": [ + { + "info": { + "id": "m2", + "role": "assistant", + "time": {"created": 1, "completed": 2}, + "tokens": {"input": 20, "output": 10}, + }, + "parts": [{"type": "text", "text": "child2 said hi"}], + } + ], + } + self._wire(httpserver, descendants, messages_for) + + paths = mod._archive_subagent_tree( + archive_dir=archive_dir, + server_url=_server_url(httpserver), + root_session_id="root", + root_tag="AUTO-IMP-PR-99", + root_started_at_iso="2026-05-10T00:00:00+00:00", + ) + assert len(paths) == 2 + # Files exist on disk and parse. + for p in paths: + assert p.exists() + payload = json.loads(p.read_text()) + assert payload["schema_version"] == 2 + assert payload["parent_session_id"] == "root" + assert payload["tag"] == "AUTO-IMP-PR-99" + assert payload["subagent_depth"] == 1 + assert payload["status"] == "subagent" + assert payload["subagent_title"].startswith(("Dispatch", "Estimate")) + + def test_filename_contains_sub_depth_infix( + self, mod, httpserver, archive_dir + ): + descendants = [ + { + "id": "child1", + "parentID": "root", + "title": "X (@agent-a subagent)", + "time": {"created": 1_000_000_000_000, "updated": 1_000_000_001_000}, + } + ] + messages_for = {"child1": []} + self._wire(httpserver, descendants, messages_for) + paths = mod._archive_subagent_tree( + archive_dir=archive_dir, + server_url=_server_url(httpserver), + root_session_id="root", + root_tag="AUTO-IMP-PR-99", + root_started_at_iso="2026-05-10T00:00:00+00:00", + ) + assert len(paths) == 1 + # Filename layout: __sub01______.json + name = paths[0].name + assert "__sub01__" in name + assert "__AUTO-IMP-PR-99__" in name + assert "__agent-a__" in name + assert name.endswith("__child1.json") + + def test_continues_after_per_descendant_fetch_failure( + self, mod, httpserver, archive_dir, caplog + ): + # Wire two descendants but only respond to one's /message GET + # — the other returns 500 and the walker must log + continue. + descendants = [ + { + "id": "ok", + "parentID": "root", + "title": "OK (@ok-agent subagent)", + "time": {"created": 1_000_000_000_000, "updated": 1_000_000_001_000}, + }, + { + "id": "broken", + "parentID": "root", + "title": "Broken (@broken-agent subagent)", + "time": {"created": 1_000_000_002_000, "updated": 1_000_000_003_000}, + }, + ] + messages_for = {"ok": []} # ``broken`` deliberately omitted + self._wire(httpserver, descendants, messages_for) + + import logging + caplog.set_level(logging.WARNING, logger="opencode_worker") + paths = mod._archive_subagent_tree( + archive_dir=archive_dir, + server_url=_server_url(httpserver), + root_session_id="root", + root_tag="AUTO-IMP-PR-99", + root_started_at_iso="2026-05-10T00:00:00+00:00", + ) + # Only ``ok`` archives; ``broken`` is skipped with a warning. + assert len(paths) == 1 + assert any( + "subagent archive: skipping broken" in rec.getMessage() + for rec in caplog.records + ) + + def test_redact_values_propagate_to_subagent_archive( + self, mod, httpserver, archive_dir + ): + # The PAT in a subagent's tool input must also be redacted — + # the dispatcher passes redact_values=[cfg.token] through. + secret = "FORGEJO-PAT-LONG-SECRET-VALUE-NEVER-COMMIT-3" + descendants = [ + { + "id": "child1", + "parentID": "root", + "title": "Clone (@git-isolator-util subagent)", + "time": {"created": 1_000_000_000_000, "updated": 1_000_000_001_000}, + } + ] + messages_for = { + "child1": [ + { + "info": { + "id": "m1", + "role": "assistant", + "time": {"created": 1, "completed": 2}, + "tokens": {"input": 10, "output": 0}, + }, + "parts": [ + { + "type": "tool", + "tool": "bash", + "state": { + "status": "completed", + "input": { + "command": ( + f"git clone https://{secret}@host/x.git /tmp/x" + ) + }, + }, + } + ], + } + ] + } + self._wire(httpserver, descendants, messages_for) + paths = mod._archive_subagent_tree( + archive_dir=archive_dir, + server_url=_server_url(httpserver), + root_session_id="root", + root_tag="AUTO-IMP-PR-99", + root_started_at_iso="2026-05-10T00:00:00+00:00", + redact_values=[secret], + ) + assert len(paths) == 1 + body = paths[0].read_text() + assert secret not in body, "secret leaked into subagent archive" + assert "" in body + + def test_returns_empty_when_no_descendants_to_walk( + self, mod, httpserver, archive_dir + ): + # Only the root in /session response → nothing to archive. + httpserver.expect_request("/session", method="GET").respond_with_json( + [{"id": "root", "parentID": "", "title": "root"}] + ) + paths = mod._archive_subagent_tree( + archive_dir=archive_dir, + server_url=_server_url(httpserver), + root_session_id="root", + root_tag="AUTO-IMP-PR-99", + root_started_at_iso="2026-05-10T00:00:00+00:00", + ) + assert paths == [] + assert list(archive_dir.glob("*.json")) == [] + + +class TestEndToEndSubagentArchive: + """End-to-end: a real ``run_session_blocking`` call that emits a + subagent descendant in /session → both archives land on disk.""" + + def test_run_session_archives_root_and_subagents( + self, mod, httpserver, archive_dir + ): + # We deliberately do NOT use ``_wire_lifecycle_with_statuses`` + # here because that helper wires its own ``/session`` GET + # (returning only the root) and pytest-httpserver matches + # ``expect_request`` registrations in FIFO order — the helper's + # registration would win over our descendant-bearing override. + # Wire the full lifecycle by hand instead so the walker sees + # the subagent we care about. + httpserver.expect_request("/session", method="POST").respond_with_json( + _session_obj("s-obs-1") + ) + httpserver.expect_request( + "/session/s-obs-1/prompt_async", method="POST" + ).respond_with_json({}) + # status sequence: busy then idle + httpserver.expect_oneshot_request( + "/session/status", method="GET" + ).respond_with_json({"s-obs-1": {"type": "busy"}}) + httpserver.expect_oneshot_request( + "/session/status", method="GET" + ).respond_with_json({"s-obs-1": {"type": "idle"}}) + httpserver.expect_request( + "/session/s-obs-1/message", method="GET" + ).respond_with_json( + [ + { + "info": { + "id": "m-final", + "role": "assistant", + "time": {"created": 1000, "completed": 5000}, + "tokens": {"input": 100, "output": 20, "reasoning": 5}, + }, + "parts": [ + { + "type": "text", + "text": '{"outcome":"resolved","files_touched":[]}', + } + ], + } + ] + ) + # ``/session`` GET returns BOTH the root and one subagent + # descendant. The walker excludes the root and archives the + # subagent. + httpserver.expect_request("/session", method="GET").respond_with_json( + [ + _session_obj("s-obs-1"), + { + "id": "ses-sub-1", + "parentID": "s-obs-1", + "title": "Dispatch (@tier-dispatcher subagent)", + "time": { + "created": 1_000_000_000_000, + "updated": 1_000_000_001_000, + }, + }, + ] + ) + # The walker fetches the descendant's /message — wire it. + httpserver.expect_request( + "/session/ses-sub-1/message", method="GET" + ).respond_with_json( + [ + { + "info": { + "id": "sm1", + "role": "assistant", + "time": {"created": 0, "completed": 1}, + "tokens": {"input": 5, "output": 1}, + }, + "parts": [{"type": "text", "text": "sub said done"}], + } + ] + ) + httpserver.expect_request( + "/session/s-obs-1", method="DELETE" + ).respond_with_data("") + result = mod.run_session_blocking( + server_url=_server_url(httpserver), + agent="test-agent", + tag="AUTO-OBS", + prompt="x", + timeout_seconds=30, + poll_interval_seconds=0.01, + ) + assert result.status == "completed" + files = sorted(archive_dir.glob("*.json")) + # One root + one subagent. + assert len(files) == 2 + names = [p.name for p in files] + # Sorting puts ``2026-...`` (root, started_at first) before + # ````. The depth-1 infix appears in + # the subagent's filename. + assert any("__sub01__" in n for n in names) + # Validate the subagent payload. + sub_file = next(p for p in files if "__sub01__" in p.name) + sub_payload = json.loads(sub_file.read_text()) + assert sub_payload["schema_version"] == 2 + assert sub_payload["parent_session_id"] == "s-obs-1" + assert sub_payload["agent"] == "tier-dispatcher" + assert sub_payload["tag"] == "AUTO-OBS" + assert sub_payload["subagent_depth"] == 1 diff --git a/tools/_opencode_worker.py b/tools/_opencode_worker.py index 9102bdbfc..65e2f2b65 100644 --- a/tools/_opencode_worker.py +++ b/tools/_opencode_worker.py @@ -535,7 +535,29 @@ _ARCHIVE_DISABLED_ENV = "OPENCODE_WORKER_ARCHIVE_DISABLED" # Schema version for the archive payload. Bump when the on-disk JSON # shape changes; the telemetry console reads this to decide which # fields to render. -_ARCHIVE_SCHEMA_VERSION = 1 +# +# v1 (initial) — top-level wrapper sessions only; carries +# session_id, agent, tag, status, started_at, +# archived_at, wallclock_seconds, per_turn, +# state_history, messages. +# v2 (2026-05-10) — adds optional ``parent_session_id``, +# ``subagent_title``, and ``subagent_depth`` fields so +# the dispatcher can archive the entire ``task`` tool +# subagent tree before deleting the root session. The +# first three fields are absent / null on top-level +# (wrapper) archives so a v1 reader treating them as +# missing is forward-compatible. +_ARCHIVE_SCHEMA_VERSION = 2 + +# Pattern that matches the trailing ``(@ subagent)`` suffix +# OpenCode appends to a subagent session's title (e.g. +# ``"Implement PR fix #30 (@task-implementor subagent)"``). The capture +# group is the agent name and is what we use as the ``agent`` field in +# the subagent archive payload. If the title does not match — which +# would mean OpenCode changed its title convention or this is a +# manually-titled top-level session — we fall back to a sentinel string +# so the archive still lands on disk. +_SUBAGENT_TITLE_RE = re.compile(r"\(@([\w.-]+)\s+subagent\)\s*$") def _resolve_archive_dir() -> Path | None: @@ -590,16 +612,28 @@ def _build_archive_payload( wallclock_seconds: float, messages: list[dict[str, Any]], state_history: list[dict[str, Any]] | None = None, + parent_session_id: str | None = None, + subagent_title: str | None = None, + subagent_depth: int | None = None, ) -> dict[str, Any]: """Build the JSON-serialisable archive payload for a session. Pulled out into a helper so :func:`_archive_session` and the test suite share the same shape; the on-disk schema is therefore derivable from this function alone. + + ``parent_session_id`` / ``subagent_title`` / ``subagent_depth`` are + populated only for subagent archives (sessions spawned by a parent's + ``task`` tool call). For the top-level wrapper session they remain + ``None`` and serialise as ``null`` so a downstream reader can tell + the two apart without checking the schema version. """ return { "schema_version": _ARCHIVE_SCHEMA_VERSION, "session_id": session_id, + "parent_session_id": parent_session_id, + "subagent_title": subagent_title, + "subagent_depth": subagent_depth, "agent": agent, "tag": tag, "status": status, @@ -667,6 +701,9 @@ def _archive_session( messages: list[dict[str, Any]], state_history: list[dict[str, Any]] | None = None, redact_values: Iterable[str] | None = None, + parent_session_id: str | None = None, + subagent_title: str | None = None, + subagent_depth: int | None = None, ) -> Path | None: """Write the session archive JSON file. Returns the resolved path on success, or ``None`` on any I/O / serialisation failure. @@ -677,13 +714,25 @@ def _archive_session( disk. The redaction runs on the serialised JSON so it catches the value in any nested location (prompt text, tool input, error string). See :func:`_redact_secret_values` for the safety floor. + + ``parent_session_id`` / ``subagent_title`` / ``subagent_depth`` are + forwarded into the payload to support subagent archiving (see + :func:`_archive_subagent_tree`). For top-level (wrapper) archives + they should remain ``None``. """ - fname = ( - f"{_safe_filename_component(started_at_iso)}__" - f"{_safe_filename_component(tag)}__" - f"{_safe_filename_component(agent)}__" - f"{_safe_filename_component(session_id)}.json" - ) + fname_parts = [ + _safe_filename_component(started_at_iso), + _safe_filename_component(tag), + _safe_filename_component(agent), + _safe_filename_component(session_id), + ] + # Subagent archives get a ``sub`` infix so a directory listing + # sorts the wrapper first, then its subagents in BFS order + # (sub1 before sub2 before sub3...). This makes ``ls -1`` a usable + # post-mortem trace tool without needing the telemetry console. + if subagent_depth is not None: + fname_parts.insert(1, f"sub{subagent_depth:02d}") + fname = "__".join(fname_parts) + ".json" path = archive_dir / fname payload = _build_archive_payload( session_id=session_id, @@ -694,6 +743,9 @@ def _archive_session( wallclock_seconds=wallclock_seconds, messages=messages, state_history=state_history, + parent_session_id=parent_session_id, + subagent_title=subagent_title, + subagent_depth=subagent_depth, ) try: # Render with a default=str so anything urllib hands back that @@ -708,6 +760,212 @@ def _archive_session( return path +def _walk_subagent_descendants( + server_url: str, + root_session_id: str, +) -> list[dict[str, Any]]: + """Walk OpenCode's ``GET /session`` response BFS starting from + ``root_session_id`` and return every descendant session record. + + The returned list excludes ``root_session_id`` itself and is sorted + by depth-then-creation-time, which matches the natural order a + human reads a trace ("the wrapper called tier-dispatcher first, + which called the estimator, which called…"). Each record is the + raw session dict from OpenCode with one added field: + ``_subagent_depth`` (1-based distance from the root). + + Returns ``[]`` on any transport error or unexpected response + shape — the caller's archive flow must continue regardless, so we + never raise here. A short warning is logged so a missed walk is + visible in the dispatcher log. + """ + try: + all_sessions = _request("GET", f"{server_url}/session") + except _TRANSPORT_EXC as e: + logger.warning( + "subagent walk failed (transport): %s; archiving root only", e + ) + return [] + if not isinstance(all_sessions, list): + logger.warning( + "subagent walk: /session returned unexpected shape " + "(%s); archiving root only", + type(all_sessions).__name__, + ) + return [] + + # parent_id -> [child_session_dict, ...] + children_by_parent: dict[str, list[dict[str, Any]]] = {} + for s in all_sessions: + if not isinstance(s, dict): + continue + pid = s.get("parentID") or s.get("parent_id") or "" + if not pid: + continue + children_by_parent.setdefault(pid, []).append(s) + # Sort children by creation time so the trace order is stable. + for kids in children_by_parent.values(): + kids.sort( + key=lambda k: (k.get("time") or {}).get("created", 0) or 0 + ) + + descendants: list[dict[str, Any]] = [] + seen: set[str] = {root_session_id} + # BFS frontier: list of (session_id, depth). + frontier: list[tuple[str, int]] = [(root_session_id, 0)] + while frontier: + parent_id, parent_depth = frontier.pop(0) + for child in children_by_parent.get(parent_id, []): + cid = child.get("id") + if not isinstance(cid, str) or cid in seen: + continue + seen.add(cid) + depth = parent_depth + 1 + child_with_depth = dict(child) + child_with_depth["_subagent_depth"] = depth + descendants.append(child_with_depth) + frontier.append((cid, depth)) + return descendants + + +def _extract_subagent_agent_name(title: str | None) -> str: + """Return the ``agent`` name from a subagent session title. + + Subagent sessions in OpenCode are titled + ``" (@ subagent)"`` — see + :data:`_SUBAGENT_TITLE_RE`. Returns ``"unknown-subagent"`` if the + title is missing or does not match the convention, so the archive + still lands on disk with a useful (if generic) filename component. + """ + if not isinstance(title, str) or not title: + return "unknown-subagent" + m = _SUBAGENT_TITLE_RE.search(title) + if not m: + return "unknown-subagent" + return m.group(1) + + +def _ms_to_iso(ms: int | float | None) -> str | None: + """Convert a Unix-epoch millisecond timestamp to an ISO-8601 + UTC string, or return ``None`` if ``ms`` is not a usable number. + + OpenCode's ``/session`` payload reports ``time.created`` / + ``time.updated`` in milliseconds since the epoch. We carry that + forward into the archive as an ISO string so the on-disk format + matches the wrapper archive's ``started_at`` field. + """ + if ms is None: + return None + try: + seconds = float(ms) / 1000.0 + except (TypeError, ValueError): + return None + return _dt.datetime.fromtimestamp( + seconds, tz=_dt.timezone.utc + ).isoformat() + + +def _archive_subagent_tree( + *, + archive_dir: Path, + server_url: str, + root_session_id: str, + root_tag: str, + root_started_at_iso: str, + redact_values: Iterable[str] | None = None, +) -> list[Path]: + """Walk every subagent descendant of ``root_session_id``, fetch + its message stream, and write an archive file per session. + + Returns the list of archive paths that were written (in BFS order). + Always runs best-effort: a transport error on any single + subagent's message fetch logs a warning and continues to the next + descendant. The caller MUST invoke this BEFORE deleting the root + session — OpenCode may garbage-collect subagent sessions when the + root is removed. + + ``root_tag`` and ``root_started_at_iso`` are inherited so a + directory listing groups every session from one dispatcher cycle + together. The subagent's own per-session ``time.created`` is + captured under the new ``subagent_started_at`` slot for the + timeline view in the telemetry console. + """ + descendants = _walk_subagent_descendants(server_url, root_session_id) + if not descendants: + return [] + paths: list[Path] = [] + for desc in descendants: + sid = desc.get("id") + if not isinstance(sid, str) or not sid: + continue + title = desc.get("title") or "" + agent_name = _extract_subagent_agent_name(title) + depth = int(desc.get("_subagent_depth") or 1) + time_info = desc.get("time") or {} + created_ms = time_info.get("created") + updated_ms = time_info.get("updated") + sub_started_iso = _ms_to_iso(created_ms) or root_started_at_iso + sub_wallclock = 0.0 + if isinstance(created_ms, (int, float)) and isinstance( + updated_ms, (int, float) + ): + sub_wallclock = max( + 0.0, (float(updated_ms) - float(created_ms)) / 1000.0 + ) + + try: + messages = _request( + "GET", f"{server_url}/session/{sid}/message" + ) + except _TRANSPORT_EXC as e: + logger.warning( + "subagent archive: skipping %s (%s) — transport: %s", + sid, agent_name, e, + ) + continue + if not isinstance(messages, list): + logger.warning( + "subagent archive: skipping %s (%s) — unexpected " + "messages shape %s", + sid, agent_name, type(messages).__name__, + ) + continue + + # Inherit the parent's tag so a ``ls -1 .dispatcher-logs/sessions`` + # listing groups every session from this dispatcher cycle + # together. The subagent's own creation time goes into + # ``subagent_started_at`` inside the payload. + try: + written = _archive_session( + archive_dir=archive_dir, + session_id=sid, + agent=agent_name, + tag=root_tag, + status="subagent", + started_at_iso=sub_started_iso, + wallclock_seconds=sub_wallclock, + messages=messages, + state_history=None, + redact_values=redact_values, + parent_session_id=desc.get("parentID") or root_session_id, + subagent_title=title, + subagent_depth=depth, + ) + except Exception as e: # noqa: BLE001 — best-effort, never raise + logger.warning( + "subagent archive raised for %s (%s): %s", + sid, agent_name, e, + ) + continue + if written is not None: + paths.append(written) + logger.info( + "subagent %s (depth=%d agent=%s) archived to %s", + sid, depth, agent_name, written, + ) + return paths + + def _summarize_per_turn(messages: list[dict[str, Any]]) -> list[dict[str, Any]]: """Aggregate per-assistant-turn metrics out of a /session/{id}/message payload. @@ -1442,6 +1700,26 @@ def run_session_blocking( "session archive raised for %s: %s", session_id, e, ) + # Subagent tree archive — runs AFTER the root archive + # but BEFORE the DELETE below, so the dispatcher + # captures the complete ``task`` tool chain even if + # OpenCode garbage-collects subagent sessions on + # root deletion. Best-effort: any failure logs a + # warning and lets the DELETE proceed. + try: + _archive_subagent_tree( + archive_dir=archive_dir, + server_url=server_url, + root_session_id=session_id, + root_tag=tag, + root_started_at_iso=started_at_iso, + redact_values=redact_snapshot, + ) + except Exception as e: # noqa: BLE001 — defensive belt-and-braces + logger.warning( + "subagent tree archive raised for root %s: %s", + session_id, e, + ) try: _request("DELETE", f"{server_url}/session/{session_id}") except _TRANSPORT_EXC as e: