fix(api_server): emit per-turn transcript on run.completed (#34703) (#34804)

* docs(code-execution): document HERMES_* env narrowing + passthrough workaround

The execute_code sandbox-child env scrub (108397726, #27303) deliberately
dropped the broad HERMES_ prefix passthrough, keeping only an operational
4-var allowlist (HERMES_HOME/PROFILE/CONFIG/ENV). A script that relied on a
non-secret HERMES_* var (HERMES_BASE_URL, HERMES_KANBAN_DB, HERMES_*_WEBHOOK,
or a plugin-defined one) now sees it unset in the child.

Document the behavior change and the two recovery routes (terminal.env_passthrough
in config.yaml, or required_environment_variables in skill frontmatter), plus
the debug log line that surfaces the drop for diagnosis.

* fix(api_server): emit per-turn transcript on run.completed (#34703)

WebUI clients lost intermediate (pre-tool-call) assistant text after
switching session pages mid-stream. The session-chat SSE stream delivers
all assistant text as assistant.delta events under one message_id
interleaved with tool.* events, then a single assistant.completed
carrying only the final reply — so a client accumulating deltas into one
buffer cannot reconstruct intermediate text segments that preceded tool
calls, and they vanish from the live view (state.db persists them
correctly).

run.completed now carries the authoritative per-turn transcript
(assistant + tool messages for this turn, in client-safe shape) so any
SSE consumer can reconcile its live view against ground truth without a
separate GET /messages round-trip. Purely additive — clients that ignore
the field are unaffected.

gateway/platforms/api_server.py

@@ -1605,6 +1605,7 @@ class APIServerAdapter(BasePlatformAdapter):
                 )
                 final_response = result.get("final_response", "") if isinstance(result, dict) else ""
                 effective_session_id = result.get("session_id", session_id) if isinstance(result, dict) else session_id
+                turn_messages = self._turn_transcript_messages(history, user_message, result) if isinstance(result, dict) else []
                 await queue.put(_event_payload("assistant.completed", {
                     "session_id": effective_session_id,
                     "message_id": message_id,
@@ -1617,6 +1618,7 @@ class APIServerAdapter(BasePlatformAdapter):
                     "session_id": effective_session_id,
                     "message_id": message_id,
                     "completed": True,
+                    "messages": turn_messages,
                     "usage": usage,
                 }))
             except Exception as exc:
@@ -3329,6 +3331,44 @@ class APIServerAdapter(BasePlatformAdapter):
             return len(prior)
         return 0
 
+    @classmethod
+    def _turn_transcript_messages(
+        cls,
+        conversation_history: List[Dict[str, Any]],
+        user_message: Any,
+        result: Dict[str, Any],
+    ) -> List[Dict[str, Any]]:
+        """Return this turn's assistant/tool messages in client-safe shape.
+
+        The streaming SSE contract delivers all assistant text as
+        ``assistant.delta`` events under one ``message_id`` interleaved with
+        ``tool.*`` events, and a single ``assistant.completed`` carrying only
+        the final reply.  A client that accumulates deltas into one buffer
+        cannot reconstruct *intermediate* assistant text segments that preceded
+        tool calls — so when the page is re-opened mid/post-stream those
+        segments appear lost, even though state.db persisted them correctly.
+
+        Emitting the authoritative per-turn transcript on ``run.completed`` lets
+        any SSE consumer reconcile its live view against ground truth without a
+        separate ``GET /messages`` round-trip.  Purely additive: clients that
+        ignore the field are unaffected.  Refs #34703.
+        """
+        agent_messages = result.get("messages") if isinstance(result, dict) else None
+        if not isinstance(agent_messages, list) or not agent_messages:
+            return []
+        start = cls._response_messages_turn_start_index(
+            conversation_history, user_message, result
+        )
+        turn = agent_messages[start:]
+        out: List[Dict[str, Any]] = []
+        for msg in turn:
+            if not isinstance(msg, dict):
+                continue
+            if msg.get("role") not in {"assistant", "tool"}:
+                continue
+            out.append(cls._message_response(msg))
+        return out
+
     @staticmethod
     def _extract_output_items(result: Dict[str, Any], start_index: int = 0) -> List[Dict[str, Any]]:
         """

tests/gateway/test_session_api.py

@@ -268,6 +268,75 @@ async def test_session_chat_stream_emits_lifecycle_events_and_keepalive_safe_sha
     assert "event: done" in body
 
 
+@pytest.mark.asyncio
+async def test_session_chat_stream_run_completed_carries_turn_transcript(adapter, session_db):
+    """run.completed must include the full interleaved turn transcript so a
+    client that lost intermediate (pre-tool-call) assistant text from the live
+    delta stream can reconcile without a separate /messages fetch. Refs #34703.
+    """
+    import json as _json
+
+    session_id = session_db.create_session("transcript-session", "api_server")
+
+    async def fake_run(**kwargs):
+        # Stream the intermediate planning text the way a real turn would.
+        kwargs["stream_delta_callback"]("Let me search for that:")
+        kwargs["stream_delta_callback"]("Here is the summary.")
+        result = {
+            "final_response": "Here is the summary.",
+            "session_id": session_id,
+            "messages": [
+                {"role": "user", "content": "search then summarize"},
+                {
+                    "role": "assistant",
+                    "content": "Let me search for that:",
+                    "tool_calls": [
+                        {
+                            "id": "call_1",
+                            "type": "function",
+                            "function": {"name": "web_search", "arguments": "{}"},
+                        }
+                    ],
+                },
+                {"role": "tool", "content": "results", "tool_call_id": "call_1", "tool_name": "web_search"},
+                {"role": "assistant", "content": "Here is the summary."},
+            ],
+        }
+        return result, {"total_tokens": 6}
+
+    app = _create_session_app(adapter)
+    with patch.object(adapter, "_run_agent", side_effect=fake_run):
+        async with TestClient(TestServer(app)) as cli:
+            resp = await cli.post(
+                f"/api/sessions/{session_id}/chat/stream",
+                json={"message": "search then summarize"},
+            )
+            assert resp.status == 200
+            body = await resp.text()
+
+    # Pull the run.completed event payload out of the SSE body.
+    run_completed_payload = None
+    for block in body.split("\n\n"):
+        if "event: run.completed" in block:
+            for line in block.splitlines():
+                if line.startswith("data: "):
+                    run_completed_payload = _json.loads(line[len("data: "):])
+            break
+    assert run_completed_payload is not None, body
+    messages = run_completed_payload.get("messages")
+    assert isinstance(messages, list) and messages, run_completed_payload
+
+    # The colon-ended intermediate text that preceded the tool call must be present.
+    contents = [m.get("content") for m in messages]
+    assert "Let me search for that:" in contents
+    assert "Here is the summary." in contents
+    # No prior-turn user message should leak into the per-turn slice.
+    assert all(m.get("role") in ("assistant", "tool") for m in messages)
+    # The tool call is preserved alongside the intermediate text.
+    assert any(m.get("tool_calls") for m in messages)
+
+
+
 @pytest.mark.asyncio
 async def test_session_endpoints_require_auth_when_key_configured(auth_adapter):
     app = _create_session_app(auth_adapter)