perf(zara-027): stream LLM and agent output with cancellation-safe text events #283

Closed

nsaspy wants to merge 0 commits from perf/zara-027-streaming-llm-impl into master

nsaspy commented

2026-09-04 22:44:09 +00:00

Owner

Closes #28

Architecture summary

Implements ZARA-027 following the merged RAGE design (rage/28-streaming-llm-design.org, PR #227) from current master (ec89bd4):

Typed provider-neutral stream events (zara/agent/stream_events.py): frozen dataclasses text_delta, sentence_ready(text, is_final), tool_call_started(name, id), tool_result(name, id), completed(full_text), cancelled, failed(error_type). No SSE frames, raw tool protocol payloads, hidden reasoning, or provider metadata escape into the event contract.
Shared pure sentence chunker (zara/agent/sentence_chunker.py): sentence breaks on . ! ? … 。！？ with abbreviation guards (Dr., e.g., i.e., vs., St., initials), decimal guards, URL guards (scheme + www.), backtick code span/fence suppression, CJK breaks without whitespace, emoji grapheme safety. Bounded forced flushes: max_chars (180 default, last word boundary) and max_wait_ms (500 default, fake-clock tested). Conservation invariant (emitted + flush == fed text exactly) is property-tested.
Real streaming in the LangGraph agent node (zara/agent/graph.py): when a stream publisher is wired and the bound LLM exposes astream, the node consumes real provider chunks via a producer-task + bounded-wait queue (making the chunker's max-wait flush genuinely time-driven), publishes deltas before the response completes, and returns the exact aggregated AIMessage (content, additional_kwargs, tool_calls, id, response_metadata) so graph/history semantics are unchanged. No fake slicing of completed responses.

Actual streaming behavior

First text_delta is published while the provider stream is still open (proven inside the fake stream before it returns, test A1).
Runtime/UI surfaces receive incremental assistant text through the existing runtime event bus: the node publishes AssistantDelta via runtime_bridge.model_streaming with turn/conversation correlation, and populates AssistantComplete.text with the exact final text (codec already maps assistant.delta / assistant.completed; desktop ConversationService reduces them into streaming messages — e2e test proves final persisted message content is exact with no duplication).
sentence_ready flows through the typed stream_publisher contract (also now threaded through AgentManager.process_async) for #29 to consume.

Buffered fallback behavior

LLMs without astream (or nodes without a publisher) keep the exact buffered ainvoke path with a single honest completed(full_text) typed event and buffered_proxy=True latency records. No fabricated deltas.

Provider matrix

Agent path (LangChain): ChatAnthropic / ChatOpenAI / ChatOllama all stream via astream; OpenRouter continues to work through its configured family (no provider configuration redesign in this PR).
zara.llm.LLMClient.stream_events_async() (new; query_async intact): OpenAI-compatible SSE (data:/[DONE]), OpenRouter through the same OpenAI-compatible contract, Anthropic SSE (content_block_delta/message_stop/error), Ollama NDJSON. Single attempt (no retry duplication); terminals: completed (non-empty), cancelled (re-raised), failed (http, rate_limit, malformed_response, provider_error, empty_response, timeout, connection). All tested against local fake HTTP servers — no credentials, no network.

Cancellation semantics

Mid-stream cancellation emits exactly one cancelled typed terminal, re-raises CancelledError, closes the provider stream promptly (producer task cancelled, stream.aclose() awaited), and no later deltas or stale runtime events appear (A4).
Every stream ends in exactly one terminal: completed, cancelled, or failed — mutually exclusive, tested.

History-integrity proof

Aggregated assistant message == concatenation of all provider text chunks (A7, seeded 60-chunk property); tool-call fragments live only in tool_call_chunks and are rebuilt into parsed tool_calls; no protocol JSON in deltas, sentences, or persisted history; no orphaned partial history after ordinary completion; transient deltas never become independent messages (single aggregated AIMessage appended by the existing reducer).

Test/gate results

New: t/test_sentence_chunker.py (36 tests), t/test_llm_streaming.py (27 tests: A1–A9 incl. latency boundaries with fake clock, desktop-service e2e, AgentManager wiring), scripts/test-streaming-llm.sh (wired into test-all.sh phase 6).
Latency: llm_first_sentence event + llm_request_to_first_sentence stage added; scripts/benchmark-voice.py emits both first-token and first-sentence metrics; buffered_proxy=False on the genuine streaming path.
Full gate at candidate head c72a930: NIX_CONFIG=... nix develop -c bash scripts/test-all.sh → ALL PHASES PASSED (10/10), 1190 pytest tests passed / 5 skipped; nix flake check → all checks passed; nix build → success.
CI: GitHub Actions run for the exact PR head (see checks tab; no merge on stale CI).

Deviations from the existing RAGE design (evidence recorded in `rage/28-streaming-llm.org`)

sentence_ready is not mirrored onto the runtime event bus/wire protocol: the daemon codec rejects unmapped event types and wire semantics are frozen for this issue; it stays in the typed stream_publisher contract that #29 consumes (design point 4 ambiguity, recorded with rationale).
Constructor injection applied to stream_publisher only; latency_trace remains state-passed because the preserved wip/latency-trace-ctor branch never landed on master and every existing node test constructs create_agent_node(llm, registry) directly.

Closes #28 ## Architecture summary Implements ZARA-027 following the merged RAGE design (`rage/28-streaming-llm-design.org`, PR #227) from current master (`ec89bd4`): - **Typed provider-neutral stream events** (`zara/agent/stream_events.py`): frozen dataclasses `text_delta`, `sentence_ready(text, is_final)`, `tool_call_started(name, id)`, `tool_result(name, id)`, `completed(full_text)`, `cancelled`, `failed(error_type)`. No SSE frames, raw tool protocol payloads, hidden reasoning, or provider metadata escape into the event contract. - **Shared pure sentence chunker** (`zara/agent/sentence_chunker.py`): sentence breaks on `. ! ? … 。！？` with abbreviation guards (Dr., e.g., i.e., vs., St., initials), decimal guards, URL guards (scheme + `www.`), backtick code span/fence suppression, CJK breaks without whitespace, emoji grapheme safety. Bounded forced flushes: `max_chars` (180 default, last word boundary) and `max_wait_ms` (500 default, fake-clock tested). Conservation invariant (emitted + flush == fed text exactly) is property-tested. - **Real streaming in the LangGraph agent node** (`zara/agent/graph.py`): when a stream publisher is wired and the bound LLM exposes `astream`, the node consumes real provider chunks via a producer-task + bounded-wait queue (making the chunker's max-wait flush genuinely time-driven), publishes deltas before the response completes, and returns the exact aggregated `AIMessage` (content, additional_kwargs, tool_calls, id, response_metadata) so graph/history semantics are unchanged. No fake slicing of completed responses. ## Actual streaming behavior - First `text_delta` is published while the provider stream is still open (proven inside the fake stream before it returns, test A1). - Runtime/UI surfaces receive incremental assistant text through the **existing** runtime event bus: the node publishes `AssistantDelta` via `runtime_bridge.model_streaming` with turn/conversation correlation, and populates `AssistantComplete.text` with the exact final text (codec already maps `assistant.delta` / `assistant.completed`; desktop `ConversationService` reduces them into streaming messages — e2e test proves final persisted message content is exact with no duplication). - `sentence_ready` flows through the typed `stream_publisher` contract (also now threaded through `AgentManager.process_async`) for #29 to consume. ## Buffered fallback behavior - LLMs without `astream` (or nodes without a publisher) keep the exact buffered `ainvoke` path with a single honest `completed(full_text)` typed event and `buffered_proxy=True` latency records. No fabricated deltas. ## Provider matrix - Agent path (LangChain): ChatAnthropic / ChatOpenAI / ChatOllama all stream via `astream`; OpenRouter continues to work through its configured family (no provider configuration redesign in this PR). - `zara.llm.LLMClient.stream_events_async()` (new; `query_async` intact): OpenAI-compatible SSE (`data:`/`[DONE]`), OpenRouter through the same OpenAI-compatible contract, Anthropic SSE (`content_block_delta`/`message_stop`/`error`), Ollama NDJSON. Single attempt (no retry duplication); terminals: `completed` (non-empty), `cancelled` (re-raised), `failed` (`http`, `rate_limit`, `malformed_response`, `provider_error`, `empty_response`, `timeout`, `connection`). All tested against local fake HTTP servers — no credentials, no network. ## Cancellation semantics - Mid-stream cancellation emits exactly one `cancelled` typed terminal, re-raises `CancelledError`, closes the provider stream promptly (producer task cancelled, `stream.aclose()` awaited), and no later deltas or stale runtime events appear (A4). - Every stream ends in exactly one terminal: `completed`, `cancelled`, or `failed` — mutually exclusive, tested. ## History-integrity proof - Aggregated assistant message == concatenation of all provider text chunks (A7, seeded 60-chunk property); tool-call fragments live only in `tool_call_chunks` and are rebuilt into parsed `tool_calls`; no protocol JSON in deltas, sentences, or persisted history; no orphaned partial history after ordinary completion; transient deltas never become independent messages (single aggregated `AIMessage` appended by the existing reducer). ## Test/gate results - New: `t/test_sentence_chunker.py` (36 tests), `t/test_llm_streaming.py` (27 tests: A1–A9 incl. latency boundaries with fake clock, desktop-service e2e, AgentManager wiring), `scripts/test-streaming-llm.sh` (wired into `test-all.sh` phase 6). - Latency: `llm_first_sentence` event + `llm_request_to_first_sentence` stage added; `scripts/benchmark-voice.py` emits both first-token and first-sentence metrics; `buffered_proxy=False` on the genuine streaming path. - Full gate at candidate head `c72a930`: `NIX_CONFIG=... nix develop -c bash scripts/test-all.sh` → **ALL PHASES PASSED (10/10)**, 1190 pytest tests passed / 5 skipped; `nix flake check` → all checks passed; `nix build` → success. - CI: GitHub Actions run for the exact PR head (see checks tab; no merge on stale CI). ## Deviations from the existing RAGE design (evidence recorded in `rage/28-streaming-llm.org`) - `sentence_ready` is **not** mirrored onto the runtime event bus/wire protocol: the daemon codec rejects unmapped event types and wire semantics are frozen for this issue; it stays in the typed `stream_publisher` contract that #29 consumes (design point 4 ambiguity, recorded with rationale). - Constructor injection applied to `stream_publisher` only; `latency_trace` remains state-passed because the preserved `wip/latency-trace-ctor` branch never landed on master and every existing node test constructs `create_agent_node(llm, registry)` directly.

nsaspy closed this pull request

2026-09-04 23:09:02 +00:00

nsaspy referenced this pull request

2026-09-04 23:09:56 +00:00

RAGE #172: add Android Trealla semantic runtime lifecycle #257

nsaspy referenced this pull request

2026-09-04 23:09:58 +00:00

RAGE #172: add Android Trealla JNI bridge boundary #256

nsaspy referenced this pull request

2026-09-04 23:10:53 +00:00

RAGE #172: add Android Trealla JNI bridge boundary #304