Merge remote-tracking branch 'origin/main' into feat/native-claude-live-tool-cards

# Conflicts:
#	tests/e2e/test_repl_approval_e2e.py

omnigent/_native_post_delivery.py

@@ -169,6 +169,7 @@ async def post_external_session_status(
     status: str,
     output: str | None = None,
     background_task_count: int | None = None,
+    response_id: str | None = None,
 ) -> None:
     """Post one ``external_session_status`` event to the Sessions API.
 
@@ -186,6 +187,13 @@ async def post_external_session_status(
         edge, forwarded so the UI can show "N background tasks still running".
         ``None`` omits the field (server leaves its sticky tally untouched) — the
         default for edges that know nothing about background shells.
+    :param response_id: Optional id of the assistant turn this status edge
+        belongs to. When set, the server attaches it to the ``session.status``
+        SSE event so ap-web can drive the bubble's streaming lifecycle — that's
+        what makes native forwarded tool cards render LIVE (spinner + elapsed
+        timer) rather than as static completed cards. ``None`` (the default)
+        preserves the bare, turn-agnostic status edges (e.g. the sub-agent
+        quiescence badge) that don't map to a turn.
     :raises httpx.HTTPError: If the Omnigent request fails or is rejected.
     """
     data: dict[str, object] = {"status": status}
@@ -193,6 +201,8 @@ async def post_external_session_status(
         data["output"] = output
     if background_task_count is not None:
         data["background_task_count"] = background_task_count
+    if response_id is not None:
+        data["response_id"] = response_id
     resp = await client.post(
         f"/v1/sessions/{session_id}/events",
         json={"type": "external_session_status", "data": data},

omnigent/claude_native_forwarder.py

@@ -521,6 +521,12 @@ class _ForwardDedupeState:
     # sub-agent spend so the gate can block mid-turn. Separate baseline
     # because it can advance while ``posted_cost`` (S) is frozen.
     posted_policy_cost: float | None = None
+    # Response id of the last turn-start ``running`` status POSTed, so the
+    # id-bearing running edge fires exactly once per turn even when an
+    # assistant item is held across polls for delta ordering (which leaves
+    # ``state.current_response_id`` unadvanced). ``None`` until the first
+    # turn-start edge. Reset on /clear and /fork like the other baselines.
+    posted_running_response_id: str | None = None
 
 
 @dataclass(frozen=True)
@@ -916,6 +922,14 @@ async def forward_claude_transcript_to_session(
                         task_subjects=task_subjects,
                         task_statuses=task_statuses,
                         task_order=task_order,
+                        # The turn-end edges (Stop→idle / StopFailure→failed)
+                        # carry the turn's response id so ap-web can CLOSE the
+                        # streaming ``activeResponse`` opened by the turn-start
+                        # ``running`` edge (_forward_available_items). The
+                        # transcript forwarder ran just above, so
+                        # ``state.current_response_id`` is the active turn's id
+                        # (the user-message reset only fires on the next turn).
+                        response_id=state.current_response_id,
                     )
                     subagent_state = await _forward_available_subagents(
                         client=client,
@@ -2510,6 +2524,7 @@ async def _forward_available_status_events(
     task_subjects: dict[str, str],
     task_statuses: dict[str, str],
     task_order: list[str],
+    response_id: str | None = None,
 ) -> HookForwardState:
     """
     Forward currently available hook events as ``session.status``.
@@ -2544,6 +2559,11 @@ async def _forward_available_status_events(
     :param task_order: Mutable ordered list of task ids in creation order,
         e.g. ``["1", "2", "3"]``. Appended in-place from ``TaskCreated``
         events. Used to render the task list in a stable order.
+    :param response_id: Active turn's response id, stamped on the
+        ``Stop``→``idle`` / ``StopFailure``→``failed`` edges so ap-web
+        closes the streaming ``activeResponse`` opened by the matching
+        turn-start ``running`` edge. ``None`` when no turn id is known
+        (the status still posts, just without a turn association).
     :returns: Updated state. On post failure, returns the last
         durable state so successfully-posted statuses are not
         retried and the failing event is retried later.
@@ -2706,6 +2726,7 @@ async def _forward_available_status_events(
                 client,
                 session_id=session_id,
                 status=effective_status,
+                response_id=response_id,
                 # Only the ``Stop`` (idle/waiting) edge carries an authoritative
                 # background-shell count — ``0`` clears the tally, ``N`` sets it.
                 # ``StopFailure`` (failed) clears it on the server regardless, so
@@ -2734,6 +2755,7 @@ async def _forward_available_status_events(
                         session_id=session_id,
                         bridge_dir=bridge_dir,
                         reason=f"hook status {status} rejected",
+                        response_id=response_id,
                     )
                 durable = next_durable
                 await _write_hook_state_async(bridge_dir, durable)
@@ -2866,6 +2888,39 @@ async def _forward_available_items(
     # never fired ``UserPromptSubmit``). PTY-activity status makes it
     # obsolete: the pane keeps changing through a mid-turn compaction, so
     # the runner's watcher holds the session ``running`` directly.
+    #
+    # Turn-start edge: the first time we see a turn's response id, publish a
+    # ``running`` status carrying it. The PTY watcher already drives the
+    # running/idle BADGE with a bare (id-less) status; this id-bearing edge is
+    # what lets ap-web open a *streaming* ``activeResponse`` for the turn, so
+    # the forwarded tool-call cards (which carry the same response id) render
+    # LIVE — spinner + elapsed timer — instead of as static completed cards.
+    # Deduped on the persistent ``dedupe`` baseline (NOT ``state``): when an
+    # assistant item is held across polls for delta ordering, this function
+    # early-returns with ``state`` unadvanced, so a ``state``-based guard would
+    # re-fire ``running`` every poll of the hold window. Best-effort — a failed
+    # status post must not abort item forwarding (the items below are the
+    # primary payload); the turn-end idle/failed edge still carries the id to
+    # close the lifecycle, and the badge is unaffected either way.
+    if (
+        current_response_id is not None
+        and dedupe.posted_running_response_id != current_response_id
+    ):
+        try:
+            await post_external_session_status(
+                client,
+                session_id=session_id,
+                status="running",
+                response_id=current_response_id,
+            )
+            dedupe.posted_running_response_id = current_response_id
+        except httpx.HTTPError:
+            _logger.warning(
+                "Failed to forward Claude turn-start running status; session=%s response_id=%s",
+                session_id,
+                current_response_id,
+                exc_info=True,
+            )
     updated = state
     for item in items:
         if item.source_id in seen:
@@ -2924,6 +2979,7 @@ async def _forward_available_items(
                     session_id=session_id,
                     bridge_dir=bridge_dir,
                     reason=f"transcript item {item.source_id} rejected",
+                    response_id=current_response_id,
                 )
                 seen.add(item.source_id)
                 seen_source_ids.append(item.source_id)
@@ -3888,6 +3944,7 @@ async def _post_forwarder_failed_status(
     session_id: str,
     bridge_dir: Path,
     reason: str,
+    response_id: str | None = None,
 ) -> None:
     """
     Best-effort publish a failed status after dropping a poison event.
@@ -3897,11 +3954,19 @@ async def _post_forwarder_failed_status(
     :param bridge_dir: Native Claude bridge directory.
     :param reason: Diagnostic reason for the failure event, e.g.
         ``"transcript item item-1 rejected"``.
+    :param response_id: Active turn's response id, so this ``failed``
+        edge closes the streaming ``activeResponse`` for the matching
+        turn rather than leaving its tool cards spinning. ``None`` when
+        no turn id is known.
     :returns: None.
     """
     try:
         await post_external_session_status(
-            client, session_id=session_id, status="failed", output=reason
+            client,
+            session_id=session_id,
+            status="failed",
+            output=reason,
+            response_id=response_id,
         )
     except httpx.HTTPError:
         _logger.warning(

omnigent/server/routes/sessions.py

@@ -840,6 +840,14 @@ _WATCHER_TASKS: set[asyncio.Task[None]] = set()
 # Used by _get_session_snapshot.
 _session_status_cache: dict[str, str] = {}
 
+# Per-session in-flight response id, tracked alongside _session_status_cache.
+# Set when a running/waiting status edge carries a response_id (native Claude's
+# turn-start edge does); popped on idle/failed. Projected onto the session
+# snapshot as ``active_response_id`` so a client reconnecting mid-turn can
+# reopen the streaming ``activeResponse`` and keep forwarded tool cards
+# rendering LIVE — the SSE stream is "snapshot + live tail, no replay", so the
+# turn-start ``running`` event is never re-sent on reconnect.
+_session_active_response_cache: dict[str, str] = {}
 # Per-session background-shell tally (claude-native), kept in lockstep with
 # ``_session_status_cache`` so a snapshot/reload re-shows "N background tasks
 # still running" after the live SSE edge is gone. The authoritative source is
@@ -2535,6 +2543,11 @@ def _build_session_response(
         # once the launch succeeds; a failed launch is retained with
         # its reason. Populated by _publish_sandbox_status.
         sandbox_status=_session_sandbox_status_cache.get(conv.id),
+        # In-flight turn id so a mid-turn reconnect can reopen a streaming
+        # ``activeResponse`` (the turn-start ``running`` edge that carried it
+        # is not replayed on the SSE stream). Populated for native-terminal
+        # sessions whose forwarder stamps a turn id; ``None`` otherwise.
+        active_response_id=_session_active_response_cache.get(conv.id),
     )
 
 
@@ -5401,8 +5414,20 @@ def _publish_status(
     # transition (compaction failure publishes ``running`` → ``idle``, not
     # ``failed``), so this is a safe, harness-agnostic invariant.
     if status == "idle" and _session_status_cache.get(session_id) == "failed":
+        # Session stays ``failed`` (terminal); the turn is over, so drop any
+        # tracked in-flight response id rather than leaving it for the
+        # snapshot to reopen a streaming bubble.
+        _session_active_response_cache.pop(session_id, None)
         return
     _session_status_cache[session_id] = status
+    # Track the in-flight response id for snapshot-based reconnect (see
+    # _session_active_response_cache). A running/waiting edge that names a
+    # turn opens it; any idle/failed edge closes it.
+    if status in ("running", "waiting"):
+        if response_id is not None:
+            _session_active_response_cache[session_id] = response_id
+    else:
+        _session_active_response_cache.pop(session_id, None)
     # Keep the background-shell tally sticky alongside the status (see the
     # cache's declaration). A ``Stop`` hook reports an authoritative count
     # (``None`` is never sent by it): a positive count sets the tally, and

omnigent/server/schemas.py

@@ -1690,6 +1690,15 @@ class SessionResponse(BaseModel):
         ``_session_sandbox_status_cache`` at snapshot build time, so
         a client opening the session mid-launch sees the current
         stage.
+    :param active_response_id: Response id of the turn currently in
+        flight, or ``None`` when the session is idle. Sourced from the
+        server's ``_session_active_response_cache`` at snapshot build
+        time so a client connecting mid-turn can reopen a streaming
+        ``activeResponse`` — the SSE stream is snapshot + live tail with
+        no replay, so the turn-start ``running`` edge that carried this
+        id is not re-sent on reconnect. Today only native-terminal
+        forwarders (claude-native) stamp a turn id on their status
+        edges; other harnesses leave this ``None``.
     """
 
     id: str
@@ -1739,6 +1748,7 @@ class SessionResponse(BaseModel):
     model_options: list[dict[str, Any]] = Field(default_factory=list)
     terminal_pending: bool = False
     sandbox_status: SandboxStatus | None = None
+    active_response_id: str | None = None
 
 
 class UpdateSessionRequest(BaseModel):

openapi.json

@@ -4225,6 +4225,18 @@
       "SessionResponse": {
         "description": "API representation of a session.\n\nReturned by `POST /v1/sessions`, `GET /v1/sessions/{id}`,\nand `PATCH /v1/sessions/{id}`.",
         "properties": {
+          "active_response_id": {
+            "anyOf": [
+              {
+                "type": "string"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "description": "Response id of the turn currently in flight, or `None` when the session is idle. Sourced from the server's `_session_active_response_cache` at snapshot build time so a client connecting mid-turn can reopen a streaming `activeResponse` \u2014 the SSE stream is snapshot + live tail with no replay, so the turn-start `running` edge that carried this id is not re-sent on reconnect. Today only native-terminal forwarders (claude-native) stamp a turn id on their status edges; other harnesses leave this `None`.",
+            "title": "Active Response Id"
+          },
           "agent_id": {
             "description": "Durable identifier of the bound agent, e.g. `\"ag_abc123\"`. Stable across renames of the agent.",
             "title": "Agent Id",

tests/e2e_ui/chat/test_working_indicator_reload.py

@@ -12,23 +12,140 @@ import httpx
 from playwright.sync_api import Page, expect
 
 
-def _publish_status(base_url: str, session_id: str, status: str) -> None:
+def _publish_status(
+    base_url: str, session_id: str, status: str, response_id: str | None = None
+) -> None:
     """Publish a session status through the same Omnigent route native harnesses use.
 
     :param base_url: Base URL of the local e2e server, e.g.
         ``"http://127.0.0.1:51234"``.
     :param session_id: Session/conversation id, e.g. ``"conv_abc123"``.
     :param status: Session status to publish, e.g. ``"running"``.
+    :param response_id: Optional in-flight turn id. When set on a
+        ``running``/``waiting`` edge, the server tracks it and projects it onto
+        the session snapshot as ``active_response_id`` — the signal native
+        Claude's forwarder now sends so a mid-turn (re)connect reopens the
+        streaming lifecycle and renders forwarded tool cards LIVE.
+    :returns: None.
+    """
+    data: dict[str, str] = {"status": status}
+    if response_id is not None:
+        data["response_id"] = response_id
+    resp = httpx.post(
+        f"{base_url}/v1/sessions/{session_id}/events",
+        json={"type": "external_session_status", "data": data},
+        timeout=10.0,
+    )
+    resp.raise_for_status()
+
+
+def _seed_function_call(
+    base_url: str,
+    session_id: str,
+    *,
+    response_id: str,
+    call_id: str,
+    name: str,
+    arguments: str,
+) -> None:
+    """Mirror one in-flight native tool call (no output yet) onto the session.
+
+    Posts the same ``external_conversation_item`` / ``function_call`` a native
+    forwarder emits, tagged with ``response_id`` so it belongs to the in-flight
+    turn. With no ``function_call_output`` following, the call is still running.
+
+    :param base_url: Base URL of the local e2e server.
+    :param session_id: Session/conversation id.
+    :param response_id: Turn id the call belongs to (matches the ``running`` edge).
+    :param call_id: Tool-call id, e.g. ``"call_live_1"``.
+    :param name: Tool name, e.g. ``"shell"``.
+    :param arguments: JSON-encoded arguments string, e.g. ``'{"command": "..."}'``.
     :returns: None.
     """
     resp = httpx.post(
         f"{base_url}/v1/sessions/{session_id}/events",
-        json={"type": "external_session_status", "data": {"status": status}},
+        json={
+            "type": "external_conversation_item",
+            "data": {
+                "item_type": "function_call",
+                "item_data": {
+                    "agent": "claude-native-ui",
+                    "name": name,
+                    "arguments": arguments,
+                    "call_id": call_id,
+                },
+                "response_id": response_id,
+            },
+        },
         timeout=10.0,
     )
     resp.raise_for_status()
 
 
+def _snapshot_active_response_id(base_url: str, session_id: str) -> str | None:
+    """Return ``active_response_id`` from the session snapshot.
+
+    :param base_url: Base URL of the local e2e server.
+    :param session_id: Session/conversation id.
+    :returns: The in-flight turn id the server is tracking, or ``None`` when idle.
+    """
+    resp = httpx.get(f"{base_url}/v1/sessions/{session_id}", timeout=10.0)
+    resp.raise_for_status()
+    return resp.json().get("active_response_id")
+
+
+def test_midturn_connect_renders_live_tool_card(
+    page: Page,
+    seeded_session: tuple[str, str],
+) -> None:
+    """A mid-turn connect renders a forwarded tool call as a LIVE card.
+
+    Reproduces native Claude's live-tool-card path without a real LLM turn: a
+    tool call is mirrored mid-turn (no output yet) and the turn-start ``running``
+    edge carries its ``response_id``. The server tracks that id and projects it
+    as ``active_response_id`` on the snapshot; a browser connecting fresh
+    (no prior local streaming state) reopens the streaming ``activeResponse``
+    from that snapshot, so the tool card renders in its running state — a
+    spinner (``Loader2`` ``animate-spin``, emitted only for ``input-available``).
+    Before this change the reconnect left the bubble non-streaming, so the same
+    call rendered as a static, spinner-less card.
+
+    :param page: Playwright page fixture.
+    :param seeded_session: ``(base_url, session_id)`` from the local server.
+    :returns: None.
+    """
+    base_url, session_id = seeded_session
+    response_id = "resp_live_tool_1"
+    _publish_status(base_url, session_id, "running", response_id=response_id)
+    _seed_function_call(
+        base_url,
+        session_id,
+        response_id=response_id,
+        call_id="call_live_1",
+        name="shell",
+        arguments='{"command": "sleep 30"}',
+    )
+
+    try:
+        # Server half: the snapshot exposes the in-flight turn id.
+        assert _snapshot_active_response_id(base_url, session_id) == response_id
+
+        # UI half: a fresh connect reopens streaming from the snapshot, so the
+        # tool card shows the running spinner. The Working indicator uses a
+        # different mark (OttoIcon/Shimmer, not animate-spin), so a spinning
+        # loader in the transcript is unambiguously the live tool card.
+        page.goto(f"{base_url}/c/{session_id}")
+        spinner = page.locator(".animate-spin")
+        expect(spinner.first).to_be_visible(timeout=20_000)
+
+        # A full reload re-hydrates from the same snapshot and stays live —
+        # this is the reconnect path, not a fluke of the live SSE tail.
+        page.reload()
+        expect(spinner.first).to_be_visible(timeout=20_000)
+    finally:
+        _publish_status(base_url, session_id, "idle", response_id=response_id)
+
+
 def test_running_empty_session_reload_keeps_working_indicator(
     page: Page,
     seeded_session_pair: tuple[str, str, str],

tests/server/integration/test_sessions_endpoints.py

@@ -2853,6 +2853,42 @@ async def test_publish_status_keeps_failed_sticky_against_trailing_idle(
     assert cache_after == "idle"
 
 
+async def test_publish_status_tracks_in_flight_response_id(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """
+    ``_publish_status`` records the in-flight response id and clears it on end.
+
+    A ``running``/``waiting`` edge carrying a ``response_id`` opens the
+    ``_session_active_response_cache`` entry (projected onto the snapshot as
+    ``active_response_id`` so a mid-turn reconnect reopens the streaming
+    ``activeResponse``); any ``idle``/``failed`` edge clears it. A bare
+    ``running`` (no id, e.g. the PTY badge edge) must NOT clobber a tracked id.
+    """
+    from omnigent.server.routes import sessions as sessions_module
+
+    monkeypatch.setattr(
+        "omnigent.server.routes.sessions.session_stream.publish",
+        lambda _session_id, _event: None,
+    )
+    sid = "conv_active_resp"
+    sessions_module._session_status_cache.pop(sid, None)
+    sessions_module._session_active_response_cache.pop(sid, None)
+    try:
+        # Turn start: running with the turn id opens the tracked entry.
+        sessions_module._publish_status(sid, "running", response_id="resp_turn_1")
+        assert sessions_module._session_active_response_cache.get(sid) == "resp_turn_1"
+        # Bare PTY running (no id) must not erase the tracked turn id.
+        sessions_module._publish_status(sid, "running")
+        assert sessions_module._session_active_response_cache.get(sid) == "resp_turn_1"
+        # Turn end: idle clears it.
+        sessions_module._publish_status(sid, "idle", response_id="resp_turn_1")
+        assert sessions_module._session_active_response_cache.get(sid) is None
+    finally:
+        sessions_module._session_status_cache.pop(sid, None)
+        sessions_module._session_active_response_cache.pop(sid, None)
+
+
 async def test_patch_runner_rebind_clears_stale_failed_status(
     client: httpx.AsyncClient,
     monkeypatch: pytest.MonkeyPatch,

tests/test_claude_native_forwarder.py

@@ -182,6 +182,30 @@ async def _get_recorded_request(
             return request
 
 
+async def _get_recorded_item_request(
+    server: _RecordingHTTPServer,
+    *,
+    timeout_s: float = 5.0,
+) -> dict[str, Any]:
+    """
+    Await the next ``external_conversation_item`` POST, skipping status edges.
+
+    The forwarder now emits a turn-start ``external_session_status: running``
+    (carrying the turn's response id, which drives the live tool-card spinner
+    in ap-web) BEFORE a turn's items each poll. Tests that only care about the
+    forwarded conversation items use this to skip that leading status edge (and
+    any trailing idle) without asserting on it.
+
+    :param server: Recording HTTP server.
+    :param timeout_s: Per-``get`` timeout while skipping non-item POSTs.
+    :returns: The next recorded ``external_conversation_item`` POST.
+    """
+    while True:
+        request = await _get_recorded_request(server, timeout_s=timeout_s)
+        if request["body"].get("type") == "external_conversation_item":
+            return request
+
+
 async def _wait_for_json_file(path: Path, *, timeout_s: float = 5.0) -> dict[str, Any]:
     """
     Wait until a JSON object file exists and can be parsed.
@@ -1142,10 +1166,11 @@ async def test_forwarder_posts_visible_transcript_items(tmp_path: Path) -> None:
         )
     )
     try:
-        # Seven transcript items, then the ``Stop`` → idle status. Items are
-        # forwarded before status each poll, so the first 7 collected are the
-        # items; the trailing idle is not asserted here.
-        requests = [await _get_recorded_request(server) for _index in range(7)]
+        # The turn-start ``running`` status (carrying the turn's response id)
+        # posts first, then the seven transcript items, then the ``Stop`` →
+        # idle status. Collect the running edge + 7 items; the trailing idle is
+        # not asserted here.
+        requests = [await _get_recorded_request(server) for _index in range(8)]
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):
@@ -1154,8 +1179,9 @@ async def test_forwarder_posts_visible_transcript_items(tmp_path: Path) -> None:
         server.server_close()
         thread.join(timeout=5.0)
 
-    assert [request["path"] for request in requests] == ["/v1/sessions/conv_abc/events"] * 7
+    assert [request["path"] for request in requests] == ["/v1/sessions/conv_abc/events"] * 8
     assert [request["body"]["type"] for request in requests] == [
+        "external_session_status",
         "external_conversation_item",
         "external_conversation_item",
         "external_conversation_item",
@@ -1164,6 +1190,10 @@ async def test_forwarder_posts_visible_transcript_items(tmp_path: Path) -> None:
         "external_conversation_item",
         "external_conversation_item",
     ]
+    # The leading status is the turn-start ``running`` edge carrying the turn's
+    # response id (what drives the live tool-card spinner on the client).
+    assert requests[0]["body"]["data"]["status"] == "running"
+    assert isinstance(requests[0]["body"]["data"].get("response_id"), str)
     posted = [
         request["body"]["data"]
         for request in requests
@@ -1206,6 +1236,11 @@ async def test_forwarder_posts_visible_transcript_items(tmp_path: Path) -> None:
     assert posted[5]["response_id"] == posted[6]["response_id"]
     assert posted[5]["response_id"] != posted[4]["response_id"]
     assert posted[1]["response_id"].startswith("resp_claude_")
+    # The turn-start running edge carries an assistant turn's response id (here
+    # the whole multi-turn transcript flushes in one poll, so it's the last
+    # turn's id). The single-turn id↔function_call match is asserted directly in
+    # test_forwarder_emits_turn_start_running_with_response_id.
+    assert requests[0]["body"]["data"]["response_id"].startswith("resp_claude_")
 
 
 @pytest.mark.asyncio
@@ -1355,6 +1390,9 @@ async def test_forwarder_posts_web_injected_terminal_transcript_items(tmp_path:
         )
     )
     try:
+        # The turn-start ``running`` status posts first (the transcript has an
+        # assistant turn), then the assistant message item.
+        running = await _get_recorded_request(server)
         request = await _get_recorded_request(server)
     finally:
         task.cancel()
@@ -1364,6 +1402,8 @@ async def test_forwarder_posts_web_injected_terminal_transcript_items(tmp_path:
         server.server_close()
         thread.join(timeout=5.0)
 
+    assert running["body"]["type"] == "external_session_status"
+    assert running["body"]["data"]["status"] == "running"
     assert request["path"] == "/v1/sessions/conv_abc/events"
     assert request["body"]["type"] == "external_conversation_item"
     assert request["body"]["data"]["item_type"] == "message"
@@ -2092,7 +2132,7 @@ async def test_forwarder_start_at_end_uses_byte_offset_for_new_lines(
         assert state["byte_offset"] == len(old_prefix.encode("utf-8"))
         with transcript_path.open("a", encoding="utf-8") as handle:
             handle.write("}\n")
-        request = await _get_recorded_request(server)
+        request = await _get_recorded_item_request(server)
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):
@@ -2177,7 +2217,7 @@ async def test_forwarder_migrates_line_cursor_state_to_byte_offset(tmp_path: Pat
         )
     )
     try:
-        request = await _get_recorded_request(server)
+        request = await _get_recorded_item_request(server)
         state = await _wait_for_json_state(
             bridge_dir / "transcript_forwarder.json",
             lambda payload: payload.get("line_cursor") == 2 and "byte_offset" in payload,
@@ -2262,7 +2302,7 @@ async def test_forwarder_waits_for_missing_fresh_transcript_without_warning(
             encoding="utf-8",
         )
 
-        request = await _get_recorded_request(server)
+        request = await _get_recorded_item_request(server)
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):
@@ -2378,7 +2418,7 @@ async def test_forwarder_skips_to_end_on_stale_byte_cursor_state(tmp_path: Path)
         with transcript_path.open("a", encoding="utf-8") as f:
             f.write(new_record)
         # The new record should be forwarded.
-        request = await _get_recorded_request(server)
+        request = await _get_recorded_item_request(server)
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):
@@ -2487,7 +2527,7 @@ async def test_forwarder_skips_to_end_on_out_of_range_byte_cursor_without_finger
         )
         with transcript_path.open("a", encoding="utf-8") as f:
             f.write(new_record)
-        request = await _get_recorded_request(server)
+        request = await _get_recorded_item_request(server)
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):
@@ -2640,8 +2680,9 @@ async def test_forwarder_does_not_replay_after_compaction(tmp_path: Path) -> Non
         with transcript_path.open("a", encoding="utf-8") as f:
             f.write(new_record)
 
-        # The new record should be the only thing forwarded.
-        request = await _get_recorded_request(server)
+        # The new record should be the only item forwarded (the turn also
+        # emits a leading running status edge, which this helper skips).
+        request = await _get_recorded_item_request(server)
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):
@@ -2799,7 +2840,7 @@ async def test_forwarder_survives_unhandled_loop_exceptions(
         )
     )
     try:
-        request = await _get_recorded_request(server)
+        request = await _get_recorded_item_request(server)
     finally:
         task.cancel()
         with pytest.raises(asyncio.CancelledError):
@@ -2893,17 +2934,23 @@ async def test_forwarder_drops_poison_item_after_bounded_permanent_retries(
         )
 
     persisted = json.loads((bridge_dir / "transcript_forwarder.json").read_text("utf-8"))
+    # The turn-start ``running`` status (carrying the turn's response id) leads,
+    # then the poison item is attempted twice, then the forwarder-failed status.
     assert [request["type"] for request in requests] == [
+        "external_session_status",
         "external_conversation_item",
         "external_conversation_item",
         "external_session_status",
     ]
-    # The failed edge carries the drop reason as ``output`` so the server
-    # surfaces it as the session's failure detail instead of a bare
-    # "failed" badge (#1113).
+    # The turn-start ``running`` edge carries the turn's response id, and the
+    # failed edge carries BOTH the drop reason as ``output`` (#1113 — the
+    # server surfaces it as the failure detail) and that same response id so
+    # it closes the streaming turn instead of leaving its tool cards spinning.
+    assert requests[0]["data"]["status"] == "running"
     assert requests[-1]["data"] == {
         "status": "failed",
         "output": "transcript item poison-item:0:message rejected",
+        "response_id": requests[0]["data"]["response_id"],
     }
     assert first.byte_offset == 0
     assert second.byte_offset == transcript_path.stat().st_size
@@ -5352,8 +5399,12 @@ async def test_assistant_item_held_until_its_deltas_forward(tmp_path: Path) -> N
             ordering=ordering,
         )
         # Held: the item was NOT posted and the durable cursor did not
-        # advance past it, so the next poll re-reads it.
-        assert [c.body["type"] for c in captured] == ["external_output_text_delta"]
+        # advance past it, so the next poll re-reads it. (The turn-start
+        # ``external_session_status: running`` edge is filtered out here — this
+        # test is about delta-vs-item ordering, not the status edge.)
+        assert [
+            c.body["type"] for c in captured if c.body["type"] != "external_session_status"
+        ] == ["external_output_text_delta"]
         assert item_state.byte_offset == 0
         assert item_state.seen_source_ids == ()
 
@@ -5388,7 +5439,9 @@ async def test_assistant_item_held_until_its_deltas_forward(tmp_path: Path) -> N
     assert item_posts[0]["data"]["item_data"]["content"] == [
         {"type": "output_text", "text": "Hello world"}
     ]
-    assert [c.body["type"] for c in captured][:2] == [
+    assert [c.body["type"] for c in captured if c.body["type"] != "external_session_status"][
+        :2
+    ] == [
         "external_output_text_delta",
         "external_output_text_delta",
     ]
@@ -5423,6 +5476,9 @@ async def test_assistant_item_posts_after_hold_timeout(
 
     ordering = forwarder._DeltaOrderingState()
     captured: list[_CapturedDeltaPost] = []
+    # Share one dedupe across polls (as the real loop does) so the turn-start
+    # ``running`` status fires once, not per call.
+    dedupe = forwarder._ForwardDedupeState()
     async with _delta_capture_client(captured) as client:
         await forwarder._forward_available_deltas(
             client=client,
@@ -5439,10 +5495,14 @@ async def test_assistant_item_posts_after_hold_timeout(
             agent_name="claude-native-ui",
             state=_transcript_state_for(transcript_path),
             retry_tracker=forwarder._PostRetryTracker(),
-            dedupe=forwarder._ForwardDedupeState(),
+            dedupe=dedupe,
             ordering=ordering,
         )
-        assert [c.body["type"] for c in captured] == ["external_output_text_delta"]
+        # Status edges (the turn-start ``running``) filtered out — this test
+        # is about the delta-then-held-item ordering.
+        assert [
+            c.body["type"] for c in captured if c.body["type"] != "external_session_status"
+        ] == ["external_output_text_delta"]
         assert state.byte_offset == 0  # held
 
         clock["now"] = 100.0 + forwarder._ASSISTANT_ITEM_DELTA_HOLD_S
@@ -5453,7 +5513,7 @@ async def test_assistant_item_posts_after_hold_timeout(
             agent_name="claude-native-ui",
             state=state,
             retry_tracker=forwarder._PostRetryTracker(),
-            dedupe=forwarder._ForwardDedupeState(),
+            dedupe=dedupe,
             ordering=ordering,
         )
 
@@ -5514,6 +5574,9 @@ async def test_assistant_item_stays_held_until_true_final_chunk(tmp_path: Path)
     delta_state = forwarder.DeltaForwardState()
     item_state = _transcript_state_for(transcript_path)
     captured: list[_CapturedDeltaPost] = []
+    # Share one dedupe across polls (as the real loop does) so the turn-start
+    # ``running`` status fires once, not per poll.
+    dedupe = forwarder._ForwardDedupeState()
 
     async def _poll(client: httpx.AsyncClient) -> None:
         nonlocal delta_state, item_state
@@ -5532,7 +5595,7 @@ async def test_assistant_item_stays_held_until_true_final_chunk(tmp_path: Path)
             agent_name="claude-native-ui",
             state=item_state,
             retry_tracker=forwarder._PostRetryTracker(),
-            dedupe=forwarder._ForwardDedupeState(),
+            dedupe=dedupe,
             ordering=ordering,
         )
 
@@ -5559,7 +5622,9 @@ async def test_assistant_item_stays_held_until_true_final_chunk(tmp_path: Path)
         await _poll(client)
 
     # Commit posts only AFTER all three deltas — the order downstream assumes.
-    assert [c.body["type"] for c in captured] == [
+    # The turn-start ``running`` status edge is filtered out (it fires once,
+    # before the deltas); this test is about delta-then-commit ordering.
+    assert [c.body["type"] for c in captured if c.body["type"] != "external_session_status"] == [
         "external_output_text_delta",
         "external_output_text_delta",
         "external_output_text_delta",
@@ -5713,8 +5778,11 @@ async def test_without_hold_commit_posts_before_final_delta(tmp_path: Path) -> N
             dedupe=forwarder._ForwardDedupeState(),
             ordering=None,
         )
-        # Bug: with no hold the commit posts immediately, before the final chunk.
-        assert [c.body["type"] for c in captured] == [
+        # Bug: with no hold the commit posts immediately, before the final
+        # chunk. The turn-start ``running`` status edge is filtered out.
+        assert [
+            c.body["type"] for c in captured if c.body["type"] != "external_session_status"
+        ] == [
             "external_output_text_delta",
             "external_conversation_item",
         ]
@@ -5731,7 +5799,7 @@ async def test_without_hold_commit_posts_before_final_delta(tmp_path: Path) -> N
         )
 
     # The final delta lands AFTER the commit — the inverted order that dupes.
-    types = [c.body["type"] for c in captured]
+    types = [c.body["type"] for c in captured if c.body["type"] != "external_session_status"]
     commit_idx = types.index("external_conversation_item")
     final_delta_idx = max(i for i, t in enumerate(types) if t == "external_output_text_delta")
     assert commit_idx < final_delta_idx
@@ -6664,3 +6732,184 @@ async def test_forwarder_posts_waiting_when_stop_has_background_tasks(
         "type": "external_session_status",
         "data": {"status": "waiting", "background_task_count": 1},
     }
+
+
+@pytest.mark.asyncio
+async def test_post_external_session_status_includes_and_omits_response_id() -> None:
+    """
+    ``post_external_session_status`` attaches ``response_id`` only when given.
+
+    The turn-bearing edges (native Claude's turn start/end) carry the response
+    id so ap-web can drive the bubble's streaming lifecycle; the bare,
+    turn-agnostic edges (e.g. the sub-agent quiescence badge) must keep posting
+    a ``data`` object with no ``response_id`` key so nothing spuriously matches.
+    """
+    bodies: list[dict[str, Any]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        bodies.append(json.loads(request.content.decode("utf-8")))
+        return httpx.Response(200, json={})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, base_url="http://ap") as client:
+        await forwarder.post_external_session_status(
+            client, session_id="conv_abc", status="running", response_id="resp_1"
+        )
+        await forwarder.post_external_session_status(client, session_id="conv_abc", status="idle")
+
+    assert bodies[0] == {
+        "type": "external_session_status",
+        "data": {"status": "running", "response_id": "resp_1"},
+    }
+    # Bare edge: no response_id key (not a null) so the server's optional
+    # validation passes and the client never opens a streaming response.
+    assert bodies[1] == {
+        "type": "external_session_status",
+        "data": {"status": "idle"},
+    }
+
+
+@pytest.mark.asyncio
+async def test_forward_status_events_stamps_response_id_on_idle(tmp_path: Path) -> None:
+    """
+    A ``Stop`` → idle edge carries the turn's ``response_id`` when one is known.
+
+    This is what closes the streaming ``activeResponse`` ap-web opened from the
+    turn-start ``running`` edge, so the trailing tool card stops spinning.
+    """
+    bridge_dir = tmp_path / "bridge"
+    record_hook_event(
+        bridge_dir,
+        {"hook_event_name": "Stop", "session_id": "claude-session"},
+    )
+    bodies: list[dict[str, Any]] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        bodies.append(json.loads(request.content.decode("utf-8")))
+        return httpx.Response(200, json={})
+
+    transport = httpx.MockTransport(handler)
+    async with httpx.AsyncClient(transport=transport, base_url="http://ap") as client:
+        hook_state = await forwarder._ensure_hook_state(
+            bridge_dir, start_at_end=False, session_id="conv_abc"
+        )
+        await forwarder._forward_available_status_events(
+            client=client,
+            session_id="conv_abc",
+            bridge_dir=bridge_dir,
+            state=hook_state,
+            retry_tracker=forwarder._PostRetryTracker(),
+            task_subjects={},
+            task_statuses={},
+            task_order=[],
+            response_id="resp_turn_1",
+        )
+
+    assert bodies == [
+        {
+            "type": "external_session_status",
+            # The Stop→idle edge carries the turn's response id AND the
+            # background-shell tally (0 here — no shells); the live-tool-card
+            # and background-task features share this one status edge.
+            "data": {
+                "status": "idle",
+                "background_task_count": 0,
+                "response_id": "resp_turn_1",
+            },
+        }
+    ]
+
+
+@pytest.mark.asyncio
+async def test_forwarder_emits_turn_start_running_with_response_id(tmp_path: Path) -> None:
+    """
+    The first assistant output of a turn publishes ``running`` + its response id.
+
+    Native Claude's running/idle BADGE stays PTY-derived; this id-bearing
+    ``running`` edge is the additional signal that lets ap-web open a streaming
+    ``activeResponse`` for the turn, so the forwarded tool cards (which share
+    the same response id) render LIVE rather than as static completed cards.
+    The running edge's response id must equal the forwarded items' response id.
+    """
+    bridge_dir = tmp_path / "bridge"
+    transcript_path = tmp_path / "session.jsonl"
+    transcript_path.write_text(
+        "\n".join(
+            [
+                json.dumps(
+                    {
+                        "type": "user",
+                        "uuid": "user-1",
+                        "message": {"role": "user", "content": "read TODO"},
+                    }
+                ),
+                json.dumps(
+                    {
+                        "type": "assistant",
+                        "uuid": "assistant-tool-1",
+                        "message": {
+                            "role": "assistant",
+                            "content": [
+                                {
+                                    "type": "tool_use",
+                                    "id": "toolu_read_1",
+                                    "name": "Read",
+                                    "input": {"file_path": "TODO.md"},
+                                }
+                            ],
+                        },
+                    }
+                ),
+            ]
+        )
+        + "\n",
+        encoding="utf-8",
+    )
+    record_hook_event(
+        bridge_dir,
+        {
+            "hook_event_name": "SessionStart",
+            "session_id": "claude-session",
+            "transcript_path": str(transcript_path),
+        },
+    )
+    server, thread, base_url = _start_recording_server()
+    task = asyncio.create_task(
+        forward_claude_transcript_to_session(
+            base_url=base_url,
+            headers={},
+            session_id="conv_abc",
+            bridge_dir=bridge_dir,
+            agent_name="claude-native-ui",
+            start_at_end=False,
+            poll_interval_s=0.01,
+        )
+    )
+    try:
+        # First POST of the poll is the turn-start running status (it runs
+        # before the items in _forward_available_items); the two items follow.
+        running = await _get_recorded_request(server)
+        item_a = await _get_recorded_request(server)
+        item_b = await _get_recorded_request(server)
+    finally:
+        task.cancel()
+        with pytest.raises(asyncio.CancelledError):
+            await task
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5.0)
+
+    assert running["body"]["type"] == "external_session_status"
+    assert running["body"]["data"]["status"] == "running"
+    running_rid = running["body"]["data"]["response_id"]
+    assert isinstance(running_rid, str) and running_rid
+    # The running edge's response id matches the ASSISTANT turn's forwarded
+    # item (the function_call), so that bubble enters the streaming lifecycle
+    # on the client. The user message carries its own distinct response id.
+    function_call = next(
+        body
+        for body in (item_a, item_b)
+        if body["body"]["type"] == "external_conversation_item"
+        and body["body"]["data"]["item_type"] == "function_call"
+    )
+    assert function_call["body"]["data"]["response_id"] == running_rid

web/src/lib/sessionsApi.test.ts

@@ -101,6 +101,7 @@ describe("createSession", () => {
       codexModelOptions: [],
       terminalPending: false,
       sandboxStatus: null,
+      activeResponseId: null,
       workspace: null,
       gitBranch: null,
     });
@@ -224,6 +225,39 @@ describe("createSession", () => {
       { pendingId: "pending_1", content: [{ type: "input_text", text: "queued" }] },
     ]);
   });
+
+  it("maps active_response_id (snake) to activeResponseId (camel)", async () => {
+    // The in-flight turn id lets a mid-turn reconnect reopen a streaming
+    // activeResponse so native Claude's tool cards keep rendering live.
+    fetchMock.mockResolvedValueOnce(
+      mockJsonResponse({
+        id: "conv_abc",
+        agent_id: "agent_xyz",
+        status: "running",
+        created_at: 1704067200,
+        items: [],
+        active_response_id: "resp_turn_1",
+      }),
+    );
+
+    const session = await createSession("agent_xyz");
+    expect(session.activeResponseId).toBe("resp_turn_1");
+  });
+
+  it("defaults activeResponseId to null when the snapshot omits it", async () => {
+    fetchMock.mockResolvedValueOnce(
+      mockJsonResponse({
+        id: "conv_abc",
+        agent_id: "agent_xyz",
+        status: "idle",
+        created_at: 1704067200,
+        items: [],
+      }),
+    );
+
+    const session = await createSession("agent_xyz");
+    expect(session.activeResponseId).toBeNull();
+  });
 });
 
 describe("forkSession", () => {

web/src/lib/sessionsApi.ts

@@ -212,6 +212,12 @@ interface SessionResponseWire {
    * `omnigent.server.schemas.SandboxStatus`.
    */
   sandbox_status?: SandboxStatus | null;
+  /**
+   * Response id of the turn currently in flight, or absent/null when
+   * idle. Lets a client reconnecting mid-turn reopen a streaming
+   * `activeResponse` (the turn-start `running` SSE edge is not replayed).
+   */
+  active_response_id?: string | null;
 }
 
 interface SessionItemsResponseWire {
@@ -298,6 +304,7 @@ function sessionFromWire(wire: SessionResponseWire): Session {
     codexModelOptions: wire.model_options ?? [],
     terminalPending: wire.terminal_pending ?? false,
     sandboxStatus: wire.sandbox_status ?? null,
+    activeResponseId: wire.active_response_id ?? null,
   };
 }
 

web/src/lib/types.ts

@@ -432,6 +432,16 @@ export interface Session {
    * sees the current stage.
    */
   sandboxStatus?: SandboxStatus | null;
+  /**
+   * Response id of the turn currently in flight, or `null`/absent when
+   * the session is idle. Sourced from the server's
+   * `_session_active_response_cache` at snapshot build time so a client
+   * reconnecting mid-turn can reopen a streaming `activeResponse` (the
+   * SSE stream is snapshot + live tail with no replay, so the turn-start
+   * `running` edge that carried this id is not re-sent). Today only
+   * native-terminal sessions (claude-native) populate it.
+   */
+  activeResponseId?: string | null;
 }
 
 /**

web/src/store/chatStore.ts

@@ -2175,6 +2175,14 @@ const RECONNECT_BACKFILL_MAX_PAGES = 4;
  * token/context/cost counters, and — when the turn ended during the gap —
  * the terminal `activeResponse` transition the missed `session.status`
  * event would have applied, so "Working…" clears.
+ *
+ * The inverse also matters: when the snapshot shows a turn STILL running and
+ * carries its in-flight `activeResponseId`, reopen the streaming
+ * `activeResponse`. The SSE stream is snapshot + live tail with no replay, so
+ * the turn-start `running` edge that originally opened it is never re-sent —
+ * without this, a client connecting mid-turn (reconnect, or first open of an
+ * already-running native session) would leave the turn's bubble non-streaming
+ * and its tool cards static for the rest of the turn.
  */
 function reconnectStatusPatch(session: Session, s: ChatState): Partial<ChatState> {
   const patch: Partial<ChatState> = { sessionStatus: session.status };
@@ -2195,6 +2203,20 @@ function reconnectStatusPatch(session: Session, s: ChatState): Partial<ChatState
       error: null,
     };
     patch.status = "idle";
+  } else if (
+    (session.status === "running" || session.status === "waiting") &&
+    session.activeResponseId != null &&
+    s.activeResponse?.responseId !== session.activeResponseId
+  ) {
+    // Mid-turn (re)connect: reopen the streaming lifecycle from the snapshot.
+    // Guarded on a differing responseId so we never downgrade a live
+    // activeResponse that already matches (e.g. one cancelled in this tab).
+    patch.activeResponse = {
+      responseId: session.activeResponseId,
+      state: "streaming",
+      error: null,
+    };
+    patch.status = "streaming";
   }
   return patch;
 }