Split agent cascade streaming into separate streamer module#267
Merged
Conversation
`assembly live` answers each spoken turn with a deepagents graph, which
auto-installs langchain's TodoListMiddleware (the `write_todos` tool) — but
the plan it writes was invisible. Surface it so a multi-step spoken request
("book a flight, then check the weather there") shows a visible checklist
alongside the spoken "first I'll…, then I'll…", a recognizable
Gemini-Live-style affordance.
- plan.py: the write_todos plan subsystem — TodoItem/TodoUpdate, the
JSON-arg parsing, and TodoCollector, which reassembles a write_todos call
from streamed chunk args (streaming) or a non-streaming model's complete
.tool_calls dict.
- streamer detects the write_todos tool result and emits a TodoUpdate (never
a generic "Using write_todos" affordance); engine forwards it to
renderer.todos_updated without gating the spoken reply.
- Voice TUI renders an in-place TodoList checklist (✓ done, ▸ in progress,
○ pending), one panel per turn, revised in place. AgentRenderer emits a
`plan` NDJSON event / a stderr checklist on the non-TUI paths.
Also splits brain.py along its build-vs-drive seam: graph assembly stays in
brain.py, the per-turn streaming leg (build_streamer + helpers) moves to the
new streamer.py (one-directional: streamer imports brain). This keeps both
files under the 500-line gate.
Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0153dUtXjPNhfMvFDyjiLgdh
Reconcile the write_todos plan UI + brain/streamer split with main's agent-cascade changes (path-scoped --auto-write write_gate, --files on by default, SummarizationMiddleware context management, auto-added subagent, project_context grounding). Both feature sets are kept. Conflicts resolved: - brain.py: kept main's write_gate-based _graph_kwargs and register_gp_subagent_profile; kept the streaming-removed structure (streaming lives in streamer.py); dropped the brain-side flow-log constants (now in streamer.py). - aai_cli/AGENTS.md: merged both narratives; fixed brain._stream_* -> streamer._stream_* references. - tests/test_agent_cascade_brain.py, test_agent_cascade_subagents.py: unioned imports (plan, streamer, write_gate, subagents). Also: trimmed two engine.py docstrings to stay under the 500-line gate after the merge, and added two TodoCollector tests killing the and->or mutants the widened diff base surfaced (plan.py:100, :109). Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0153dUtXjPNhfMvFDyjiLgdh
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Refactors the agent cascade's per-turn streaming reply logic into a dedicated
streamermodule, splitting it frombrainalong the natural build-vs-drive seam. This keeps both files within the 500-line gate while maintaining a clean one-directional dependency (streamer imports brain, never the reverse).Key Changes
New module
aai_cli/agent_cascade/streamer.py: Contains all streaming/driving logic previously inbrain.py:build_streamer()— the streaming reply leg factory_stream_graph()— per-turn graph streaming with error handling_stream_gated()— write-approval pause/resume loop for--filesmode_pending_writes()/_decide()— approval decision handling_events_from_chunk()— chunk-to-event translation_FLOW_LOG,_RESULT_LOG_CAP,_clip(),_content_text())New module
aai_cli/agent_cascade/plan.py: Extracts plan/todo subsystem:TodoItem/TodoUpdatedataclasses for the agent's task listTodoCollector— accumulates streamedwrite_todosargs across chunks_parse_todos()/_todos_from_list()— JSON parsing for plan eventsWRITE_TODOS_TOOL_NAMEconstantUpdated
aai_cli/agent_cascade/brain.py:build_streamer()and all streaming functions (moved tostreamer)planandstreamer)SpeechDelta,ToolNotice,ApprovalPause)Updated
aai_cli/agent_cascade/messages.py:TodoListwidget for rendering the plan checklist_todos_markup()helper to style plan items (completed/in-progress/pending)Updated
aai_cli/agent_cascade/tui.py:todos_updated()renderer methodTodoListwidget mounting and refresh logic_todo_widgettracking for in-place plan revisionsUpdated
aai_cli/agent_cascade/engine.py:_consume()to handleplan.TodoUpdateevents_terminal_result()helper for cleaner event dispatchUpdated
aai_cli/agent/events.py:TodoItemandPlanUpdateevent types for JSON/text renderingUpdated
aai_cli/agent/render.py:_mark()helper for status-based checklist markersTest updates:
tests/test_live_tui_plan.py— TUI plan panel tests (500-line gate compliance)tests/test_tui_snapshots/test_live_plan.raw— snapshot golden for plan renderingtests/test_agent_cascade_streamer.py— tests forwrite_todos→TodoUpdateflowtests/test_agent_cascade_brain.py— adjusted imports, moved utility tests to streamertests/test_agent_render.py— added plan event rendering teststests/test_agent_cascade_engine.py— addedTodoUpdateevent handling teststests/test_agent_cascade_files.py,test_agent_cascade_subagents.py,test_agent_cascade_command.py— adjusted importsNotable Implementation Details
https://claude.ai/code/session_0153dUtXjPNhfMvFDyjiLgdh