fix(streaming): default to official astream output, fix duplicated Send parallel updates, restore tool/replay/interrupt

[LiangMuYuan]

Summary

Root cause (#48): develop used astream_events (event stream) to fake the updates stream. On Send-parallel graphs, astream_events emits two layers of events per node 鈥?the node's bare channel value (dict) and the root-level node wrapper (tuple). The run_executor on_chain_stream handler treated both layers as updates and sent them to the client, causing every node's updates to appear twice.

Fix: the default streaming path now uses the official graph.astream(stream_mode=[...]) 鈥?langgraph internally emits exactly one standard {"node_name": {...}} chunk per super-step / stream mode, so updates has a single source of truth and duplication is eliminated at the root. The migration also fixed the regressions it introduced:

• tool events: restored via the native tools stream mode (tool-started/tool-finished/tool-error, tool_name tracked by tool_call_id)
• replay endpoint: default GET /runs/{id}/stream now replays thread-level protocol events (values/updates/messages/tools), supports Last-Event-ID resume, and closes with end
• interrupts: aligned with official 鈥?when updates is not explicitly requested, interrupts are rewritten as values.__interrupt__ (parsable by the official SDK stream()); when updates is requested, the __interrupt__-bearing updates pass through; already-delivered interrupts are not re-emitted
• protocol command now forwards stream_modes/stream_subgraphs

How it fixes

• run_executor.py: default path switched to graph.astream(), consuming (mode, chunk) / (ns, mode, chunk) events; shares _handle_stream_mode / _handle_live_message; the events mode keeps the astream_events path
• interrupts: under _only_interrupt_updates, interrupt-bearing updates are rewritten as values.__interrupt__; explicitly requested updates pass through
• runs.py: replay endpoint adds thread-level protocol events + saw_interrupt dedup + seq/Last-Event-ID resume

Test changes

Tests were adapted and extended to cover the new execution path:

• unit (tests/unit/test_run_executor.py): adapted to the official astream path; added updates-no-duplication regression, interrupt-rewritten-to-values.__interrupt__ assertion, interrupt pass-through when updates requested; added coverage for the events mode / subgraphs triples / interrupt merge

• integration: test_runs_streaming.py tool assertions changed from legacy tool_start/tool_end to protocol tool-started/tool-finished; test_protocol_v2_streaming.py namespace test adapted to stream_mode

• live e2e (test_live_provider_api.py): store PUT assertion relaxed from ==200 to (200, 204) (official contract is 204); HITL streaming endpoint moved to /stream/events; added real-time messages/partial incremental accumulation end-to-end test (verifies token-by-token streaming that accumulates to the final answer)

• CI script (scripts/verify_docker_api.py): adapted to the replay endpoint's protocol-event format (tool events 鈫?tool-started/tool_name, end assertions 鈫?.get() to tolerate mixed payloads)

• reconnect regressions: HTTP-level mid-run disconnect/reconnect exactly-once tests for both executor backends (test_run_stream_midrun_reconnect_is_exactly_once for inline, test_run_stream_midrun_reconnect_is_exactly_once_in_redis for Redis); a protocol run.start invalid-stream-mode → 400 test (test_protocol_run_start_invalid_stream_mode_returns_400); a live-namespace-filter tuple regression (test_thread_protocol_stream_live_filter_rejects_tuple_and_accepts_list_namespace); and a real-HTTP reconnect assertion in scripts/verify_docker_api.py that runs in both the cli-docker and redis-durable CI jobs against the real store

• ci: verify_docker_api.py --mode full now also asserts run-stream mid-run reconnect exactly-once (real Redis + real SeekDB in redis-durable, real HTTP in both)

Reproduction (real Send-parallel graph, 11 nodes)

Scenario	develop	fix branch
stream_subgraphs=false	two updates per node (duplicated)	one updates per node (no dup)
stream_subgraphs=true	12 (root aggregate dropped, info missing)	13 (full node wrappers, includes aggregate)

Notes

This fix started because updates were duplicated, but the root cause was initially unclear and filtering attempts failed, so I switched to the official astream approach for alignment. Only after switching did I discover the old astream_events path has problems with stream_subgraphs both on and off: off duplicates, on drops the root aggregate node. The new astream path works correctly under both switches, resolving the issue as a side effect.

This change is large (run_executor.py overhaul + test refactor), and my understanding of langgraph's streaming internals is limited, so there may be oversights. I'd appreciate careful review of the streaming event formats, protocol v2 compatibility, and whether each channel (messages/updates/tools/values) behaves as expected.

One more note about a pre-existing test blind spot unrelated to this change: three protocol v2 issues (GET /threads/{id}/stream join endpoint missing, /stream/events not supporting subscribe-before-run.start, messages channel mixing in messages/partial) currently exist only in the live-provider e2e, and CI skips those tests because LIVE_PROVIDER_KIND is not configured 鈥?so they have never been covered by CI. These same three issues exist on develop; I'd suggest tracking them separately (or moving the LLM-independent protocol assertions into regular integration tests so CI can cover them). They are out of scope for this PR.

Test Plan

• tests/unit: 526 passed
• full tests/unit tests/integration: 794 passed, 5 skipped (coverage 90.45% 鈮?90%)
• tests/e2e (seekdb): 15 passed
• live provider e2e: 7 passed (streaming / create-time / store / HITL / MCP / real-time incremental, real model calls)
• full local CI replica: all 9 job categories and every matrix pass (incl. redis-durable, cli-docker, pgsql-metadata, mysql-family checkpoint 脳 3 backends)

Target Branch Check

• feature/* PRs target develop

Closes #48

[LiangMuYuan]

Hi @webup, could you help re-run the CI? The current run failed entirely due to GitHub Actions infrastructure issues, not code:

• 8 jobs failed at \Failed to resolve action download info. Service Unavailable\ (never reached checkout/code)
• 5 jobs failed with \The job was not acquired by Runner of type hosted\ (runner never started, no logs)

The jobs that did run all passed (Embedded SeekDB Smoke, Sample Graphs, Redis Durable Execution/seekdb). I've verified the full suite locally including the 90% coverage gate (90.19%), so a re-run should go green. Thanks!

[webup]

Thank you for the substantial work on this streaming migration—the switch to graph.astream() is a sound direction, and the focused tests and green CI provide good coverage. I found one merge-blocking replay defect: the default run stream now combines two independent sequence domains, so Last-Event-ID no longer represents a monotonic cursor. Please keep all emitted SSE ids in one cursor domain and add a regression showing that reconnecting after the terminal end does not replay previously delivered protocol events.

[LiangMuYuan]

Hi @webup, I've addressed the replay defect. The default \GET /runs/{id}/stream\ now assigns every emitted SSE frame a single monotonic cursor (1-based position in the ordered replay set of run lifecycle + thread protocol events + terminal end), so \Last-Event-ID\ is a valid cursor again and reconnecting after the terminal \end\ no longer replays previously delivered protocol events.

Added two regression tests:

• \ est_run_stream_sse_ids_are_monotonic: all SSE ids strictly increasing and unique
• \ est_run_stream_resume_after_terminal_end_does_not_replay: resume with the terminal frame's Last-Event-ID returns no frames

Full suite: 788 passed, 5 skipped, coverage 90.18% (>= 90%). Thanks!

[webup]

Thank you for the substantial follow-up and the added regression coverage. The astream() migration is directionally sound, and the duplicate-updates test is valuable. A full-range independent review plus targeted reproductions found that the current head still drops live protocol frames and uses an unstable response-position cursor; explicit events, live subgraph namespace filtering, the canonical provider workflow, invalid stream-mode validation, and the declared minimum LangGraph tool path also regress. Please address the inline findings together and add HTTP-level regressions for inline and Redis reconnects before approval. Hosted CI is green, but the affected paths are not covered by that run.

[webup]

[P1] Tail protocol events after the initial snapshot

This is the only read of values/updates/messages/tools. After it completes, both executor branches tail only run lifecycle records, even though the astream migration stopped publishing translated run-scoped protocol frames. Connecting while a run is active therefore returns start/end while omitting frames that are persisted later; I reproduced those missing values appearing only on a post-terminal request. Please persist or merge protocol and lifecycle frames into one run-scoped ordered log, tail it live for inline and Redis, and commit end only after earlier frames.

[webup]

[P1] Use a persisted cursor, not replay-list position

The replay set is mutable during active execution. I reproduced initial IDs start=1,end=2; after a protocol frame became visible, reconnecting with Last-Event-ID: 2 skipped that unseen frame (now position 2) and replayed end as ID 3. Assign the cursor at publication time in one run-scoped sequence, and add disconnect/reconnect mid-run exact-once tests.

[webup]

[P1] Keep the canonical provider workflow consistent

This replacement accepts a final values snapshot, but scripts/test-live-provider.sh still runs tests/integration/test_live_provider_streaming.py, which requires multiple default-stream message_chunk frames plus node_start/node_end. This head emits none; a deterministic equivalent fails, and no live-provider workflow run exists for this branch. Either preserve the default wire contract or update the canonical proof and workflow coherently while retaining a real incremental-provider assertion.

[webup]

[P1] Publish the requested events channel

When stream_mode=events, this branch translates selected messages/custom/values side effects but never publishes each raw astream_events() item onto the events channel. At the HTTP boundary, POST /runs/stream with stream_mode: events returns only metadata. Please emit raw events on the requested channel and add an HTTP-level regression.

[webup]

[P2] Return 400 for invalid stream modes

normalize_stream_modes() raises ValueError, but the shared handler below maps every ValueError to 404 for missing resources. I reproduced an invalid run.start stream mode returning invalid_argument with HTTP 404. Validate stream controls separately and return 400; reserve 404 for unknown assistants or graphs.

[webup]

[P1] Normalize live subgraph namespaces

Real astream(..., subgraphs=True) returns tuple namespaces. Passing ns through unchanged stores a tuple in the in-memory broker, whose live filter treats every non-list namespace as root. I reproduced a tuple-namespaced update disappearing when filtered by its real prefix, even though persistence can later mask the problem by JSON-normalizing it to a list. Convert ns to list(ns) before publication and test the live path.

[webup]

[P2] Honor the declared LangGraph minimum for tools

The project still allows langgraph>=1.0.3, but 1.0.3 has no langgraph.pregel._tools module. This fallback therefore disables the native tools mode, while the old astream_events tool translation has been removed, so allowed installations silently lose tool lifecycle events. Either raise the minimum to a verified version providing this mode or retain a functional fallback and test the minimum supported dependency.

[LiangMuYuan]

Thanks for the second round. All findings are addressed, each with a regression, including the HTTP-level inline + Redis reconnect tests you asked for.

• P1 - run stream in one sequence domain / tail protocol frames after snapshot / persisted cursor: protocol frames are appended to the run stream at publication time, sharing one run-scoped monotonic seq with lifecycle records, and GET /runs/{id}/stream reads that single domain. Added test_run_stream_sse_ids_are_monotonic and test_run_stream_resume_after_terminal_end_does_not_replay, plus HTTP-level mid-run disconnect/reconnect exactly-once tests for inline and Redis, and a real-HTTP reconnect assertion in scripts/verify_docker_api.py that runs in both the cli-docker and redis-durable CI jobs. Verified against a real Redis + SeekDB + worker stack (phase1 ids 1,2,3 -> reconnect 4..60, no replay, no loss).
• P1 - publish the requested events channel: each raw astream_events() item is now published onto the events channel; HTTP-level regression test_create_run_stream_events_mode_emits_raw_astream_events.
• P1 - normalize live subgraph namespaces: tuples are converted to lists before publication (run_executor.py), with test_execute_run_normalizes_tuple_namespaces_for_live_filter and a live-filter regression test_thread_protocol_stream_live_filter_rejects_tuple_and_accepts_list_namespace.
• P2 - 400 for invalid stream modes: protocol run.start now returns 400 invalid_argument; regression test_protocol_run_start_invalid_stream_mode_returns_400.
• P2 - LangGraph minimum: raised to >=1.2.0.
• P1 - canonical provider workflow: test_live_provider_streaming.py aligned with the replay contract; the real incremental-provider assertion is retained in test_live_provider_realtime_stream_incremental_messages_partial (verified against real DeepSeek).

Full suite: 794 passed, 5 skipped, coverage 90.45% (>= 90%).

[webup]

Thank you for the substantial follow-up. The raw events channel, live namespace normalization, invalid-mode 400 response, LangGraph minimum, and ordinary inline/Redis reconnect coverage are all meaningful improvements, and the exact-head regular CI is green.

A fresh exact-head review found three remaining merge blockers:

1. Resumed runs still violate the monotonic SSE cursor contract. The default renderer removes every historical end record from sequence order and emits it after newer resume frames. I reproduced IDs 1..9, 11..17, 10, 18. Please emit the persisted run log strictly by sequence and add resumed-run monotonic/reconnect regressions for inline and Redis.

2. Inline sequence allocation remains process-local. After clearing broker state before resuming the same persisted run, allocation restarted at 1, collided with existing rows, and the stream returned IDs 1..7, 9, 8, 10 with terminal statuses success, interrupted. Please allocate from persistent state, shared by lifecycle and protocol publication, and add a cold-broker resume regression that clears _next_seq as well as event state.

3. The canonical live-provider proof was weakened from incremental message_chunk assertions to a final values snapshot. The repository contract requires the manual workflow to prove real provider-backed SSE token chunks, and there is still no live-provider-streaming.yml run for this branch. Please restore the token-level assertion and run the manual workflow on the repaired exact head.

Focused streaming/replay verification passed 126 tests; the broad local suite reached 794 passed and 25 skipped, and the two localhost fixtures blocked by the sandbox passed separately. Please address these three findings together before approval.

[LiangMuYuan]

Thanks for the third round. All three findings are fixed on the current head (9bd73bc), each with a regression. The full CI for that head is green (including coverage >= 90%).

1. Resumed runs now keep the monotonic SSE cursor
The default renderer no longer defers historical end records past newer resume frames; it emits the persisted run log strictly by sequence, so every id stays ascending across a resume (previously 1..9, 11..17, 10, 18). test_resumed_run_stream_preserves_each_terminal_status now asserts strictly monotonic, unique ids.

2. Sequence allocation now comes from persistent state (inline)
next_run_stream_seq (run domain) and next_thread_stream_seq (thread domain) now allocate from the persisted max(seq)+1 instead of a process-local counter, and both brokers clamp their in-memory watermark so a cold broker re-seeded mid-run never reuses an id. Regression: test_run_stream_cold_broker_resume_keeps_monotonic_ids and test_thread_protocol_cold_broker_keeps_monotonic_seq_inline (both clear _next_seq and assert strictly monotonic ids with ordered terminal statuses). Verified by injecting the old behavior back (returning None / process-local) — both tests fail exactly as you reproduced.

3. Token-level live-provider assertion restored
test_live_provider_streaming.py now opens an explicit stream_mode=messages run and asserts real incremental messages/partial frames accumulate to the final answer (>=2 partial frames, no content-block noise), in addition to the default replay snapshot. Passed against real DeepSeek locally.

All three reverted-injection checks reproduce your exact reported failures, and the full suite is 796 passed / 5 skipped at 90.37% coverage on 9bd73bc.

[webup]

Thank you for the substantial follow-up. The strict resumed-run ordering, cold-broker sequence lookup, raw events publication, live namespace normalization, invalid-mode handling, and added regressions are meaningful improvements. I rechecked exact head 9bd73bc1731fd09005c492970bc25847cf6c6144; all 17 regular CI checks are green.

A knowledge-graph blast-radius review plus focused exact-head reproductions found four remaining merge blockers:

1. [P1] Non-Redis sequence allocation is not atomic. next_run_stream_seq() and next_thread_stream_seq() execute SELECT MAX(seq) + 1 separately from the eventual insert. Concurrent publishers can therefore receive the same sequence; I reproduced [1, 1] for both run and thread domains. The unique constraints then reject one insert, and the persistence helpers swallow that failure, dropping a stream frame. Please use a database-atomic append/allocation shared by lifecycle and protocol publication, and add concurrent-publisher regressions for the supported metadata databases.

2. [P1] A live cursor can be exposed before it is durable. The inline paths record an event in run_broker / thread_protocol_broker before committing it, while persist_run_stream_event() and persist_thread_stream_event() catch every database exception and return. A client can therefore receive an SSE id that disappears after broker loss or restart and may later be reused. Please make the durable append succeed before exposing the event (or surface/repair the failure), with a failure-injection test that reconnects after clearing broker state.

3. [P2] Id-less streamed messages collide across nodes/subgraphs. The fallback id is f"{run_id}:message:{message_index}", but message_index restarts for every yielded stream event. A focused reproduction with two id-less chunks from different subgraph namespaces produced only one messages/metadata identity; the second chunk was accumulated into the first message. Please derive a stable identity from the stream/node/namespace context or maintain a per-stream ordinal, and add a two-message regression.

4. [P1 verification gate] The canonical provider proof is still incomplete. The repository's manual provider workflow is the source of truth for real-provider SSE streaming and its proof target is incremental message_chunk events. The current tests assert messages/partial instead, and there are zero live-provider-streaming.yml runs for this exact SHA. Please make the workflow and asserted wire contract coherent, retain the real incremental-provider assertion, and run the manual workflow on the repaired head.

Please address these together before approval. For the first two items, an atomic append API that allocates and persists the sequence in one database operation would close both the concurrency and restart-consistency gaps more reliably than a split read/publish/write path.

[LiangMuYuan]

Thanks for the fourth round. All four items are addressed: items 1-3 are fixed and verified, item 4 is aligned on the wire contract with the workflow gate called out as needing a maintainer.

1. [P1] Non-Redis sequence allocation is now atomic.
Added a persistent StreamSequence counter table plus append_run_stream_event_atomic / append_thread_stream_event_atomic: row-lock + allocation + event insert happen in one transaction, with uniqueness-retry on collision. Verified on real MySQL 8.4 / PostgreSQL 16 / SQLite with 12 concurrent run + 12 concurrent thread appends each -> unique, gapless 1..12, zero dropped frames (new scripts/check_atomic_append_concurrency.py, regression test_stream_persistence_atomic.py). This also surfaced and fixed two MySQL-specific pitfalls: the gap-lock deadlock on a missing counter row, and REPEATABLE READ snapshot isolation hiding a just-seeded row.

2. [P1] A live cursor can no longer be exposed before it is durable.
Every inline publish path is now durable-before-expose: the event row (and its seq) is committed atomically first, and only then is it published to the in-memory broker. A failed durable append raises instead of being swallowed, so a client can never receive a seq that was not durably committed. Terminal events are staged in the same transaction as the run status and exposed only after commit. Regression: test_atomic_append_raises_on_db_failure_and_is_not_exposed (failure injection -> event not exposed).

3. [P2] Id-less streamed messages no longer collide across namespaces.
The fallback id changed from {run}:message:{index} (index restarts per yielded event) to {run}:message:{namespace}:{index}, deriving a stable identity from the subgraph namespace. Regression asserts two id-less messages from different namespaces produce two distinct messages/metadata identities; re-injecting the old behavior collapses them into one and the test reproduces exactly the "only one messages/metadata identity" you reported.

4. [P1] Canonical provider proof - wire contract aligned, workflow needs a maintainer.

• messages/metadata assertions added to test_live_provider_streaming.py: a real provider run must surface metadata identities, the payload shape is {message_id: {"metadata": {...}}}, and the metadata id must match the AI message id accumulated in messages/partial (verified against real DeepSeek).
• AGENTS.md proof target updated from the legacy message_chunk to the official v1 messages wire contract (messages/partial + messages/metadata/messages/complete). message_chunk does not exist in the official SDK wire format; langgraph_api/stream.py v1 path emits messages/metadata + messages/partial/messages/complete.
• Running the manual workflow requires a maintainer to configure the OPENAI_COMPAT_MODEL / OPENAI_COMPAT_BASE_URL repo variables and the OPENAI_COMPAT_API_KEY secret, then trigger live-provider-streaming.yml - as a contributor I do not have that permission.

Verification: full suite 800 passed / 5 skipped (90.08% coverage) on the latest HEAD, and the three-dialect concurrency script passes.

[webup]

Thank you for the substantial follow-up. The atomic metadata-DB append API, strict sequence replay, namespace-aware identity attempt, and expanded regression coverage are meaningful progress. I rechecked exact head 582e7b17b257f7ba0e329b874f53d04a51aa6cbe: all 17 ordinary CI checks are green, the broad local suite passed 802 tests with 25 skips after rerunning the two sandbox-blocked localhost tests, and the standalone SQLite concurrency check is gapless.

A fresh exact-head review with failure injection and adversarial concurrency checks found the following remaining merge blockers:

1. [P1] Redis append failures still expose non-durable events. src/agentseek_api/services/run_jobs.py:142-153 and src/agentseek_api/services/thread_protocol.py:227-271 catch atomic Redis append failures and then publish broker-local events. Injection produced visible sequence 1 events even though nothing was appended to Redis. Because /runs/{id}/stream merges broker snapshots, a client can observe a cursor that disappears after restart; with split workers, the frame is silently absent instead. Please propagate the append failure or suppress publication, and cover both run and thread paths with failure-injection/reconnect tests.

2. [P1] First appends can exhaust the metadata connection pool. src/agentseek_api/services/stream_persistence.py:219-230 checks for the counter row using the caller session, retaining that connection, and then opens a second seed session. Concurrent first appends for distinct stream IDs bypass the per-key lock relationship and can occupy all 10+5 configured connections while each waits for another. An exact-head reproduction with two distinct streams and a two-connection pool made both operations time out. Please seed before acquiring the caller connection or use a dialect-safe insert/upsert-and-lock design that needs only one connection.

3. [P1] An initial lifecycle append failure strands the run before submission. src/agentseek_api/services/run_preparation.py:268-270 calls run_started() and awaits the fail-fast durable lifecycle append before entering the submission recovery try. Injection left the committed run pending, the thread busy, the broker active-run count at one, and submitted no job. Please put lifecycle publication and submission under one compensation boundary so every failure calls run_finished() and persists consistent terminal run/thread state.

4. [P1] Id-less message identity is still incomplete. src/agentseek_api/services/run_executor.py:890-897 synthesizes an ID for messages/metadata, but messages/partial is serialized from the original id-less message at lines 952-974 and still carries id: null. Also, message_index restarts for each yielded event, so sequential root or same-namespace messages still reuse r1:message:0 and merge. Please assign one stable identity to the normalized message before generating any wire event, and test both different namespaces and sequential messages in the same/root namespace across metadata, partial, and complete frames.

5. [P2] SQLite in-session terminal appends still have an unhandled allocation race. The standalone _db_append() retries uniqueness collisions, but add_run_stream_event_to_session() and add_thread_stream_event_to_session() call _stage_db_event() directly. A forced two-session diagnostic allocated sequence 1 twice; one commit failed and only one terminal row persisted. Please either make the enclosing transaction retryable, retain the allocation lock through commit, or enforce and test a single-writer invariant.

6. [P1 verification gate] The canonical provider proof is still absent. There are zero Live Provider Streaming runs for this branch. This update changes AGENTS.md from the previously required message_chunk target to messages/partial; that is a contract decision, not execution evidence, and needs explicit maintainer acceptance. After agreeing on the wire contract, please run the canonical workflow on the repaired exact head and show incremental real-provider frames rather than only a final response.

These failures share one pattern: each local repair addresses the happy path but leaves an adjacent transaction, failure, or wire-contract boundary open. To avoid another patch cycle, I recommend pausing production changes and first posting a short design that defines these invariants:

• a sequence exists only after its durable append succeeds, and broker publication always follows it;
• the same append/publish contract applies to inline and Redis run/thread events;
• sequence initialization cannot require a nested connection while one is held;
• lifecycle and submission form one recoverable state transition with guaranteed cleanup;
• each logical message has one non-null identity across metadata, partial, and complete events;
• the real-provider acceptance contract is agreed explicitly and verified on the exact SHA.

Before revising the implementation, please add failing regressions for: Redis append failure with no broker event; a pool-capacity burst of distinct new streams; initial lifecycle failure cleanup; sequential id-less same/root-namespace messages; ID parity across wire events; concurrent in-session terminal appends; and the canonical exact-head provider workflow. Keeping each invariant in a separate commit with its exact test command/result will make the next review substantially faster and safer.

Please address these findings together before approval.

[LiangMuYuan]

Thanks for the fifth round. All six findings are confirmed against 582e7b1, and they share the pattern you flagged: each earlier fix closed the happy path but left an adjacent failure/transaction/wire boundary open. Per your suggestion I stopped changing code and wrote the design first — this is that design, for confirmation before implementation. I also self-reviewed it against the code before posting; the review surfaced and folded in one additional boundary (I8 below).

Core principle

A sequence and its event exist only after their durable append succeeds; broker publication always — and only — follows that success. The in-memory broker is a projection of durable state and never holds a seq the persistence layer does not.

Unified primitive

All publish paths delegate to one primitive instead of inline/Redis/lifecycle/protocol each doing their own:

append_and_publish(scope, scope_id, payload):
    seq = _durable_append(...)        # any failure raises
    _broker_publish(scope, ..., seq)  # only on success
    return seq

_broker_publish dispatches on scope × backend × event type (faithful to today's behavior — zero behavioral change):

scope	backend	event type	publish
run	inline	all	run_broker
run	Redis	lifecycle	run_broker (as today)
run	Redis	protocol	none (as today)
thread	both	all	thread_protocol_broker (with wire envelope)

Notes:

• In Redis mode, GET /runs/{id}/stream unconditionally merges run_broker snapshots (runs.py:926), but load_run_stream_events's Redis branch already reads every lifecycle frame back from the Redis stream (stream_persistence.py:503-505) and the live tail reads Redis directly (runs.py:949-951). Keeping the Redis-lifecycle broker copy breaks no invariant (it publishes only after success), and test_publish_run_event_uses_atomic_redis_append explicitly asserts it — this design does not change it. Protocol frames not publishing to run_broker is today's behavior, also unchanged. I1 touches only the failure path.
• A thread event carrying a run_id still first does a scope=run append to the run stream (_persist_protocol_to_run_stream) and then a scope=thread append to the thread stream before publishing the thread broker — this call order is today's behavior and the primitive does not change it; each append independently obeys "publish only on success".

The return int | None is intentional: None only in the offline/uninitialized-DB posture; production fails fast.

Invariants (one commit each, failing regression first)

#	Invariant	Your finding	Regression
I1	Redis append failure raises; no broker event is published	1	test_redis_run_append_failure_emits_no_broker_event, test_redis_thread_append_failure_emits_no_broker_event
I2	inline and Redis share the same append→publish contract (zero change on success path; only the failure path is aligned)	1	(folded into I1)
I3	counter seeding uses one connection and keeps max_seq self-healing	2	test_first_append_pool_burst_distinct_streams, test_counter_row_deleted_then_append_self_heals_from_max_seq
I4	lifecycle + submission form one compensation boundary; run_finished() is guaranteed even if publishing the failed event itself fails	3	test_initial_lifecycle_append_failure_cleans_up
I5	one logical message gets one non-null id written back to the object, spanning metadata/partial/complete	4	test_sequential_idless_same_namespace_messages_distinct_ids, test_idless_id_parity_across_wire_events
I6	in-session terminal append is retryable on unique collision (savepoint retry + counter-missing fallback)	5	test_concurrent_in_session_terminal_append (implementation detail: IntegrityError is raised at the exit of async with session.begin_nested(), so the catch must sit outside the block; SQLite concurrent writes can raise OperationalError: database is locked instead of a constraint violation, so the retry also handles that and sets busy_timeout, or the regression will be flaky)
I7	real-provider acceptance (see end of post)	6	live-provider-streaming.yml on the repaired head

I8 (new, from my self-review, lands first): execute_run_job's start/terminal publish points get the same compensation boundary. run_jobs.py:213-214 commits run=running/thread=busy and then publishes start outside the inner try — a failed start append leaves the run permanently running with no terminal state. The inline path already propagates today, so this is a pre-existing bug; once I1 lands, a Redis append failure enters the same uncompensated window. The two publish points are handled separately:

• start failure: execution_session was already committed at :213 with no dirty rows — no rollback needed; open a separate session and best-effort write run=error/thread=idle, then raise. Redis mode: the DB session is likewise committed, no rollback.
• terminal failure: the terminal UPDATEs (:245-251) are not yet committed, and the subsequent _stage_db_event query triggers an autoflush that sends those UPDATEs to the DB and holds their row locks — roll back the original execution_session first to release the locks, then best-effort write the terminal state from a separate session. Redis mode: the DB session carries no Redis transaction but still needs the DB rollback.
• Regression: test_execute_run_job_start_append_failure_persists_terminal.

Decisions to confirm

1. Redis append failure → propagate, not suppress. The run goes to error; no seq=None fallback publish. Applies uniformly to run lifecycle and thread events. The success path is unchanged (Redis lifecycle still publishes run_broker, Redis protocol frames still do not); I1 only touches the failure path.
2. In-session terminal append → savepoint retry. The run status and the end event stay in one transaction (the end event is not split out and written separately, which would open a "status committed, end missing" window). This is a localized version of the first option in your round-5 suggestion: instead of retrying the whole outer terminal transaction, we retry the sequence allocation inside the transaction via a savepoint, keeping run status and end atomic. Implementation notes: the collision is caught at the nested-transaction boundary (IntegrityError raised at async with exit, catch outside); SQLite may surface database is locked as OperationalError under concurrency, so the retry handles that too and sets busy_timeout.
3. add_*_to_session keeps its in-session semantics. The end event commits atomically with the run status and is published only after that commit; it is not folded into the primitive (the primitive is "append-then-publish", the terminal event is "commit-then-publish"; forcing it in would break the status/end atomicity).

Commit plan (order adjusted for regression safety — I8 first, so I1 does not introduce a known regression)

#	Commit	Regression
1	fix(streaming): execute_run_job publish compensation (I8; closes the pre-existing inline window first)	test_execute_run_job_start_append_failure_persists_terminal
2	fix(streaming): append failure never publishes (I1/I2; propagate on top of the I8 boundary)	test_redis_*_append_failure_emits_no_broker_event
3	fix(streaming): seed counter in caller session with max_seq (I3)	pool burst + self-heal
4	fix(api): lifecycle+submit one compensation boundary (I4)	test_initial_lifecycle_append_failure_cleans_up
5	fix(streaming): one id across metadata/partial/complete (I5)	sequential same-ns + parity
6	fix(streaming): retryable in-session terminal append (I6)	test_concurrent_in_session_terminal_append

Each commit message carries the exact test command and result; each commit adds the regression before the fix.

Open items for maintainer

• AGENTS.md's message_chunk → messages/partial needs your explicit acceptance as a contract decision.
• The live-provider-streaming.yml run record needs a maintainer to trigger it on the repaired head: the tests themselves run locally with LIVE_PROVIDER_KIND / LIVE_OPENAI_COMPAT_MODEL / LIVE_OPENAI_COMPAT_BASE_URL / LIVE_OPENAI_COMPAT_API_KEY set, and I will run the same suite locally as supporting evidence; the GitHub Actions run record requires a manual dispatch.

Does this set of invariants, the ordering, and the three decisions match what you expect? If so I'll start with I8.

[webup]

Thanks for stopping the patch cycle and writing the invariants first. The core principle is right: a broker is only a projection of durable state, and no event or cursor should become visible before its durable append succeeds. I also agree with landing the execute_run_job compensation work before making Redis failures propagate.

I am not yet confirming the design exactly as written, because several partial-success and transaction boundaries remain underspecified. Please revise the design around the following points; after these are explicit, proceeding one invariant and one red regression at a time is the right approach.

1. Define one terminal unit of work; do not use a state-only “best effort” fallback

I8 currently says that, after a terminal append failure, the original transaction is rolled back and terminal run/thread state is written from another session. That contradicts Decision 2, which correctly says terminal status and the end event must not be split. The fallback would create exactly that forbidden state: a terminal database row with no durable terminal event.

The Redis path has a larger cross-store version of the same problem. Today the effective order is:

1. mutate run/thread terminal state in the SQL session, without committing;
2. append and expose the run end record in Redis;
3. append and expose the thread lifecycle record in Redis;
4. commit SQL state.

This leaves at least three distinct failure windows:

• run end succeeds, thread lifecycle append fails;
• both Redis appends succeed, SQL commit fails;
• append fails, then the separate compensation write also fails.

In the first two cases, Redis can permanently report a terminal state that the metadata database does not share; the Redis record cannot be undone safely. append_and_publish() solves durable-before-broker for one event, but it cannot by itself make SQL state plus two Redis logs atomic.

Please define an explicit cross-store terminal state machine. My preferred direction is a transactional outbox: commit the run/thread state transition and outbox records in one SQL transaction, then have an idempotent dispatcher append the required Redis records and publish broker projections only after acknowledgement. An explicit terminal_pending plus idempotent reconciliation can also work, but the source of truth, retry key, recovery after restart, and client-visible behavior must be stated. A state-only best-effort compensation is not sufficient.

Please document the required converged outcome for each boundary:

Failure point	Required outcome
start durable append fails	no broker event; run/thread state is recoverable; active-run accounting returns to its prior value
run terminal append fails	no broker-only terminal cursor and no permanently running run
thread lifecycle append fails after run append	the two logs and metadata state converge through an idempotent retry/reconciliation record
final SQL commit fails after Redis appends	the durable recovery state identifies and repairs the disagreement; no duplicate terminal frames
compensation write fails	durable reconciliation remains pending rather than silently abandoning the run
broker publication fails after durable commit	replay from durable state still returns the committed event exactly once

Add failure-injection regressions for every boundary: start append, run terminal append, thread lifecycle append, SQL terminal commit, outbox/dispatcher retry, and compensation write. Each test should assert all three externally relevant views: database run/thread state, durable run/thread stream contents, and broker snapshots after restart/clear.

2. Retry the complete SQLite terminal transaction, not only a savepoint

The localized savepoint proposal does not fully handle SQLite contention. SQLAlchemy's begin_nested() flushes pending run/thread updates before the SAVEPOINT is established. If that pre-savepoint flush raises SQLITE_BUSY / database is locked, the outer transaction is already failed; retrying only sequence allocation inside the same session cannot restore it. busy_timeout reduces frequency but is not a correctness mechanism.

Please use a bounded whole-unit retry for recognized transient errors: roll back, reopen/reload the run and thread, recompute the intended terminal transition, allocate/stage both stream events, and commit the complete unit again. Use bounded backoff/jitter and do not retry arbitrary integrity or programming errors. Savepoints may still be useful for a narrowly scoped uniqueness collision, but there must be an outer-transaction fallback.

The regression should use two real sessions and a synchronization barrier, and assert more than “both calls returned”: no missing terminal record, no duplicate sequence, monotonic/gapless durable sequence, correct final run/thread state, and no broker publication before the successful commit. Run it on SQLite and the supported MySQL/OceanBase and PostgreSQL metadata backends.

3. Specify the logical-message identity algorithm, not only the output shape

Writing a generated ID onto one id-less chunk object does not explain how the next independently created chunk for the same logical model message finds and reuses that ID. Conversely, namespace plus a per-yield index still merges the next logical message in the same/root namespace.

Please define both the stream-lane key and the message-boundary rule. An explicit provider message ID should always win. The fallback needs a stable model-invocation/lane discriminator (for example provider run identity plus namespace/node/step where available) and a per-lane logical-message ordinal that advances only when a new message begins. Inject the chosen ID into a normalized/copy-on-write wire message before producing metadata, partial, complete, or tuple events; do not rely on mutating a provider object to carry state into later chunk objects.

Required regressions:

• three id-less chunks of one message reuse one non-null ID and accumulate correctly;
• the next id-less message in the same/root namespace gets a different ID;
• different subgraph namespaces remain distinct;
• metadata is emitted once per logical ID;
• every partial and the applicable complete/tuple frame use that same ID;
• both astream and astream_events paths satisfy the contract.

4. Make counter creation dialect-safe and cross-process safe

Moving counter seeding onto one connection addresses pool exhaustion, but “seed in the caller session” is not yet an algorithm. Two processes can still concurrently create the same missing counter, and deleted-counter recovery must never lower the counter below the durable MAX(seq).

Please specify the dialect-safe insert/upsert/lock sequence for SQLite, MySQL/OceanBase, and PostgreSQL. It must use one connection per append, tolerate a concurrent creator without aborting the outer unit of work, lock/read the winning row, and monotonically reconcile it with existing durable events. The in-process _stream_seed_locks cannot be part of the correctness proof because workers do not share it.

For example, PostgreSQL needs conflict-safe creation followed by locked allocation/RETURNING; MySQL/OceanBase needs duplicate-safe creation followed by row-locked allocation; SQLite needs conflict-safe creation plus whole-transaction retry where necessary. If two callers both observe a missing counter while durable MAX(seq)=10, the result must be allocations 11 and 12—the loser of counter creation must reload the winner's row rather than continuing from its stale maximum.

In addition to the distinct-stream pool-capacity test, add cross-process or independent-session tests for simultaneous first append to the same stream, simultaneous recovery after deleting the counter row while events remain, and mixed run/thread appends. Verify the real dialects in CI rather than relying only on SQLite.

5. Cover the protocol-to-run Redis path and partial dual-log success

I1 names run and thread append failure tests, but the previous finding also includes _persist_protocol_to_run_stream(). A thread event carrying run_id first writes the run log and then the thread log. Please define what happens when:

• the run-log append fails;
• the run-log append succeeds but the thread-log append fails;
• both durable appends succeed but broker publication fails.

At minimum, no broker event may be exposed until every durable write required by that API operation succeeds. If both Redis streams are required to represent one logical event, consider one Lua operation for the two appends; otherwise specify idempotent retry/reconciliation, the allowed temporary asymmetry, and a stable operation/event ID so a retry cannot duplicate the already-written run record. Add a regression for each partial-success point, not only lifecycle _publish_run_event().

6. Fence ambiguous queue submission and make cleanup unconditional

The lifecycle/submission boundary also has an ambiguous Redis outcome: LPUSH may succeed on the server and the client may still receive a timeout/disconnect. Marking the run failed immediately can then race a worker that already received the job. Please make submission idempotent or fenced: use a stable job identity derived from run_id, make worker claim/start a compare-and-set transition so duplicate deliveries cannot execute twice, distinguish “definitely not submitted” from “outcome unknown,” and retain a durable dispatch record until acceptance or reconciliation.

run_finished() and state cleanup should live in unconditional finally paths and must not depend on successfully publishing the compensating failed lifecycle event. Add tests for “queue accepted, client raised,” failed-event append failure, cleanup DB failure, and repeated compensation.

7. Refresh the branch and verification baseline before implementation

The PR still has no commit after 582e7b17b257f7ba0e329b874f53d04a51aa6cbe. Current develop is 1e1f980bf73209733974d274e7eaf80a50418257, and the PR is now CONFLICTING in pyproject.toml and uv.lock (langgraph>=1.0.6 on current develop versus this PR's >=1.2.0). Please rebase/merge current develop, resolve the minimum-version decision explicitly, regenerate uv.lock, and run the full ordinary/backend matrix on the resulting exact head. The existing 17 green checks are from August 12 and do not validate the current-base merge result.

The provider gate is also still open: there are zero Live Provider Streaming runs for this branch. The latest scheduled develop run currently fails because the provider returns HTTP 402 “account balance is insufficient,” so a maintainer must restore provider credit/configuration before an exact-head proof can succeed. I am not accepting the message_chunk → messages/partial contract change implicitly in this design review; resolve that contract explicitly, update the governing documentation/tests coherently, and then run the canonical workflow on the final repaired SHA. The proof should show multiple incremental frames, stable message identity across them, and reconstruction of the final answer.

What is approved directionally

• The durable-before-expose principle is correct.
• Redis append failure should propagate; there should be no seq=None broker fallback.
• I8 should land before enabling that propagation.
• Keeping inline terminal events in the same transaction as run/thread state is correct, provided the entire unit is retryable and publication follows commit.
• One focused invariant and its failing regression per commit is the right reviewable structure.

Please update the design with the terminal state machine, retry boundary, identity state machine, and dialect-specific counter-creation algorithm before starting implementation. Once those decisions are explicit, the proposed commit-by-commit approach should make the next review substantially more predictable.