diff --git a/omnigent/opencode_native_forwarder.py b/omnigent/opencode_native_forwarder.py index ec43b627..195f2e70 100644 --- a/omnigent/opencode_native_forwarder.py +++ b/omnigent/opencode_native_forwarder.py @@ -181,6 +181,14 @@ class OpenCodeNativeForwarder: # the cumulative reasoning text on each ``part.updated``; we forward only # the new suffix so the web reasoning block grows once, not duplicated. self._reasoning_posted: dict[str, int] = {} + # The in-flight turn's assistant messageID (its per-turn ``response_id``), + # captured from ``message.updated`` and stamped on the running/idle status + # edges so the web chat renders this turn's tool calls live — the mirrored + # ``function_call`` items carry the SAME id. ``_running_response_id`` + # records the id the ``running`` edge went out with, gating it to once per + # turn; both reset in :meth:`_end_turn`. + self._active_message_id: str | None = None + self._running_response_id: str | None = None async def seed_dedupe_from_history(self) -> None: """ @@ -342,7 +350,17 @@ class OpenCodeNativeForwarder: return None async def _post_status(self, status: str, *, extra: Mapping[str, Any] | None = None) -> None: - """Publish a coarse session status edge.""" + """Publish a coarse session status edge. + + :param extra: Extra fields merged into the edge payload. On the + ``running``/``idle`` edges this carries ``{"response_id": }``: when it matches the ``response_id`` on this turn's + mirrored ``function_call`` items, the web chat renders the in-flight + tool calls live (spinner + ticking elapsed timer) instead of static + completed cards, and the server tracks it (``active_response_id``) so + a mid-turn reconnect stays live. A ``failed`` edge instead carries + ``output`` / ``reauth_required``. + """ data: dict[str, Any] = {"status": status} if extra: data.update(extra) @@ -420,23 +438,54 @@ class OpenCodeNativeForwarder: ) async def _begin_turn_if_needed(self) -> None: - """Post a single ``running`` status at the start of a turn.""" - if not self.state.turn_active: - self.state.turn_active = True - await self._post_status(_STATUS_RUNNING) + """Emit the turn's id-bearing ``running`` edge once, when the id is known. + + The ``running`` edge carries the assistant ``response_id`` (the opencode + messageID held in ``_active_message_id``) so the web chat can render this + turn's in-flight tool calls live — the mirrored ``function_call`` items + carry the SAME id. It fires once per turn and is deferred until the id is + known: a bare ``session.status`` busy can open the turn before the + assistant ``message.updated`` supplies the id, and emitting an id-less + (session-id-fallback) edge then would never match the tool-call items. + """ + self.state.turn_active = True + if self._running_response_id is None and self._active_message_id is not None: + self._running_response_id = self._active_message_id + await self._post_status( + _STATUS_RUNNING, extra={"response_id": self._running_response_id} + ) async def _end_turn( self, *, status: str = _STATUS_IDLE, extra: Mapping[str, Any] | None = None ) -> None: - """Post the terminal status (idle by default) and clear active state.""" + """Post the terminal status (idle by default), stamped with the turn's id. + + The terminal edge carries the same ``response_id`` the ``running`` edge + used so the server retires this turn's live tool-call cards for the right + response; a caller may pass extra fields (e.g. ``output`` / + ``reauth_required`` on a ``failed`` edge), which are merged on top. + """ self.state.turn_active = False # Reasoning deltas are per-turn; drop the per-part offsets so the map # can't grow across a long-lived session (the next turn's reasoning # parts carry fresh ids anyway). self._reasoning_posted.clear() + # Stamp the terminal edge with the id the ``running`` edge actually went + # out with (``_running_response_id``), then merge any caller-supplied + # fields on top. If a turn produced more than one assistant messageID, + # ``_active_message_id`` has advanced past the id that went live; using + # the running id keeps both edges consistent so the web retires the cards + # that were rendered live. Fall back to the latest assistant id (then the + # session id) when no running edge fired. + terminal_id = self._running_response_id or self._active_message_id + merged_extra: dict[str, Any] = {"response_id": self._response_id(terminal_id)} + if extra: + merged_extra.update(extra) if self._bridge_dir is not None: update_active_message_id(self._bridge_dir, None, status="idle") - await self._post_status(status, extra=extra) + await self._post_status(status, extra=merged_extra) + self._active_message_id = None + self._running_response_id = None # --- per-event handlers ---------------------------------------------- @@ -455,6 +504,10 @@ class OpenCodeNativeForwarder: return self._msg_role[message_id] = role if role == "assistant": + # This turn's per-turn ``response_id`` — the running/idle edges carry + # it so the web chat can correlate them with the tool-call items that + # already stamp the same id (renders in-flight tool calls live). + self._active_message_id = message_id if self._bridge_dir is not None: update_active_message_id(self._bridge_dir, message_id, status="busy") await self._begin_turn_if_needed() diff --git a/tests/test_opencode_native_forwarder.py b/tests/test_opencode_native_forwarder.py index 28684f40..1b753f98 100644 --- a/tests/test_opencode_native_forwarder.py +++ b/tests/test_opencode_native_forwarder.py @@ -292,6 +292,140 @@ async def test_session_error_message_aborted_takes_idle_path() -> None: assert "output" not in status +def _status_edges(posts: list[tuple[str, dict[str, Any]]]) -> list[dict[str, Any]]: + return [b["data"] for _u, b in posts if b["type"] == "external_session_status"] + + +async def test_running_and_idle_carry_assistant_response_id() -> None: + """running/idle edges carry the turn's assistant messageID as ``response_id``. + + The web chat renders in-flight tool calls live only when the ``running`` edge + and the mirrored ``function_call`` items share the SAME ``response_id``. Here + the tool call and both status edges must all group under ``msg_1``. + """ + server, opencode = _RecordingServerClient(), _FakeOpenCodeClient() + fwd = _forwarder(server, opencode) + await fwd.handle_event(_event("message.updated", info={"id": "msg_1", "role": "assistant"})) + await fwd.handle_event( + _event( + "message.part.updated", + part={ + "id": "prt_t", + "messageID": "msg_1", + "type": "tool", + "callID": "call_1", + "tool": "bash", + "state": {"status": "completed", "input": {"command": "ls"}, "output": "ok"}, + }, + ) + ) + await fwd.handle_event(_event("session.idle")) + + edges = _status_edges(server.posts) + assert [(e["status"], e["response_id"]) for e in edges] == [ + ("running", "msg_1"), + ("idle", "msg_1"), + ] + call = next(b for _u, b in server.posts if b["data"].get("item_type") == "function_call") + # The live-card contract: same id on the running edge and the tool call. + assert call["data"]["response_id"] == edges[0]["response_id"] + + +async def test_running_edge_fires_once_per_turn() -> None: + """A turn's many parts still produce exactly one ``running`` edge.""" + server, opencode = _RecordingServerClient(), _FakeOpenCodeClient() + fwd = _forwarder(server, opencode) + await fwd.handle_event(_event("message.updated", info={"id": "msg_1", "role": "assistant"})) + for part in ( + {"id": "s", "messageID": "msg_1", "type": "step-start"}, + {"id": "prt_x", "messageID": "msg_1", "type": "text", "text": "hi"}, + { + "id": "prt_t", + "messageID": "msg_1", + "type": "tool", + "callID": "c1", + "tool": "bash", + "state": {"status": "running", "input": {"command": "ls"}}, + }, + ): + await fwd.handle_event(_event("message.part.updated", part=part)) + running = [e for e in _status_edges(server.posts) if e["status"] == "running"] + assert len(running) == 1 + assert running[0]["response_id"] == "msg_1" + + +async def test_running_edge_deferred_until_message_id_known() -> None: + """A bare ``session.status`` busy before ``message.updated`` still yields the id. + + opencode can open a turn with ``session.status`` busy (no messageID) before + the assistant ``message.updated`` arrives. The ``running`` edge must defer + until the id is known and carry ``msg_1`` — not an id-less/session-id edge + that would never match the tool-call items — and still fire exactly once. + """ + server, opencode = _RecordingServerClient(), _FakeOpenCodeClient() + fwd = _forwarder(server, opencode) + await fwd.handle_event(_event("session.status", status={"type": "busy"})) + # No running edge yet: the id is unknown. + assert _status_edges(server.posts) == [] + await fwd.handle_event(_event("message.updated", info={"id": "msg_1", "role": "assistant"})) + running = [e for e in _status_edges(server.posts) if e["status"] == "running"] + assert len(running) == 1 + assert running[0]["response_id"] == "msg_1" + + +async def test_second_turn_gets_its_own_running_response_id() -> None: + """Each turn's running/idle edges carry that turn's own assistant id.""" + server, opencode = _RecordingServerClient(), _FakeOpenCodeClient() + fwd = _forwarder(server, opencode) + for msg in ("msg_a", "msg_b"): + await fwd.handle_event(_event("message.updated", info={"id": msg, "role": "assistant"})) + await fwd.handle_event(_event("session.idle")) + edges = _status_edges(server.posts) + assert [(e["status"], e["response_id"]) for e in edges] == [ + ("running", "msg_a"), + ("idle", "msg_a"), + ("running", "msg_b"), + ("idle", "msg_b"), + ] + + +async def test_multi_assistant_message_turn_retires_with_the_live_id() -> None: + """Two assistant messages in ONE turn: idle carries the id that went live. + + If opencode emits more than one assistant ``message.updated`` before + ``session.idle`` (no idle between them), the ``running`` edge locks to the + first id (``msg_1``) while ``_active_message_id`` advances to ``msg_2``. The + terminal ``idle`` edge must still carry ``msg_1`` — the id the running edge + used — so the web retires the tool cards that were actually rendered live. + """ + server, opencode = _RecordingServerClient(), _FakeOpenCodeClient() + fwd = _forwarder(server, opencode) + await fwd.handle_event(_event("message.updated", info={"id": "msg_1", "role": "assistant"})) + await fwd.handle_event(_event("message.updated", info={"id": "msg_2", "role": "assistant"})) + await fwd.handle_event(_event("session.idle")) + edges = _status_edges(server.posts) + assert [(e["status"], e["response_id"]) for e in edges] == [ + ("running", "msg_1"), + ("idle", "msg_1"), + ] + + +async def test_turn_without_assistant_message_idles_with_session_fallback() -> None: + """A turn that opens (busy) and idles with no assistant ``message.updated``. + + No ``running`` edge fires (there was never an id to carry) and the terminal + ``idle`` edge falls back to the session id. Benign — there are no live tool + cards to retire — but the fallback id is deliberate, not a mismatch bug. + """ + server, opencode = _RecordingServerClient(), _FakeOpenCodeClient() + fwd = _forwarder(server, opencode) + await fwd.handle_event(_event("session.status", status={"type": "busy"})) + await fwd.handle_event(_event("session.idle")) + edges = _status_edges(server.posts) + assert [e["status"] for e in edges] == ["idle"] + assert edges[0]["response_id"] == _SESSION + + async def test_permission_asked_rejects_when_no_policy_wired() -> None: """Absent a policy evaluator the forwarder FAILS CLOSED (no auto-approve).