diff --git a/CHANGELOG.md b/CHANGELOG.md index 93af680..f83ef24 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -34,6 +34,16 @@ story `, the same annotation a sweep bundle writes. Both sprint and stories upstream skill emits it yet (BMAD-METHOD#2619 proposes it at breakdown), and re-deriving `stories.yaml` drops it unless the intent is logged in `.memlog.md`. +- **Readable run logs for `opencode-http` (#306).** Contributed by + [@jackmcintyre](https://github.com/jackmcintyre). The HTTP adapter has no tmux pane to replay, + so a finished opencode run left `logs/.log` holding nothing but the server's own INFO + stdout. Per-session logs now split three ways, rendered off the SSE stream the adapter already + reads for completion: + - `.log` — curated transcript: role-prefixed agent/user prose plus `[bmad]` marker + lines for tool calls, slash commands, file edits, permission asks/replies and session errors. + - `.server.out` — the server's own stdout, no longer mixed into the transcript. + - `.sse.jsonl` — structured trace of the raw SSE frames for post-hoc debugging. + - **`review.on_timeout` policy knob (#271).** A timeout-like review verdict (`timeout` / `stalled` / `over_budget`) previously always burned a review cycle (RETRY) until `max_review_cycles`, then deferred — even when the dev product was already finalized and diff --git a/docs/FEATURES.md b/docs/FEATURES.md index 53e8da5..466d3f4 100644 --- a/docs/FEATURES.md +++ b/docs/FEATURES.md @@ -133,7 +133,7 @@ See [README.md](../README.md) for the narrative overview and [setup-guide.md](se - Generic adapter drives any CLI fitting the injection + hook-signal transport; CLI specifics live in declarative TOML profiles. Two independent axes: the **CLI** (`CodingCLIAdapter` + profile) and the **terminal transport** (`TerminalMultiplexer`) — tmux ships bundled (with an experimental native-Windows `psmux` backend alongside it), and external backends (e.g. the [herdr adapter](https://github.com/pbean/bmad-loop-adapter-herdr)) co-install as packages that self-register ([how](multiplexer-backends.md)), behind a pluggable seam that lets a new backend slot in without touching the engine (see the [adapter authoring guide](adapter-authoring-guide.md#two-axes-cli-vs-transport)). - The OS is abstracted by a **registry of seams**, each selecting an implementation by platform (with a test-override env var) and extended by a single registration line: the terminal multiplexer (`register_multiplexer`, with availability-aware selection: env var → persisted `[mux] backend` via `bmad-loop mux set` → platform default → first available platform match), the process-lifecycle `ProcessHost` (`register_process_host` — `terminate`/`force_kill`/`is_alive`/`identity`), and the hook interpreter (`ProcessHost.hook_interpreter()`); `bmad-loop validate` runs a platform preflight over them. Porting to a new OS is new files + registrations, no core edits — see [Porting bmad-loop to a new OS](porting-to-a-new-os.md). - Supported, E2E-verified: `claude` (reference), `codex` (≥ 0.139), `gemini` (≥ 0.46), `copilot` (GitHub Copilot CLI ≥ 2026-02 — the `copilot` binary, not the VS Code extension; `agentStop` turn-end, `-i` interactive launch, `--allow-all-tools`; pin a capable model — the free GPT-5 mini default is unreliable for multi-step skills). -- Supported, E2E-verified over HTTP/SSE (no tmux window): `opencode` (OpenCode ≥ 1.18, profile `opencode-http`, alias `opencode`) — one headless `opencode serve` per session, SSE `session.idle` completion with an HTTP poll fallback, per-session server password, token usage read back over the API. Hookless (`[hooks] dialect = "none"`, no hook registration). Install the extra (`pip install 'bmad-loop[opencode]'`), auth once globally (`opencode auth login`), and set `model` as `provider/model`; the Unity plugin's window guards don't apply (there is no window). +- Supported, E2E-verified over HTTP/SSE (no tmux window): `opencode` (OpenCode ≥ 1.18, profile `opencode-http`, alias `opencode`) — one headless `opencode serve` per session, SSE `session.idle` completion with an HTTP poll fallback, per-session server password, token usage read back over the API. Hookless (`[hooks] dialect = "none"`, no hook registration). With no pane to replay, the run logs split three ways: a curated readable transcript in `logs/.log` (agent/user prose, tool calls, slash commands, file edits, permission asks/replies, errors), the server's own stdout in `.server.out`, and a structured SSE trace in `.sse.jsonl`. Install the extra (`pip install 'bmad-loop[opencode]'`), auth once globally (`opencode auth login`), and set `model` as `provider/model`; the Unity plugin's window guards don't apply (there is no window). - Experimental, `isolation = "none"` only: `antigravity` (Google's `agy` ≥ 1.1.3) — `-i` interactive launch, `Stop` turn-end hook (flat handler in `.agents/hooks.json`, no SessionStart/SessionEnd), `--dangerously-skip-permissions` for unattended runs; `usage_parser = "none"` permanently — agy's transcript exposes no usage data (tokens live only in an internal SQLite/protobuf store). `agy` gates each workspace on an exact-path `trustedWorkspaces` entry and blocks on an interactive trust dialog, which `--dangerously-skip-permissions` does not bypass — so worktree isolation hangs ([#169](https://github.com/bmad-code-org/bmad-loop/issues/169)). Verify against your `agy` build with `probe-adapter antigravity`. - Per-stage CLI/model overrides: run dev on one CLI/model, review on another (`[adapter.dev]`, `[adapter.review]`, `[adapter.triage]`). - Add a CLI without touching Python: drop a TOML profile in `.bmad-loop/profiles/.toml` (binary, prompt template, bypass flags, hook dialect, native→canonical event map). diff --git a/docs/adapter-authoring-guide.md b/docs/adapter-authoring-guide.md index 24ee9f0..c44f7a0 100644 --- a/docs/adapter-authoring-guide.md +++ b/docs/adapter-authoring-guide.md @@ -514,6 +514,20 @@ decisions worth stealing: loss. Completion evidence is gated by a forward-advancing floor so an idle event without proof of new work never completes a session — the same artifact-distrust invariant the tmux adapters enforce. +- **A transport with no pane still owes the operator a transcript.** The tmux + adapters get `logs/.log` for free by replaying the pane; over HTTP that + file would hold nothing but `opencode serve`'s own INFO stdout, leaving finished + runs unreadable. So the transcript is _curated_ off the SSE stream the adapter + already consumes for control — a second consumer on the same + `sessionID`-filtered dispatch, costing one `write()` per interesting frame. + Three sinks, one per audience: `.log` is the readable transcript + (role-prefixed prose plus `[bmad]`-marked `tool:` / `cmd:` / `file:` / + `perm ask:` / `perm reply:` / `error:` lines), `.server.out` takes the + server's own stdout so it cannot drown that transcript, and + `.sse.jsonl` keeps the raw frames for post-hoc debugging (behind the + `sse_trace` module knob, per-token deltas excluded). The catch of curating: + what you render is only as good as the event names you pinned — check every + branch against the running binary, not against the names that read plausibly. - **Hookless profile.** The profile sets `[hooks] dialect = "none"`: no hook registration, no hook-config merge into worktrees; `init`, `validate` and worktree provisioning all understand `profile.hookless`. Skills still install diff --git a/docs/setup-guide.md b/docs/setup-guide.md index 2871265..265d3e9 100644 --- a/docs/setup-guide.md +++ b/docs/setup-guide.md @@ -274,7 +274,10 @@ them to whoever owns the machine: workspace-trust dialog to answer). Requires OpenCode ≥ 1.18. Set the model as `provider/model` (e.g. `[adapter] model = "anthropic/claude-haiku-4-5"`). No hooks are registered — the adapter drives a headless `opencode serve` over HTTP/SSE, so there is no - tmux window to attach to; watch a session via its `logs/.log` or the TUI Log tab. + tmux window to attach to; watch a session via its `logs/.log` — a curated + transcript of the agent's prose, tool calls, file edits and permission decisions — or the + TUI Log tab. The server's own stdout lands beside it in `.server.out`, and a + structured trace of the raw SSE frames in `.sse.jsonl`. ### Skill location diff --git a/src/bmad_loop/adapters/opencode_http.py b/src/bmad_loop/adapters/opencode_http.py index b2f1c87..50e9180 100644 --- a/src/bmad_loop/adapters/opencode_http.py +++ b/src/bmad_loop/adapters/opencode_http.py @@ -36,6 +36,62 @@ - All ``time.*`` fields are epoch **milliseconds**; ``POST /session/:id/abort`` returns ``200 true`` even when nothing is running. +``/event`` frame types the run-log renderer reads — pinned by a second live +probe of 1.18.2 (2026-07-25, 388 frames over 5 turns across two servers; the +full evidence table is posted on PR #279). Cross-checked against the server's +own OpenAPI at ``GET /doc``, whose ``Event`` union has 89 variants and is the +static contract — worth re-dumping on any version bump: + +- ``message.updated`` — turn metadata only, no text. Carries ``info.id`` + + ``info.role``; **re-emits out of order** (a stale re-emit for the user + message lands mid-assistant-turn), so role must be keyed by message id, never + tracked as "last role seen". +- ``message.part.updated`` — **complete-once, NOT cumulative.** A part fires + exactly twice: once at creation with ``text: ""``, then once carrying the + full final text. Held at 4 / 711 / 5689 chars (2 fires each, 1 non-empty). + Rendering on this frame therefore yields one clean line per statement, with + no de-duplication needed. **Tool execution also surfaces here** — and only + here — as ``part.type == "tool"`` with ``part.tool`` and a ``state`` whose + ``status`` steps pending → running → terminal (``state.input`` throughout, + ``state.output`` on ``completed``, ``state.error`` on ``error``). Tool parts + carry no ``part.text``, so the tool branch must sit ABOVE the empty-text + bail-out in ``_inline_line`` to be reachable at all. ``ToolState`` is a + 4-variant union (pending / running / completed / error), so rendering on the + terminal pair is exhaustively once-per-tool-call; ``error`` is included from + the static contract (no tool failed during the live probe) so a failed call + cannot vanish from the transcript. +- ``message.part.delta`` — the separate per-token stream. At 5689 chars its 50 + deltas concatenate **byte-exactly** to the single ``part.updated`` text, so + it is pure redundancy: never rendered inline, and excluded from the SSE trace + (``logs/`` is never trimmed by retention). +- ``command.executed`` — the **slash-command** surface, kept as that and nothing + more. It does **not** fire for agent tool use: 0 frames across live ``bash`` / + ``write`` / ``read`` / ``edit`` calls. Payload is ``name`` / ``arguments`` / + ``messageID`` / ``sessionID`` (all required, so it needs no allowlist). +- ``file.edited`` — payload is exactly ``{"file": "/abs/path"}``, schema-pinned + ``additionalProperties: false`` over that one key, so it carries **no + ``sessionID``** and the ``properties.sessionID`` filter would drop it before + either sink. It is reachable only through the explicit ``_SESSIONLESS_TYPES`` + allowlist, sound here only because bmad-loop runs one server per session. + ``file.watcher.updated`` is session-less too and deliberately NOT allowlisted + (see that constant). +- ``permission.asked`` — the ask frame. There is **no ``permission.updated``** in + 1.18.2. Payload is a ``permission`` **string** plus ``patterns`` / + ``metadata`` / ``always`` / ``tool`` / ``id`` / ``sessionID`` — not a nested + object with ``type`` / ``pattern``. +- ``permission.replied`` — carries **``reply``**, not ``response`` (plus + ``sessionID`` / ``requestID``). +- ``permission.v2.asked`` / ``permission.v2.replied`` exist in the union and are + **deliberately not consumed**: neither was seen live, and the v2 ask is a + different payload (``action`` / ``resources`` / ``save`` / ``metadata`` / + ``source``), so a branch for it would be written from the schema alone — + exactly the guess that made the three corrections above necessary. The v2 + reply is shape-identical to the v1 one, but consuming it without the ask would + log a decision with no request. Add both together once a live sighting pins + what ``action`` / ``resources`` actually hold. +- There is **no ``tool.call`` / ``tool.response``** on this surface — + confirmed both statically in the 89-variant union and live. + Transport shape (the settled design drivers): - **One ``opencode serve`` per session.** The API has no per-session env, and @@ -121,6 +177,31 @@ REAP_POLL_S = 0.1 SPAWN_ATTEMPTS = 3 POLL_TICK_S = 5.0 # max event-queue wait per loop tick (generic's cadence) +# Structured SSE trace (``logs/.sse.jsonl``). Tier-1 knob deliberately: +# a module default plus the ``sse_trace`` instance attribute, NOT a policy field +# in core.toml. It changes nothing about how a run behaves — it only decides +# whether a debugging artifact is written — so it belongs with the timing knobs +# an operator flips in a REPL or a subclass, not in the run contract every +# settings file has to carry. Off means the sink is never opened. +SSE_TRACE = True + +# ``/event`` frame types that carry NO ``properties.sessionID`` but are still +# attributed to this session, exempting them from the sessionID filter in +# `_dispatch_sse`. Sound ONLY because bmad-loop spawns one `opencode serve` per +# session (see the transport notes in the module docstring): the server is +# single-tenant, so a server-global frame is unambiguously this session's. Any +# design that ever shares a server across sessions must empty this set. +# +# Deliberately an allowlist and not "allow every session-less frame": the live +# probe's session-less traffic was overwhelmingly noise that says nothing about +# the run — `plugin.added` alone was 90 of 388 frames, plus `server.heartbeat`, +# `catalog.updated`, `server.connected`, `reference.updated` and +# `integration.updated`. `file.watcher.updated` is excluded on purpose too: it +# is the editor-integration watcher firing for any change under the project +# (our own git operations, a build, an editor save), so it is not evidence the +# agent did anything, and for the changes the agent DID make it just +# double-reports what `file.edited` already carries. +_SESSIONLESS_TYPES = frozenset({"file.edited"}) # Fixed basic-auth username in OPENCODE_SERVER_PASSWORD mode (Bearer is rejected). AUTH_USER = "opencode" @@ -178,6 +259,58 @@ def _parse_sse_lines(lines) -> Any: data.append(line[5:].lstrip()) +def _sum_args(arguments: Any) -> str: + """Compact one-line summary of an arbitrary event payload value (a tool + part's ``state.input``, a ``permission.asked`` ``patterns``, a + ``command.executed`` ``arguments``, a ``session.error`` ``error``) for the + inline run log. Truncates long dicts/strings so a single line never explodes + the tail view, and accepts any shape — these payloads are server-controlled + and not all of them have a pinned schema.""" + if not arguments: + return "" + try: + if isinstance(arguments, (dict, list)): + text = json.dumps(arguments, ensure_ascii=False) + else: + text = str(arguments) + except (TypeError, ValueError): + text = str(arguments) + if len(text) > 120: + text = text[:119] + "\u2026" + return text + + +# ANSI colors for inline-log marker lines, keyed by message role. The bmad-loop +# TUI renders .log through a pyte terminal emulator (tui/data.py LogView), +# which dispatches SGR to per-Char styles and maps pyte named colors via +# _rich_color straight through to Rich — so every [bmad] marker shows colored in +# the LogView. A plain `cat`/editor shows the escape bytes; the structured trace +# stays uncoloured in .sse.jsonl. Keys are opencode message.info.role +# values; an unseen role falls back to _ROLE_COLOR_DEFAULT so it still renders +# distinctly — add it to the map to pin its hue. +_ROLE_COLORS = { + "user": "\x1b[33m", # SGR 33 = yellow + "assistant": "\x1b[36m", # SGR 36 = cyan +} +_ROLE_COLOR_DEFAULT = "\x1b[35m" # SGR 35 = magenta — unseen roles +_TOOL_COLOR = "\x1b[32m" # SGR 32 = green — tool/cmd/file/permission (role-less) +_RESET = "\x1b[0m" + +# Tool-part statuses that end a tool call. `ToolState` is a 4-variant union in +# the 1.18.2 OpenAPI (`pending`, `running`, `completed`, `error`), so these two +# are exhaustively the terminal ones — a tool call reaches exactly one of them, +# which is what makes "render on terminal" equal to "one line per tool". The +# live probe only ever saw pending → running → completed (nothing failed during +# it); `error` comes from the static contract, and including it is what keeps a +# FAILED tool call from vanishing from the transcript entirely. +_TOOL_TERMINAL_STATES = frozenset({"completed", "error"}) + + +def _role_color(role: str) -> str: + """ANSI color for a message role; unseen roles fall back to magenta.""" + return _ROLE_COLORS.get(role, _ROLE_COLOR_DEFAULT) + + @dataclass class _ServerSession: """Everything the adapter tracks for one live ``opencode serve``.""" @@ -187,6 +320,26 @@ class _ServerSession: base_url: str password: str log_fh: Any + # The spawned server's own stdout/stderr sink (``.server.out``), + # kept separate from ``log_fh`` so the readable transcript stays clean. The + # server's INFO/diagnostic lines land here; ``log_fh`` carries only the + # curated ``[bmad]`` lines written from the SSE reader thread. + server_fh: Any = None + # Structured SSE-trace JSONL sink (``.sse.jsonl``); None when the + # ``sse_trace`` knob is off. Written from the SSE reader thread only. + event_fh: Any = None + # Monotonic line number stamped into the trace. Per SESSION, while the file + # is per TASK and opened in append mode — so a retried task restarts the seq + # at 1 partway down the file. Read a run of seq back to 1 as "new session", + # not as corruption; the ``ts`` field orders records across the whole file. + event_seq: int = 0 + # Role keyed by opencode message id (``msg_*``), refreshed from + # ``message.updated.info.{id,role}``. The per-part / delta events carry a + # ``messageID`` but no role of their own, and ``message.updated`` frames + # arrive out of order (re-emits for an earlier message land mid-turn), so a + # "last role seen" would mislabel the assistant's reply as ``user:``. Keying + # by message id makes the lookup ordering-independent. + msg_roles: dict = field(default_factory=dict) client: Any = None # control httpx.Client — main thread only session_id: str = "" events: queue.Queue = field(default_factory=queue.Queue) @@ -259,6 +412,7 @@ def __init__( self.reconnect_sleep_s = RECONNECT_SLEEP_S self.kill_wait_s = KILL_WAIT_S self.poll_tick_s = POLL_TICK_S + self.sse_trace = SSE_TRACE # False = never open the .sse.jsonl sink self.result_grace_s: float | None = None # None = the mixin default self.tasks_dir = run_dir / "tasks" self.logs_dir = run_dir / LOGS_DIR @@ -336,6 +490,33 @@ def _spawn_server(self, spec: SessionSpec) -> _ServerSession: env = self._session_env(spec, password) log_path = self.logs_dir / f"{spec.task_id}.log" log_fh = log_path.open("ab") # append: retries share one log + event_fh = None + server_fh = None + # The remaining sinks open under their own guard: each `open()` can fail + # on its own (ENOSPC, EMFILE, a permission race) with the earlier ones + # already open, and that happens ABOVE the retry loop's try/except — so + # without this the handles opened so far never reach `_close_spawn_sinks` + # and stay open for as long as the propagating traceback keeps this frame + # alive. One sink had nothing to leak here; three do. + try: + # Structured SSE trace, off entirely when the knob is down. Append + # like the other two sinks: the file is per TASK and every retry of + # that task writes into it, while `event_seq` is per SESSION and + # restarts at 1 for each one — so a seq dropping back to 1 mid-file + # marks a new session, not a truncation. `ts` (epoch ms) is the + # cross-session ordering key. + if self.sse_trace: + event_path = self.logs_dir / f"{spec.task_id}.sse.jsonl" + event_fh = event_path.open("a", encoding="utf-8") # one record per line + # The server's own stdout/stderr (INFO/diagnostic lines) is kept in a + # separate file so the readable transcript in .log stays + # clean. This is also where a failed spawn's diagnostics land — the + # give-up error below names this path, not log_path. + server_path = self.logs_dir / f"{spec.task_id}.server.out" + server_fh = server_path.open("ab") # append: retries share one server log + except BaseException: + self._close_spawn_sinks(log_fh, server_fh, event_fh) + raise last_error = "server did not become healthy" try: for _ in range(SPAWN_ATTEMPTS): @@ -344,7 +525,7 @@ def _spawn_server(self, spec: SessionSpec) -> _ServerSession: self._serve_argv(resolved, port), cwd=str(spec.cwd), env=env, - stdout=log_fh, + stdout=server_fh, stderr=subprocess.STDOUT, stdin=subprocess.DEVNULL, ) @@ -354,6 +535,8 @@ def _spawn_server(self, spec: SessionSpec) -> _ServerSession: base_url=f"http://127.0.0.1:{port}", password=password, log_fh=log_fh, + server_fh=server_fh, + event_fh=event_fh, ) if self._await_healthy(sess): sess.client = self._make_client(sess) @@ -366,14 +549,34 @@ def _spawn_server(self, spec: SessionSpec) -> _ServerSession: else: last_error = f"server exited rc={process.returncode} during startup" except BaseException: - log_fh.close() + self._close_spawn_sinks(log_fh, server_fh, event_fh) raise - log_fh.close() + self._close_spawn_sinks(log_fh, server_fh, event_fh) raise OpencodeServerError( f"could not start `{self.binary} serve` after {SPAWN_ATTEMPTS} attempts " - f"({last_error}); log: {log_path}" + f"({last_error}); server log: {server_path}" ) + @staticmethod + def _close_spawn_sinks(log_fh: Any, server_fh: Any, event_fh: Any) -> None: + """Close whichever of the three per-task sinks are open, on the spawn + failure paths. Any of them can be None: `event_fh` when the sse_trace + knob is off, and either of the last two when the failure WAS the open + that would have created it. + + Every close is guarded, matching `_teardown`. Both callers are already + reporting a failure — one is mid-`raise`, the other is about to raise + `OpencodeServerError` — so an OSError from a flush-on-close would + replace the failure actually worth reporting, and would strand the sinks + after it in this loop.""" + for fh in (log_fh, server_fh, event_fh): + if fh is None: + continue + try: + fh.close() + except OSError: + pass + def _await_healthy(self, sess: _ServerSession) -> bool: """Poll /global/health (authenticated) until it answers healthy. The process liveness is re-checked after *every* probe: a foreign server @@ -519,13 +722,251 @@ def _dispatch_sse(self, sess: _ServerSession, event: Any) -> None: # streams, exactly like subagent output in a tmux pane log). sess.activity += 1 props = event.get("properties") or {} - if props.get("sessionID") != sess.session_id: + # Session-scoped frames must name this session (child/subagent sessions + # share the stream). A short allowlist of SESSION-LESS types passes + # anyway: some frames worth logging carry no sessionID at all, so there + # is no id to match and this filter would drop them before either sink. + # See `_SESSIONLESS_TYPES` for which ones, why it is an allowlist, and + # what deliberately stays out. + # + # Both sinks sit below this gate, so an allowlisted frame is traced as + # well as rendered. That is intended: the trace's contract is one record + # per frame the adapter acted on, and it is the file you read to explain + # a line in the transcript — a rendered line with no matching record + # would make the two sinks disagree. It costs nothing here because the + # allowlist admits one low-rate type, not the noisy session-less bulk. + if props.get("sessionID") != sess.session_id and not ( + # `in` on a frozenset raises TypeError for an unhashable value, and + # `type` is server-controlled, so narrow before the membership test + # (the isinstance check below this filter is deliberately kept there + # — moving it up would change the activity counter above). + isinstance(etype, str) + and etype in _SESSIONLESS_TYPES + ): return + if not isinstance(etype, str): + # Every branch below compares `type` against a string literal, so a + # frame carrying a non-string (or absent) type can match none of + # them. Dropping it here narrows the value for the typed helpers and + # keeps a malformed frame out of the trace — it is not a frame the + # adapter acted on. Liveness/activity above already counted it. + return + # Before the control queue: emit a structured trace record and render + # one human-readable line into the run log. Both helpers no-op on + # unknown types, so unhandled frames leave both sinks byte-identical to + # today, and idle/error still queue below — control flow is unchanged. + # + # Guarded as a block because these two sinks sit UPSTREAM of the + # idle/error queue puts on a frame-shaped payload we do not control: + # both helpers walk nested `.get` chains (props["part"], ["info"], + # ["permission"]), so a server that ships a string where a dict is + # expected raises AttributeError here. Unguarded that unwinds all the + # way to the reader's connection-level catch in `_sse_loop`, which tears + # the stream down, reconnects, and drops whatever the server had already + # buffered — a logging bug silently degrading completion signaling. Same + # never-die doctrine as the reader itself: logging is advisory, the + # queue puts below are not. `_emit_event` runs first (it is the simpler + # of the two) so a render bug still leaves the offending frame recorded + # in the trace that would be used to diagnose it. + try: + self._emit_event(sess, etype, props) + self._render_inline(sess, etype, props) + except Exception: # nosec B110 - logging must never disturb the queue puts below + pass if etype == "session.idle": sess.events.put("idle") elif etype == "session.error": sess.events.put("error") + def _render_inline(self, sess: _ServerSession, etype: str, props: dict) -> None: + """Append one human-readable progress line to the run log for the event + types worth watching live. No-ops on anything else, so an unhandled + frame writes nothing. + + ``log_fh`` is the readable transcript sink (``.log``): it + carries ONLY these curated ``[bmad]`` lines. The server's own INFO / + diagnostic stdout is kept apart in ``.server.out`` (see + ``server_fh``) so the transcript stays a clean, line-wrapped + conversation rather than a jumble of server logging. Only this SSE + reader thread writes ``log_fh``, and each line is a single ``write()`` + plus ``flush()`` — so it always lands as a whole line at EOF. + + Every event-type string below, and the payload shape each branch reads, + is pinned in the module docstring's ``/event`` section against a live + 1.18.2 probe — read that before adding or changing a branch here. That + is deliberately the only copy of those facts: three of this renderer's + original branches named frames the server does not send, and a second + copy is a second thing to get wrong on the next version bump. + + ``message.updated`` carries only turn metadata (notably ``info.role`` + keyed by ``info.id``) and renders no inline line of its own — it just + records the role under the message id so the following + ``message.part.*`` lines resolve the right speaker. Keying by message + id (not "last role seen") is required because ``message.updated`` + frames arrive out of order: a re-emit for the user message lands mid- + assistant-turn and would otherwise flip the prefix to ``user:`` on the + assistant's own reply text. That also avoids a flood of role-only + announce lines like ``[bmad] assistant: assistant``.""" + if etype == "message.updated": + info = props.get("info") or {} + mid = info.get("id") + role = info.get("role") + if mid and role: + sess.msg_roles[mid] = role + return + role = "assistant" + if etype == "message.part.updated": + mid = (props.get("part") or {}).get("messageID") + if mid and mid in sess.msg_roles: + role = sess.msg_roles[mid] + line = self._inline_line(etype, props, role) + if line is None: + return + if sess.log_fh is None: # bare-sess unit tests / disabled inline log + return + try: + # CRLF line endings: the bmad-loop TUI renders this log through a + # pyte terminal emulator, where a bare LF moves the cursor down + # without returning to column 0 (a real PTY's ONLCR does that, but + # pyte is a pure VT100 emulator). LF-only lines staircase right, + # so emit CRLF like real terminal output (the claude/tmux captures + # carry it via cursor-addressing; our plain text must carry it raw). + sess.log_fh.write(line.replace("\n", "\r\n").encode("utf-8")) + sess.log_fh.flush() + except OSError: + pass + + def _inline_line(self, etype: str, props: dict, role: str) -> str | None: + # ``message.updated`` is consumed in ``_render_inline`` (role refresh + # only); it renders no line here. + # assistant / user assembled text. Per-token ``message.part.delta`` + # frames are NOT rendered inline: they concatenate to exactly the text + # ``message.part.updated`` already carries complete, so emitting both + # duplicates every statement as a one-word-per-line flood. For the same + # reason they are excluded from the SSE trace too (see ``_emit_event``); + # nothing consumes deltas anywhere in this adapter. + if etype == "message.part.updated": + part = props.get("part") or {} + # Tool execution rides this same frame as a `tool`-typed part — + # there is no tool.call/tool.response event on this surface, and + # `command.executed` never fires for it (both pinned live). Tool + # parts carry no `text`, so this branch MUST sit above the + # empty-text `return None` below or it can never be reached: that + # exact ordering is the difference between rendering the agent's + # actions and rendering nothing but its prose. + if part.get("type") == "tool": + return self._tool_line(part) + text = part.get("text") or "" + body = text.strip() # drop surrounding blanks; keep internal structure + if not body: # empty/non-text part — no value live + return None + role = role or "assistant" + # A blank line separates each turn from the previous one (the + # separator sits ABOVE the header, body follows directly beneath), + # and the role-colored marker line anchors the turn boundary. + # Internal newlines in the body are preserved so multi-paragraph + # reasoning reads as prose under the one header. + return f"\n{_role_color(role)}[bmad] {role}:{_RESET}\n{body}\n" + # The SLASH-COMMAND surface — NOT the tool surface. Agent tool use never + # reaches here (0 `command.executed` frames across live bash/write/read/ + # edit calls); it renders from the `tool`-typed part above. Kept because + # a slash command invoked through this server is still worth a line and + # its payload is pinned (`name`/`arguments`/`messageID`, all required), + # but do not "fix" tool rendering back into this branch. + if etype == "command.executed": + args = _sum_args(props.get("arguments")) + return f"{_TOOL_COLOR}[bmad] cmd: {props.get('name') or '?'}{(' ' + args) if args else ''}{_RESET}\n" + if etype == "file.edited": + # Session-less: reachable only via `_SESSIONLESS_TYPES`. + return f"{_TOOL_COLOR}[bmad] file: {props.get('file') or '?'}{_RESET}\n" + # Permission surface. Field names are the live 1.18.2 ones: the ask frame + # is `permission.asked` (there is no `permission.updated`) carrying a + # `permission` STRING plus a `patterns` list — not a nested object with + # `type`/`pattern` — and the reply carries `reply`, not `response`. + # `metadata` (the concrete command) and `always` are left to the trace; + # `patterns` is what the permission actually matched on. + if etype == "permission.asked": + pats = _sum_args(props.get("patterns")) + return ( + f"{_TOOL_COLOR}[bmad] perm ask: {props.get('permission') or '?'}" + f"{(' ' + pats) if pats else ''}{_RESET}\n" + ) + if etype == "permission.replied": + return f"{_TOOL_COLOR}[bmad] perm reply: {props.get('reply') or '?'}{_RESET}\n" + # A failing turn is the one thing a reader most wants in the transcript: + # without this the log just stops mid-turn while the queue put below + # ends the session. Same green/tool hue as the other role-less markers — + # the color family is "not a speaker", not "severity". `error` has no + # pinned shape on this surface, so summarize whatever it carries rather + # than reaching into fields that may not exist. + if etype == "session.error": + detail = _sum_args(props.get("error")) + return f"{_TOOL_COLOR}[bmad] error:{(' ' + detail) if detail else ''}{_RESET}\n" + return None + + @staticmethod + def _tool_line(part: dict) -> str | None: + """One line for a finished tool call, or None while it is still running. + + A `tool`-typed ``message.part.updated`` part fires once per state + transition — ``pending`` → ``running`` → terminal — carrying the SAME + part id each time. Rendering every fire would print each tool call three + times, so the line is emitted only on a terminal status + (``_TOOL_TERMINAL_STATES``): a tool call reaches exactly one terminal + state, which makes that gate exactly "once per tool call". The gate is + the whole correctness argument here — a test that only asserts a tool + renders at all passes with it removed. + + Same green/tool hue as the other role-less markers: a tool call is an + action, not a speaker. ``state.input`` is summarized inline (that is the + "what did the agent just do" the reader wants); ``state.output`` is + deliberately NOT — it is a full command's stdout or a whole file's text + on this surface, and the complete payload is in the SSE trace for anyone + who needs it. A failure appends the error, because a tool that failed + silently is exactly what a reader would go looking for.""" + state = part.get("state") or {} + status = state.get("status") + if status not in _TOOL_TERMINAL_STATES: + return None + args = _sum_args(state.get("input")) + suffix = "" + if status == "error": + detail = _sum_args(state.get("error")) + suffix = f" -> error{(': ' + detail) if detail else ''}" + return ( + f"{_TOOL_COLOR}[bmad] tool: {part.get('tool') or '?'}" + f"{(' ' + args) if args else ''}{suffix}{_RESET}\n" + ) + + def _emit_event(self, sess: _ServerSession, etype: str, props: dict) -> None: + """Append one structured JSON object to the session's + ``.sse.jsonl`` sink — a record for every acted-on frame except + the per-token deltas, for post-hoc replay/debugging. No-ops when the + sink is off (``event_fh is None``, i.e. the ``sse_trace`` knob is down). + + ``message.part.delta`` is the one exclusion. Its text concatenates + byte-exactly to the ``message.part.updated`` record already in the file + (pinned live — see the module docstring), so every delta is bytes spent + re-storing text the trace already holds, and they dominate the frame + count on any real turn. ``logs/`` is never trimmed by retention, so that + cost is permanent. Drop the branch to get them back.""" + if sess.event_fh is None: + return + if etype == "message.part.delta": + return + sess.event_seq += 1 + record = { + "seq": sess.event_seq, + "type": etype, + "properties": props, + "ts": _now_ms(), # epoch ms — the unit every opencode time.* uses + } + try: + sess.event_fh.write(json.dumps(record, ensure_ascii=False) + "\n") + sess.event_fh.flush() + except (OSError, TypeError, ValueError): + pass + # ------------------------------------------------------ completion loop def _result_json(self, handle: SessionHandle, spec: SessionSpec, *, wait: bool) -> dict | None: @@ -991,6 +1432,16 @@ def _teardown(self, sess: _ServerSession) -> None: sess.log_fh.close() except OSError: pass + if sess.server_fh is not None: + try: + sess.server_fh.close() + except OSError: + pass + if sess.event_fh is not None: + try: + sess.event_fh.close() + except OSError: + pass def _kill_process(self, sess: _ServerSession) -> None: process = sess.process diff --git a/tests/test_opencode_http.py b/tests/test_opencode_http.py index 52d8688..2d3760f 100644 --- a/tests/test_opencode_http.py +++ b/tests/test_opencode_http.py @@ -14,6 +14,7 @@ import json import os import queue +import re import signal import subprocess import sys @@ -27,13 +28,17 @@ from bmad_loop.adapters.base import SessionHandle, SessionSpec from bmad_loop.adapters.generic import BUDGET_NUDGE_TEXT, NUDGE_TEXT, STALL_NUDGE_TEXT from bmad_loop.adapters.opencode_http import ( + _RESET, + _TOOL_COLOR, OpencodeDevAdapter, OpencodeHttpAdapter, OpencodeServerError, _free_port, _now_ms, _parse_sse_lines, + _role_color, _ServerSession, + _sum_args, _sum_usage, ) from bmad_loop.adapters.profile import get_profile @@ -91,6 +96,7 @@ assert argv and argv[0] == "serve", argv PORT = int(argv[argv.index("--port") + 1]) HOST = argv[argv.index("--hostname") + 1] +print(f"FAKE_STDOUT_CANARY listening on {HOST}:{PORT}", flush=True) if START_FAILURES: counter = os.path.join(REC_DIR, "start-count") @@ -478,6 +484,757 @@ def test_sse_dispatch_filters_child_sessions(tmp_path): assert sess.events.get_nowait() == "error" +# ------------------------------- readable run logs + structured event JSONL +# +# The opencode-http adapter keeps three on-disk sinks per session: +# .log — the readable transcript: curated one-line human +# progress via _render_inline ([bmad] lines only). +# .server.out — the spawned server's own stdout/stderr (INFO / +# diagnostic lines), kept apart so .log +# reads as a clean conversation, not a jumble. +# .sse.jsonl — one JSON record per acted-on frame except the +# per-token deltas: the trace for post-hoc replay +# (_emit_event), gated by the sse_trace knob. +# The two text sinks (transcript + server stdout) are written from disjoint +# sources (SSE reader thread vs the server process); the JSONL is written from +# the SSE reader thread. These are pure unit tests: canned event dicts straight +# through _dispatch_sse, no real or fake opencode binary (the zero-token +# invariant from PR #167). + + +def _sess_with_sinks(tmp_path: Path, task_id: str = "t-log"): + """A _ServerSession wired to real .log (binary append) and .sse.jsonl + text sinks under tmp_path, so dispatch behaviour is assertable file-to-file. + Mirrors how _spawn_server wires the real sinks (O_APPEND log + append jsonl).""" + adapter = make_adapter(tmp_path) + log_path = tmp_path / f"{task_id}.log" + event_path = tmp_path / f"{task_id}.sse.jsonl" + sess = _ServerSession( + process=None, + port=0, + base_url="", + password="", + log_fh=log_path.open("ab"), + event_fh=event_path.open("a", encoding="utf-8"), + ) + sess.session_id = "ses_test" + return adapter, sess, log_path, event_path + + +_ANSI_RE = re.compile(r"\x1b\[[0-9;]*m") + + +def _log_text(path: Path) -> str: + # The transcript log now carries CRLF line endings (the TUI renders it + # through a pyte terminal emulator); normalize to LF so assertions check + # logical content, not the wire encoding. ANSI color escapes (the cyan + # [bmad] marker) are stripped too so content assertions stay color-agnostic. + if not path.is_file(): + return "" + return _ANSI_RE.sub("", path.read_bytes().decode("utf-8").replace("\r\n", "\n")) + + +def test_inline_line_returns_none_for_unhandled_types(): + """_inline_line is the curate/no-op switch: None for anything not worth a + live line, so those frames leave the run log byte-identical to today.""" + adapter = make_adapter(Path("/tmp")) # no session built; _inline_line is pure + assert adapter._inline_line("server.heartbeat", {}, "assistant") is None + assert adapter._inline_line("server.connected", {}, "assistant") is None + assert adapter._inline_line("session.created", {"sessionID": "x"}, "assistant") is None + assert adapter._inline_line("session.idle", {"sessionID": "x"}, "assistant") is None + assert adapter._inline_line("catalog.updated", {}, "assistant") is None + assert adapter._inline_line("totally.unknown", {"sessionID": "x"}, "assistant") is None + # message.updated renders no line (role refresh lives in _render_inline) + assert ( + adapter._inline_line("message.updated", {"info": {"role": "assistant"}}, "assistant") + is None + ) + # empty/whitespace text carries no live value → None + assert ( + adapter._inline_line("message.part.updated", {"part": {"text": " "}}, "assistant") is None + ) + assert adapter._inline_line("message.part.updated", {"part": {}}, "assistant") is None + # message.part.delta never renders inline — it duplicates the complete + # text message.part.updated already carries (and is kept out of the trace + # for the same reason) + assert adapter._inline_line("message.part.delta", {"delta": "wor"}, "assistant") is None + assert adapter._inline_line("message.part.delta", {"delta": ""}, "assistant") is None + # permission.updated does not exist on this surface at all (the live ask + # frame is permission.asked) — nothing may be keyed to the dead name. + assert ( + adapter._inline_line( + "permission.updated", {"permission": {"type": "edit", "pattern": "src/**"}}, "assistant" + ) + is None + ) + + +def test_inline_line_renders_each_curated_type(): + """Every event type worth a live line renders deterministically. The type + strings and which of them actually fire on 1.18.2 are pinned in the adapter's + module docstring (live probes 2026-07-16 / 2026-07-25) — notably there is no + tool.call / tool.response on this SSE surface at all. The message.part.* + prefix is the role recorded for that part's message id (tracked in + _render_inline from message.updated.info.role).""" + adapter = make_adapter(Path("/tmp")) + assert adapter._inline_line("message.part.updated", {"part": {"text": "hi"}}, "assistant") == ( + f"\n{_role_color('assistant')}[bmad] assistant:{_RESET}\nhi\n" + ) + assert adapter._inline_line("message.part.updated", {"part": {"text": "hi"}}, "user") == ( + f"\n{_role_color('user')}[bmad] user:{_RESET}\nhi\n" + ) + # roles are distinct hues — the map pins them, so assistant != user colour + assert _role_color("assistant") != _role_color("user") + # an unseen role still renders, falling back to the default hue (no crash) + assert adapter._inline_line("message.part.updated", {"part": {"text": "hi"}}, "system") == ( + f"\n{_role_color('system')}[bmad] system:{_RESET}\nhi\n" + ) + # multi-line part.text: blank line above the role-colored header, then the + # full body (internal newlines preserved) directly beneath — reads as a + # contiguous block, not a prefix on every line. + assert ( + adapter._inline_line( + "message.part.updated", + {"part": {"text": "para one\n\npara two\npara three"}}, + "assistant", + ) + == f"\n{_role_color('assistant')}[bmad] assistant:{_RESET}\npara one\n\npara two\npara three\n" + ) + assert adapter._inline_line( + "command.executed", {"name": "Read", "arguments": None}, "assistant" + ) == (f"{_TOOL_COLOR}[bmad] cmd: Read{_RESET}\n") + assert adapter._inline_line("file.edited", {"file": "src/app.py"}, "assistant") == ( + f"{_TOOL_COLOR}[bmad] file: src/app.py{_RESET}\n" + ) + # Permission fields are the live 1.18.2 ones: the ask frame is + # permission.asked (permission.updated does not exist) carrying a + # `permission` STRING plus a `patterns` list, and the reply carries `reply`. + assert ( + adapter._inline_line( + "permission.asked", + { + "id": "per_1", + "permission": "bash", + "patterns": ["echo permcheck"], + "metadata": {"command": "echo permcheck"}, + "always": ["echo *"], + }, + "assistant", + ) + == f'{_TOOL_COLOR}[bmad] perm ask: bash ["echo permcheck"]{_RESET}\n' + ) + # no patterns at all still names the permission rather than printing '?' + assert adapter._inline_line("permission.asked", {"permission": "edit"}, "assistant") == ( + f"{_TOOL_COLOR}[bmad] perm ask: edit{_RESET}\n" + ) + assert adapter._inline_line( + "permission.replied", {"requestID": "per_1", "reply": "once"}, "assistant" + ) == (f"{_TOOL_COLOR}[bmad] perm reply: once{_RESET}\n") + + +def test_inline_line_renders_session_error(): + """session.error ends the session via the queue put in _dispatch_sse; without + a line here the transcript would simply stop mid-turn with no stated cause. + Role-less like the other tool-family markers, so it takes the same hue. The + error payload has no pinned shape on this surface, so any shape summarizes + rather than raising or printing a bare '?'.""" + adapter = make_adapter(Path("/tmp")) + assert adapter._inline_line( + "session.error", {"sessionID": "x", "error": {"name": "ProviderAuthError"}}, "assistant" + ) == (f'{_TOOL_COLOR}[bmad] error: {{"name": "ProviderAuthError"}}{_RESET}\n') + # a bare string payload + assert adapter._inline_line("session.error", {"error": "boom"}, "assistant") == ( + f"{_TOOL_COLOR}[bmad] error: boom{_RESET}\n" + ) + # no detail at all still marks the failure — the line is the signal + assert adapter._inline_line("session.error", {"sessionID": "x"}, "assistant") == ( + f"{_TOOL_COLOR}[bmad] error:{_RESET}\n" + ) + # long payloads are truncated like any other one-line summary + long_line = adapter._inline_line("session.error", {"error": "x" * 500}, "assistant") + assert long_line is not None and "…" in long_line + + +def test_dispatch_session_error_renders_line_and_still_queues_error(tmp_path): + """The inline line is additive: the control-path 'error' put still happens.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse( + sess, + {"type": "session.error", "properties": {"sessionID": "ses_test", "error": "boom"}}, + ) + assert "[bmad] error: boom\n" in _log_text(log_path) + assert sess.events.get_nowait() == "error" + assert [r["type"] for r in read_jsonl(event_path)] == ["session.error"] + + +def test_dispatch_writes_assistant_text_to_log_and_jsonl(tmp_path): + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse( + sess, + { + "type": "message.part.updated", + "properties": {"sessionID": "ses_test", "part": {"text": "hello"}}, + }, + ) + assert "\n[bmad] assistant:\nhello\n" in _log_text(log_path) + # raw bytes still carry the assistant-role SGR (the write path does not strip it) + assert _role_color("assistant") in log_path.read_bytes().decode("utf-8") + records = read_jsonl(event_path) + assert len(records) == 1 + assert records[0]["type"] == "message.part.updated" + assert records[0]["properties"]["part"]["text"] == "hello" + + +def test_message_updated_keys_role_by_message_id(tmp_path): + """``message.updated`` carries ``info.role`` under ``info.id`` but renders no + line of its own; it records the role keyed by opencode message id so the + following ``message.part.updated`` line prefixes with the right speaker + (user vs assistant).""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + # a user turn: message.updated(role=user, id=u1) then the user's prompt text + adapter._dispatch_sse( + sess, + { + "type": "message.updated", + "properties": {"sessionID": "ses_test", "info": {"id": "msg_u1", "role": "user"}}, + }, + ) + adapter._dispatch_sse( + sess, + { + "type": "message.part.updated", + "properties": { + "sessionID": "ses_test", + "part": {"text": "do the thing", "messageID": "msg_u1"}, + }, + }, + ) + assert sess.msg_roles["msg_u1"] == "user" + log = _log_text(log_path) + assert "\n[bmad] user:\ndo the thing\n" in log + assert "assistant: do the thing" not in log + # message.updated itself renders no inline line, but IS in the JSONL trace + types = [r["type"] for r in read_jsonl(event_path)] + assert types == ["message.updated", "message.part.updated"] + + +def test_role_lookup_survives_out_of_order_message_updated(tmp_path): + """``message.updated`` frames re-emit out of order: a stale user re-emit can + land mid-assistant-turn. Keying role by message id (not "last seen") keeps + the assistant's reply labeled ``assistant:`` even when a user re-emit + arrives between the assistant's announcement and its reply text.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + # assistant announced + adapter._dispatch_sse( + sess, + { + "type": "message.updated", + "properties": {"sessionID": "ses_test", "info": {"id": "msg_a1", "role": "assistant"}}, + }, + ) + # stale re-emit for the earlier user message (must NOT clobber the assistant) + adapter._dispatch_sse( + sess, + { + "type": "message.updated", + "properties": {"sessionID": "ses_test", "info": {"id": "msg_u1", "role": "user"}}, + }, + ) + # assistant's reply part arrives — must still be labeled assistant: + adapter._dispatch_sse( + sess, + { + "type": "message.part.updated", + "properties": { + "sessionID": "ses_test", + "part": {"text": "done", "messageID": "msg_a1"}, + }, + }, + ) + log = _log_text(log_path) + assert "\n[bmad] assistant:\ndone\n" in log + assert "user: done" not in log + + +def test_part_without_known_message_id_falls_back_to_assistant(tmp_path): + """A part whose message id has no recorded role falls back to ``assistant`` + so the line still renders rather than dropping.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse( + sess, + { + "type": "message.part.updated", + "properties": { + "sessionID": "ses_test", + "part": {"text": "hi", "messageID": "msg_unseen"}, + }, + }, + ) + assert "\n[bmad] assistant:\nhi\n" in _log_text(log_path) + + +def test_dispatch_skips_empty_text_and_drops_deltas_from_both_sinks(tmp_path): + """Empty/whitespace part text carries no live value, so it renders nothing + while still being traced. message.part.delta is dropped from BOTH sinks: its + tokens concatenate byte-exactly to the text message.part.updated already + carries complete (pinned live), so tracing them re-stores text the file + already holds — and logs/ is never trimmed by retention, so those bytes are + permanent. Deltas also must not consume a seq: the trace's seq numbering has + to match the lines actually in the file.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + for evt in ( + { + "type": "message.part.updated", + "properties": {"sessionID": "ses_test", "part": {"text": ""}}, + }, + { + "type": "message.part.updated", + "properties": {"sessionID": "ses_test", "part": {"text": " "}}, + }, + { + "type": "message.part.delta", + "properties": {"sessionID": "ses_test", "delta": "streaming token"}, + }, + ): + adapter._dispatch_sse(sess, evt) + assert _log_text(log_path) == "" + records = read_jsonl(event_path) + assert [r["type"] for r in records] == ["message.part.updated", "message.part.updated"] + assert [r["seq"] for r in records] == [1, 2] + assert sess.event_seq == 2 # the delta burned no sequence number + + +def test_dispatch_writes_command_file_permission_lines(tmp_path): + """The role-less marker family, dispatched with the payload shapes 1.18.2 + actually sends: command.executed carries a sessionID, file.edited carries + none (it rides the session-less allowlist), and the permission pair uses + permission/patterns + reply.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + for evt in ( + { + "type": "command.executed", + "properties": {"sessionID": "ses_test", "name": "Edit", "arguments": {"x": 1}}, + }, + {"type": "file.edited", "properties": {"file": "a.py"}}, + { + "type": "permission.asked", + "properties": {"sessionID": "ses_test", "permission": "bash", "patterns": ["rm *"]}, + }, + { + "type": "permission.replied", + "properties": {"sessionID": "ses_test", "requestID": "per_1", "reply": "deny"}, + }, + ): + adapter._dispatch_sse(sess, evt) + log = _log_text(log_path) + assert "[bmad] cmd: Edit " in log + assert "[bmad] file: a.py\n" in log + assert '[bmad] perm ask: bash ["rm *"]\n' in log + assert "[bmad] perm reply: deny\n" in log + types = [r["type"] for r in read_jsonl(event_path)] + assert types == ["command.executed", "file.edited", "permission.asked", "permission.replied"] + + +# ------------------------------------------------------------ tool-part render +# +# Agent tool use has no event of its own on this SSE surface: no tool.call, no +# tool.response, and command.executed never fires for it (0 frames across live +# bash/write/read/edit). It arrives as message.part.updated with a `tool`-typed +# part — which carries NO `part.text`, so the tool branch has to sit above +# _inline_line's empty-text bail-out to be reachable at all. +# +# A tool part re-fires on every state transition with the same part id +# (pending → running → terminal), so the render is gated on a TERMINAL status. +# That gate, not the rendering, is the correctness claim here — see the ABLATION +# note above test_tool_part_renders_once_per_call. + + +def _tool_part(status: str, **state) -> dict: + """A `tool`-typed message part in the given state, shaped like 1.18.2's + ToolPart (same part id across every transition of one call).""" + return { + "id": "prt_tool1", + "sessionID": "ses_test", + "messageID": "msg_a1", + "type": "tool", + "callID": "toolu_1", + "tool": "bash", + "state": {"status": status, **state}, + } + + +def _dispatch_tool(adapter, sess, part: dict) -> None: + adapter._dispatch_sse( + sess, + {"type": "message.part.updated", "properties": {"sessionID": "ses_test", "part": part}}, + ) + + +def test_tool_line_renders_name_and_input_summary(): + """The terminal fire renders the tool name plus a one-line input summary — + the "what did the agent just do" the transcript exists for. state.output is + deliberately absent from the line (it is a whole command's stdout or a whole + file on this surface); the complete payload stays in the SSE trace.""" + adapter = make_adapter(Path("/tmp")) # _inline_line is pure + line = adapter._inline_line( + "message.part.updated", + {"part": _tool_part("completed", input={"command": "echo hi"}, output="OUTPUT_MARKER")}, + "assistant", + ) + assert line == f'{_TOOL_COLOR}[bmad] tool: bash {{"command": "echo hi"}}{_RESET}\n' + assert "OUTPUT_MARKER" not in line # output is traced, never rendered inline + # a tool with no input still names itself rather than dropping + assert adapter._inline_line( + "message.part.updated", {"part": _tool_part("completed", input={}, output="")}, "assistant" + ) == (f"{_TOOL_COLOR}[bmad] tool: bash{_RESET}\n") + # long inputs truncate like any other one-line summary + long_line = adapter._inline_line( + "message.part.updated", + {"part": _tool_part("completed", input={"command": "x" * 500})}, + "assistant", + ) + assert long_line is not None and "…" in long_line + + +def test_tool_line_marks_a_failed_call(): + """`error` is the other terminal status (ToolState is a 4-variant union), and + a tool that failed is precisely what a reader goes to the transcript for — + so it renders, with the error appended to the same line shape.""" + adapter = make_adapter(Path("/tmp")) + assert adapter._inline_line( + "message.part.updated", + {"part": _tool_part("error", input={"command": "false"}, error="exit status 1")}, + "assistant", + ) == ( + f'{_TOOL_COLOR}[bmad] tool: bash {{"command": "false"}} -> error: exit status 1{_RESET}\n' + ) + # an error with no detail still marks the failure — the marker is the signal + assert adapter._inline_line( + "message.part.updated", {"part": _tool_part("error", input={})}, "assistant" + ) == (f"{_TOOL_COLOR}[bmad] tool: bash -> error{_RESET}\n") + + +def test_tool_part_renders_to_the_log(tmp_path): + """End to end through _dispatch_sse: a completed tool call lands in the + readable transcript (this is the half of the feature that rendered nothing + at all before — tool parts have no part.text).""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + _dispatch_tool(adapter, sess, _tool_part("completed", input={"file": "a.py"}, output="ok")) + assert '[bmad] tool: bash {"file": "a.py"}\n' in _log_text(log_path) + # role-less: no "[bmad] assistant:" header is emitted for a tool part + assert "[bmad] assistant:" not in _log_text(log_path) + assert _TOOL_COLOR in log_path.read_bytes().decode("utf-8") + + +# ABLATION (both directions run 2026-07-25). "One line per tool" is a PLACEMENT +# claim, and it is also what a renderer that emits NOTHING satisfies — so the +# presence ablation alone would not test the gate: +# (a) delete the `part.get("type") == "tool"` branch from _inline_line → all +# five tool tests fail; this one at `assert 0 == 1` on the count. +# (b) INVERSE: drop the `status not in _TOOL_TERMINAL_STATES` gate from +# _tool_line so it renders on every transition → this test fails on the +# FIRST intermediate assert (pending rendered '[bmad] tool: bash'), which +# is the proof the mutation actually fires: the pending and running frames +# really do reach the branch, so the count assertion is not passing +# vacuously. Left to run to completion the count is 3 != 1, with running +# and completed rendering byte-identical lines. +# The intermediate "nothing yet" asserts are the placement claim stated directly; +# the count is the backstop. + + +def test_tool_part_renders_once_per_call(tmp_path): + """One tool call = one transcript line, even though the part re-fires on + pending, running and the terminal status with the same part id. The trace + keeps all three fires: the sinks split curated-vs-complete by design.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + # the live progression: pending (no input yet) → running → completed + _dispatch_tool(adapter, sess, _tool_part("pending", input={}, raw="")) + assert _log_text(log_path) == "" # nothing rendered while it is queued + _dispatch_tool(adapter, sess, _tool_part("running", input={"command": "echo hi"})) + assert _log_text(log_path) == "" # nor while it runs + _dispatch_tool( + adapter, sess, _tool_part("completed", input={"command": "echo hi"}, output="hi\n") + ) + + log = _log_text(log_path) # ANSI-stripped, CRLF normalized + assert log.count("[bmad] tool: bash") == 1 + assert log == '[bmad] tool: bash {"command": "echo hi"}\n' + # all three fires are in the trace — only the transcript is deduplicated + records = read_jsonl(event_path) + assert [r["properties"]["part"]["state"]["status"] for r in records] == [ + "pending", + "running", + "completed", + ] + + +def test_tool_part_never_shadows_assistant_text(tmp_path): + """The tool branch sits inside the message.part.updated arm, above the + empty-text bail-out — it must not swallow ordinary text parts on the way.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + for part in ( + {"type": "text", "text": "thinking about it", "messageID": "msg_a1"}, + _tool_part("completed", input={"command": "ls"}, output="a.py"), + {"type": "text", "text": "done", "messageID": "msg_a1"}, + ): + _dispatch_tool(adapter, sess, part) + log = _log_text(log_path) + assert "\n[bmad] assistant:\nthinking about it\n" in log + assert '[bmad] tool: bash {"command": "ls"}\n' in log + assert "\n[bmad] assistant:\ndone\n" in log + assert log.index("thinking about it") < log.index("tool: bash") < log.index("done") + + +def test_jsonl_seq_is_monotonic_with_ts_and_shape(tmp_path): + """Each traced event gets one JSONL record with a monotonic seq, the raw + type, the raw properties, and an epoch-ms ts that never goes backwards + across the batch.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + for n in range(4): + adapter._dispatch_sse( + sess, + {"type": "session.diff", "properties": {"sessionID": "ses_test", "diff": [str(n)]}}, + ) + records = read_jsonl(event_path) + assert [r["seq"] for r in records] == [1, 2, 3, 4] + assert all(set(r) == {"seq", "type", "properties", "ts"} for r in records) + ts = [r["ts"] for r in records] + assert ts == sorted(ts) # epoch ms, non-decreasing within the batch + # epoch ms, not ns and not seconds — the unit every opencode time.* field + # uses, and the unit the poll fallback's floor comparison assumes + assert all(abs(r["ts"] - _now_ms()) < 60_000 for r in records) + assert all(r["properties"]["diff"] == [str(i)] for i, r in enumerate(records)) + + +def test_unhandled_session_event_recorded_in_jsonl_not_log(tmp_path): + """A session-scoped event _render_inline does not curate (e.g. session.diff) + still passes the sessionID filter, so it lands in the complete-trace JSONL — + but writes no human-readable line. The two sinks split by design: curated + human log vs complete machine trace.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse( + sess, {"type": "session.diff", "properties": {"sessionID": "ses_test", "diff": []}} + ) + assert _log_text(log_path) == "" + records = read_jsonl(event_path) + assert len(records) == 1 and records[0]["type"] == "session.diff" + + +def test_no_session_id_event_filtered_from_both_sinks(tmp_path): + """An event with no matching sessionID (noise like catalog.updated) is + filtered before either sink fires — neither a log line nor a JSONL record.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse(sess, {"type": "catalog.updated", "properties": {}}) + assert _log_text(log_path) == "" + assert read_jsonl(event_path) == [] + assert sess.event_seq == 0 + + +# ----------------------------------------------- session-less type allowlist +# +# Some frames worth logging carry no sessionID at all: file.edited is exactly +# {"file": "/abs/path"} live, and 1.18.2 pins it additionalProperties:false over +# that one key — so the sessionID filter in _dispatch_sse drops it above both +# sinks and its render branch is unreachable without an explicit exemption. +# _SESSIONLESS_TYPES is that exemption, and it is sound ONLY because bmad-loop +# spawns one opencode serve per session, which makes a server-global frame +# unambiguously this session's. +# +# It is an allowlist rather than "allow anything session-less" because the +# session-less traffic is mostly noise that says nothing about the run: +# plugin.added alone was 90 of 388 frames in the live probe. +# +# ABLATION (verified 2026-07-25): revert _dispatch_sse's filter to the plain +# `props.get("sessionID") != sess.session_id` and the two tests that assert a +# session-less frame REACHES a sink fail. The two negative tests below +# (non-allowlisted, control-queue) keep passing under that ablation — they +# assert absence, which holds however the frame was dropped, so they document +# scope rather than prove reachability. The positive test carries that claim. + + +def test_sessionless_allowlisted_type_renders_and_traces(tmp_path): + """file.edited has no sessionID at all, yet reaches BOTH sinks — the + allowlist exempts it from the filter, and the trace stays consistent with the + transcript (it is the file you read to explain a rendered line).""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse(sess, {"type": "file.edited", "properties": {"file": "/abs/src/app.py"}}) + assert "[bmad] file: /abs/src/app.py\n" in _log_text(log_path) + records = read_jsonl(event_path) + assert [r["type"] for r in records] == ["file.edited"] + assert records[0]["properties"] == {"file": "/abs/src/app.py"} + + +def test_sessionless_non_allowlisted_type_writes_nothing(tmp_path): + """The allowlist is not a blanket exemption. plugin.added is session-less + too and was 90 of 388 live frames; it must still be dropped above both + sinks, burning no seq. Same for the file watcher, which fires for any change + under the project (not just the agent's) and double-reports file.edited.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + for etype in ("plugin.added", "file.watcher.updated", "server.instance.disposed"): + adapter._dispatch_sse(sess, {"type": etype, "properties": {"plugin": "x", "file": "y"}}) + assert _log_text(log_path) == "" + assert read_jsonl(event_path) == [] + assert sess.event_seq == 0 + # liveness is unaffected: these frames still counted as activity, exactly as + # before the allowlist existed (that counter sits above the filter) + assert sess.activity == 3 + + +def test_sessionless_allowlist_does_not_touch_the_control_queue(tmp_path): + """The allowlist is a logging exemption only. A session-less frame must never + reach the idle/error puts — those stay keyed to this session's id, or a + foreign server-global frame could end the turn.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + for etype in ("session.idle", "session.error"): + adapter._dispatch_sse(sess, {"type": etype, "properties": {}}) + assert sess.events.empty() + # and an allowlisted one does not either + adapter._dispatch_sse(sess, {"type": "file.edited", "properties": {"file": "a.py"}}) + assert sess.events.empty() + + +def test_unhashable_event_type_never_raises_in_the_filter(tmp_path): + """The allowlist membership test sits OUTSIDE the try/except that guards the + two sinks, so it must not raise on a server-controlled `type`: `in` on a + frozenset raises TypeError for an unhashable value, hence the isinstance + narrowing ahead of it.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + for etype in ({"a": 1}, ["file.edited"], {"file.edited"}): + adapter._dispatch_sse(sess, {"type": etype, "properties": {}}) # must not raise + assert _log_text(log_path) == "" + assert read_jsonl(event_path) == [] + # the control path still works after them + adapter._dispatch_sse(sess, {"type": "session.idle", "properties": {"sessionID": "ses_test"}}) + assert sess.events.get_nowait() == "idle" + + +def test_render_inline_noops_when_log_fh_is_none(tmp_path): + """A bare _ServerSession (log_fh=None, as built by the older unit tests) + must not raise when a curated event is dispatched — inline rendering is + best-effort, never a crash path.""" + adapter = make_adapter(tmp_path) + sess = _ServerSession(process=None, port=0, base_url="", password="", log_fh=None) + sess.session_id = "ses_test" + adapter._dispatch_sse( + sess, + { + "type": "message.part.updated", + "properties": {"sessionID": "ses_test", "part": {"text": "x"}}, + }, + ) + # event_fh defaults to None too: no jsonl written, no seq bumped + assert sess.event_seq == 0 + + +# --------------------------------------------- logging never breaks the stream +# +# Both sinks are written from _dispatch_sse ABOVE the idle/error queue puts, and +# both walk nested .get chains into server-controlled payloads. A frame that +# ships a string (or a list, or null) where the renderer expects a dict raises +# AttributeError. Unguarded, that propagates out of _dispatch_sse into the SSE +# reader's connection-level `except Exception` in _sse_loop, which treats it as +# a dead connection: it tears the stream down, puts a `gap`, sleeps, and +# reconnects — losing whatever the server had already buffered on that +# connection. That is a logging bug degrading completion signaling, so these +# tests assert the two properties that matter: the malformed frame does not +# raise, and a following session.idle still reaches the control queue. +# +# ABLATION (verified 2026-07-25): delete the try/except around the two helper +# calls in _dispatch_sse and the five tests below that feed a wrong-shaped +# payload through the RENDER path fail with AttributeError. The sixth +# (test_unserializable_properties_...) still passes ablated — its protection is +# _emit_event's own narrow `except (OSError, TypeError, ValueError)` around +# json.dumps, not this guard. It is kept as a regression test for that inner +# catch; do not read it as evidence for the outer one. + +MALFORMED_FRAMES = [ + # part is a string, not a dict → _render_inline's part.get("messageID") + {"type": "message.part.updated", "properties": {"part": "not-a-dict"}}, + # part is a list → same chain, different wrong shape + {"type": "message.part.updated", "properties": {"part": ["text"]}}, + # info is a string → _render_inline's info.get("id") on the role refresh + {"type": "message.updated", "properties": {"info": "not-a-dict"}}, + # a tool part whose state is a truthy non-dict → `part.get("state") or {}` + # keeps it, then state.get("status") blows up. (The permission frames no + # longer offer this shape: post-correction they read `permission` / + # `patterns` / `reply` straight off props, which is always a dict here.) + {"type": "message.part.updated", "properties": {"part": {"type": "tool", "state": "running"}}}, +] + + +@pytest.mark.parametrize("frame", MALFORMED_FRAMES, ids=lambda f: f["type"]) +def test_malformed_frame_never_breaks_dispatch(tmp_path, frame): + """A server-shaped payload the renderer can't walk must not escape dispatch.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + event = {**frame, "properties": {**frame["properties"], "sessionID": "ses_test"}} + + adapter._dispatch_sse(sess, event) # must not raise + + # and the control path is untouched: the next idle still queues + adapter._dispatch_sse(sess, {"type": "session.idle", "properties": {"sessionID": "ses_test"}}) + assert sess.events.get_nowait() == "idle" + + +def test_malformed_frame_does_not_lose_the_trace_record(tmp_path): + """_emit_event runs before _render_inline inside the guard, so the frame that + broke rendering is still in the trace — which is the file you would read to + diagnose it.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse( + sess, + { + "type": "message.part.updated", + "properties": {"sessionID": "ses_test", "part": "not-a-dict"}, + }, + ) + records = read_jsonl(event_path) + assert [r["type"] for r in records] == ["message.part.updated"] + assert records[0]["properties"]["part"] == "not-a-dict" + assert _log_text(log_path) == "" # rendering aborted, nothing half-written + + +def test_unserializable_properties_never_break_dispatch(tmp_path): + """The trace's own failure mode: json.dumps raising on a payload it can't + encode must not take the stream down either. Held by _emit_event's own + catch rather than the outer guard (see the ABLATION note above).""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse( + sess, + { + "type": "message.part.updated", + "properties": {"sessionID": "ses_test", "part": {"text": "hi"}, "blob": {1, 2}}, + }, + ) + adapter._dispatch_sse(sess, {"type": "session.idle", "properties": {"sessionID": "ses_test"}}) + assert sess.events.get_nowait() == "idle" + # the unserializable record is dropped, but the readable line still landed + # and the trace keeps working for the frames that follow + assert [r["type"] for r in read_jsonl(event_path)] == ["session.idle"] + assert "\n[bmad] assistant:\nhi\n" in _log_text(log_path) + + +def test_non_string_event_type_is_dropped(tmp_path): + """`type` absent or non-string can match no branch below, so the frame is + dropped rather than traced — it is not something the adapter acted on.""" + adapter, sess, log_path, event_path = _sess_with_sinks(tmp_path) + adapter._dispatch_sse(sess, {"properties": {"sessionID": "ses_test"}}) + adapter._dispatch_sse(sess, {"type": 7, "properties": {"sessionID": "ses_test"}}) + assert read_jsonl(event_path) == [] + assert _log_text(log_path) == "" + assert sess.events.empty() + + +def test_sum_args_truncates_and_handles_shapes(): + """_sum_args keeps a command's arguments on one log line: short passthrough, + dict→json, long→truncated with an ellipsis, falsy→empty.""" + assert _sum_args(None) == "" + assert _sum_args("") == "" + assert _sum_args({"a": 1}) == '{"a": 1}' + assert _sum_args([1, 2]) == "[1, 2]" + long = _sum_args("x" * 500) + assert len(long) == 120 and long.endswith("\u2026") + + def test_sum_usage_maps_opencode_tokens(): messages = [ {"info": {"role": "user", "tokens": {"input": 999}}}, # not assistant: ignored @@ -744,6 +1501,31 @@ def test_e2e_completed(tmp_path, fake_opencode): assert adapter._sessions == {} assert_server_gone(rec) assert (tmp_path / "run" / "logs" / "t-1.log").exists() + # the server's own stdout is redirected to its own file, keeping the + # readable transcript in .log clean of server INFO noise + server_log = (tmp_path / "run" / "logs" / "t-1.server.out").read_text() + assert "FAKE_STDOUT_CANARY" in server_log + assert "FAKE_STDOUT_CANARY" not in (tmp_path / "run" / "logs" / "t-1.log").read_text() + # the SSE trace is on by default and named .sse.jsonl + trace = read_jsonl(tmp_path / "run" / "logs" / "t-1.sse.jsonl") + assert trace and "session.idle" in {r["type"] for r in trace} + + +def test_e2e_sse_trace_knob_off_never_opens_the_sink(tmp_path, fake_opencode): + """sse_trace is a tier-1 instance knob, not a policy field: flipping it off + means the file is never created at all (not created-then-empty), while the + run itself and the readable transcript are unaffected.""" + launcher, rec = fake_opencode + adapter = make_adapter(tmp_path, binary=str(launcher)) + adapter.sse_trace = False + spec = make_spec(tmp_path, rec, "completed") + + result = adapter.run(spec) + + assert result.status == "completed" + assert not (tmp_path / "run" / "logs" / "t-1.sse.jsonl").exists() + assert (tmp_path / "run" / "logs" / "t-1.log").exists() + assert_server_gone(rec) def test_e2e_result_less_stop_nudges_then_completes(tmp_path, fake_opencode): @@ -1300,15 +2082,95 @@ def test_e2e_spawn_retry_survives_one_early_death(tmp_path, fake_opencode): def test_e2e_spawn_gives_up_after_attempts(tmp_path, fake_opencode): + """The give-up error has to point at the file that actually holds the + diagnostics. Server stdout is redirected to .server.out, so naming + .log — which now carries only the SSE-rendered transcript, and is + EMPTY when the server never got far enough to stream anything — would send + an operator to a blank file on every failed spawn.""" launcher, rec = fake_opencode adapter = make_adapter(tmp_path, binary=str(launcher)) spec = make_spec(tmp_path, rec, "completed", extra_env={"FAKE_OPENCODE_START_FAILURES": "99"}) + logs = tmp_path / "run" / "logs" - with pytest.raises(OpencodeServerError, match="after 3 attempts"): + with pytest.raises(OpencodeServerError, match="after 3 attempts") as excinfo: adapter.run(spec) + assert adapter._sessions == {} - # every attempt appended to one log for the task - assert (tmp_path / "run" / "logs" / "t-1.log").exists() + message = str(excinfo.value) + assert str(logs / "t-1.server.out") in message + assert str(logs / "t-1.log") not in message + # and the path it names holds the goods: every attempt appended to one + # server log, so all 3 spawns' stdout is there to read + server_out = (logs / "t-1.server.out").read_text(encoding="utf-8") + assert server_out.count("FAKE_STDOUT_CANARY") == 3 + # the transcript is the file that would have been useless here + assert (logs / "t-1.log").read_bytes() == b"" + + +def test_spawn_open_failure_closes_the_sinks_already_open(tmp_path, fake_opencode, monkeypatch): + """A failing sink `open()` must not strand the sinks opened before it. + + All three sinks open ABOVE the retry loop's try/except, so an `open()` that + fails partway — ENOSPC, EMFILE, a permission race — propagates straight out + of `_spawn_server` without ever reaching the loop's handler. Only the guard + around the opens themselves can close what was already handed out; without + it those handles stay open for as long as the propagating traceback keeps + this frame alive. One sink had nothing to leak here; three do. + """ + launcher, rec = fake_opencode + adapter = make_adapter(tmp_path, binary=str(launcher)) + spec = make_spec(tmp_path, rec, "completed") + opened: dict = {} + real_open = Path.open + + def flaky_open(self, *args, **kwargs): + # the LAST of the three sinks to open — so .log and .sse.jsonl are both + # already live when it blows up + if self.name.endswith(".server.out"): + raise OSError(28, "No space left on device") + fh = real_open(self, *args, **kwargs) + opened[self.name] = fh + return fh + + monkeypatch.setattr(Path, "open", flaky_open) + + with pytest.raises(OSError, match="No space left"): + adapter._spawn_server(spec) + + # both earlier sinks were opened, and both were closed on the way out + assert set(opened) >= {"t-1.log", "t-1.sse.jsonl"}, sorted(opened) + still_open = [name for name, fh in opened.items() if not fh.closed] + assert still_open == [], f"leaked sink handles: {still_open}" + + +def test_close_spawn_sinks_survives_a_failing_close(tmp_path): + """A close that raises must not displace the failure being reported. + + Both callers are already reporting something worse — one is mid-`raise`, the + other is about to raise `OpencodeServerError` — so an OSError from a + flush-on-close would replace the real diagnosis. It must also not strand the + sinks after it in the loop. + """ + adapter = make_adapter(tmp_path) + closed = [] + + class _Fh: + def __init__(self, name, boom=False): + self.name, self._boom = name, boom + + def close(self): + closed.append(self.name) + if self._boom: + raise OSError(5, "Input/output error") + + # the FIRST sink fails to close: an unguarded loop would propagate here and + # never reach the other two + adapter._close_spawn_sinks(_Fh("log", boom=True), _Fh("server"), _Fh("event")) + + assert closed == ["log", "server", "event"] + # and None (sse_trace off, or the sink whose open was the failure) is skipped + adapter._close_spawn_sinks(_Fh("log2"), None, None) + assert closed[-1] == "log2" # --------------------------------------------- OpencodeDevAdapter (Phase 4)