diff --git a/roboco/agent_sdk/intake_driver.py b/roboco/agent_sdk/intake_driver.py index 2f69a581..db7f6f64 100644 --- a/roboco/agent_sdk/intake_driver.py +++ b/roboco/agent_sdk/intake_driver.py @@ -235,12 +235,20 @@ def _block_to_chunk( return None, None, None -def _blocks_to_chunks(content: list[Any]) -> list[StreamChunk]: +def _blocks_to_chunks( + content: list[Any], *, emit_text: bool = False +) -> list[StreamChunk]: """Map an assistant message's content blocks to chunks (duck-typed). - Text is deliberately NOT re-emitted here: with ``include_partial_messages`` + Text is normally NOT re-emitted here: with ``include_partial_messages`` the live token deltas (``StreamEvent``) already streamed it, so re-emitting the complete ``TextBlock`` would render every reply twice on the panel. + ``emit_text=True`` is the no-deltas fallback — the CLI's partial-message + emission is remotely gated and can silently vanish (2026-07-23 live + incident: zero ``stream_event`` lines in the NAS containers while the + identical binary streams elsewhere), and without this fallback the + double-render guard becomes a total blackout: replies are generated but + never reach the panel. The canonical draft signal is the agent calling the **``propose_draft``** tool — that ToolUseBlock becomes a single ``draft`` chunk. As a fallback (if @@ -257,6 +265,8 @@ def _blocks_to_chunks(content: list[Any]) -> list[StreamChunk]: chunks.append(chunk) if text_part is not None: text_parts.append(text_part) + if emit_text and text_part: + chunks.append(StreamChunk(kind="text", text=text_part)) draft = draft or block_draft draft = draft or _extract_draft("".join(text_parts)) if draft is not None: @@ -275,17 +285,23 @@ def _stream_event_to_chunks(msg: Any) -> list[StreamChunk]: return [] -def normalize(msg: Any) -> list[StreamChunk]: +def normalize(msg: Any, *, text_already_streamed: bool = True) -> list[StreamChunk]: """Map a single ``claude-agent-sdk`` message to panel-facing chunks. Duck-typed on type name + attributes so it works on real SDK messages and - on test fakes alike (no SDK import required). + on test fakes alike (no SDK import required). ``text_already_streamed`` + is the caller's per-turn "any text delta seen yet?" tracker: when False, + an AssistantMessage emits its complete text as chunks (the no-deltas + fallback — see ``_blocks_to_chunks``); the default preserves the + suppress-always behavior for callers that don't track. """ name = type(msg).__name__ if name == "StreamEvent": return _stream_event_to_chunks(msg) if name == "AssistantMessage": - return _blocks_to_chunks(getattr(msg, "content", [])) + return _blocks_to_chunks( + getattr(msg, "content", []), emit_text=not text_already_streamed + ) if name == "ResultMessage": return [ StreamChunk( @@ -617,6 +633,16 @@ class SdkIntakeSession: # pragma: no cover - requires the live claude binary async def send(self, text: str) -> AsyncIterator[StreamChunk]: await self._client.query(text) + # Per-turn delta tracker: once a live text delta arrives, complete + # AssistantMessage text stays suppressed (double-render guard); if + # none ever do (the remotely-gated no-partial-messages mode), the + # complete text is emitted instead so replies still reach the panel. + saw_text_delta = False async for msg in self._client.receive_response(): - for chunk in normalize(msg): + chunks = normalize(msg, text_already_streamed=saw_text_delta) + if type(msg).__name__ == "StreamEvent" and any( + c.kind == "text" for c in chunks + ): + saw_text_delta = True + for chunk in chunks: yield chunk diff --git a/tests/unit/agent_sdk/test_intake_driver.py b/tests/unit/agent_sdk/test_intake_driver.py index 0964dca4..64c7b006 100644 --- a/tests/unit/agent_sdk/test_intake_driver.py +++ b/tests/unit/agent_sdk/test_intake_driver.py @@ -16,6 +16,7 @@ import pytest from roboco.agent_sdk.intake_driver import ( _INTAKE_BASE_TOOLS, IntakeDriver, + SdkIntakeSession, StreamChunk, normalize, ) @@ -98,6 +99,63 @@ def test_normalize_assistant_message_blocks() -> None: assert chunks[1].data["input"] == {"file": "metrics.tsx"} +def test_normalize_assistant_message_emits_text_when_no_deltas_streamed() -> None: + """The no-deltas fallback (2026-07-23 incident): when the CLI's remotely + gated partial-message emission is off, no StreamEvent ever carries the + reply — the AssistantMessage's complete text must then be emitted instead + of being swallowed by the double-render guard.""" + msg = AssistantMessage([TextBlock("hello there"), ThinkingBlock("hmm")]) + chunks = normalize(msg, text_already_streamed=False) + assert [c.kind for c in chunks] == ["text", "thinking"] + assert chunks[0].text == "hello there" + + +def test_normalize_assistant_message_default_still_suppresses_text() -> None: + msg = AssistantMessage([TextBlock("hello there")]) + assert normalize(msg) == [] + + +@pytest.mark.asyncio +async def test_sdk_session_send_falls_back_to_complete_text_without_deltas() -> None: + """End-to-end through SdkIntakeSession.send: a turn with zero StreamEvents + yields the reply text from the AssistantMessage; a turn WITH deltas keeps + the complete text suppressed (no double render).""" + + class _FakeClient: + def __init__(self, messages: list) -> None: + self._messages = messages + + async def query(self, _text: str) -> None: ... + + async def receive_response(self) -> AsyncIterator[object]: + for m in self._messages: + yield m + + # Gated mode: no StreamEvents at all. + session = SdkIntakeSession(options=None) + session._client = _FakeClient( + [ + SystemMessage("init"), + AssistantMessage([TextBlock("the reply")]), + ResultMessage("s1"), + ] + ) + kinds = [(c.kind, c.text) async for c in session.send("hi")] + assert ("text", "the reply") in kinds + + # Streaming mode: deltas first — complete text stays suppressed. + session._client = _FakeClient( + [ + StreamEvent({"delta": {"type": "text_delta", "text": "the "}}), + StreamEvent({"delta": {"type": "text_delta", "text": "reply"}}), + AssistantMessage([TextBlock("the reply")]), + ResultMessage("s2"), + ] + ) + texts = [c.text async for c in session.send("hi") if c.kind == "text"] + assert texts == ["the ", "reply"] + + def test_normalize_assistant_message_extracts_draft_block() -> None: # A finished reply that ends with a fenced roboco-draft block yields a # single `draft` chunk carrying the parsed object — and no `text` chunk.