mirror of
https://github.com/rennf93/roboco.git
synced 2026-08-03 07:23:24 +02:00
fix(intake): emit the complete reply when no text deltas streamed (#665)
The intake/secretary chat driver treated StreamEvent text deltas as the ONLY text channel: AssistantMessage TextBlocks were always swallowed as already-streamed (the double-render guard). The CLI's partial-message emission turned out to be remotely gated — on 2026-07-23 the NAS containers got zero stream_event lines from the identical binary, flags, SDK, model, and settings that stream fine elsewhere — so the guard became a total blackout: replies were generated, the relay carried only init/status/turn_end, and the CEO saw nothing. SdkIntakeSession.send now tracks whether any text delta arrived during the turn; normalize() emits an AssistantMessage's complete text only when none did. Streaming mode is byte-identical (deltas render live, completes stay suppressed); gated mode delivers the reply as one block instead of nothing. Covers Secretary (same machinery). Regression tests: fallback emission, default suppression, and both modes end-to-end through the session layer. Co-authored-by: Renn F <rennf93@users.noreply.github.com>
This commit is contained in:
@@ -235,12 +235,20 @@ def _block_to_chunk(
|
|||||||
return None, None, None
|
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).
|
"""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 live token deltas (``StreamEvent``) already streamed it, so re-emitting
|
||||||
the complete ``TextBlock`` would render every reply twice on the panel.
|
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``**
|
The canonical draft signal is the agent calling the **``propose_draft``**
|
||||||
tool — that ToolUseBlock becomes a single ``draft`` chunk. As a fallback (if
|
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)
|
chunks.append(chunk)
|
||||||
if text_part is not None:
|
if text_part is not None:
|
||||||
text_parts.append(text_part)
|
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 block_draft
|
||||||
draft = draft or _extract_draft("".join(text_parts))
|
draft = draft or _extract_draft("".join(text_parts))
|
||||||
if draft is not None:
|
if draft is not None:
|
||||||
@@ -275,17 +285,23 @@ def _stream_event_to_chunks(msg: Any) -> list[StreamChunk]:
|
|||||||
return []
|
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.
|
"""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
|
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__
|
name = type(msg).__name__
|
||||||
if name == "StreamEvent":
|
if name == "StreamEvent":
|
||||||
return _stream_event_to_chunks(msg)
|
return _stream_event_to_chunks(msg)
|
||||||
if name == "AssistantMessage":
|
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":
|
if name == "ResultMessage":
|
||||||
return [
|
return [
|
||||||
StreamChunk(
|
StreamChunk(
|
||||||
@@ -617,6 +633,16 @@ class SdkIntakeSession: # pragma: no cover - requires the live claude binary
|
|||||||
|
|
||||||
async def send(self, text: str) -> AsyncIterator[StreamChunk]:
|
async def send(self, text: str) -> AsyncIterator[StreamChunk]:
|
||||||
await self._client.query(text)
|
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():
|
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
|
yield chunk
|
||||||
|
|||||||
@@ -16,6 +16,7 @@ import pytest
|
|||||||
from roboco.agent_sdk.intake_driver import (
|
from roboco.agent_sdk.intake_driver import (
|
||||||
_INTAKE_BASE_TOOLS,
|
_INTAKE_BASE_TOOLS,
|
||||||
IntakeDriver,
|
IntakeDriver,
|
||||||
|
SdkIntakeSession,
|
||||||
StreamChunk,
|
StreamChunk,
|
||||||
normalize,
|
normalize,
|
||||||
)
|
)
|
||||||
@@ -98,6 +99,63 @@ def test_normalize_assistant_message_blocks() -> None:
|
|||||||
assert chunks[1].data["input"] == {"file": "metrics.tsx"}
|
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:
|
def test_normalize_assistant_message_extracts_draft_block() -> None:
|
||||||
# A finished reply that ends with a fenced roboco-draft block yields a
|
# A finished reply that ends with a fenced roboco-draft block yields a
|
||||||
# single `draft` chunk carrying the parsed object — and no `text` chunk.
|
# single `draft` chunk carrying the parsed object — and no `text` chunk.
|
||||||
|
|||||||
Reference in New Issue
Block a user