diff --git a/Cargo.lock b/Cargo.lock index fdb4bc0..1e4a7e6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -5996,8 +5996,8 @@ dependencies = [ [[package]] name = "serdes-ai" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "futures", "serdes-ai-agent", @@ -6017,8 +6017,8 @@ dependencies = [ [[package]] name = "serdes-ai-agent" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "anyhow", "async-trait", @@ -6039,8 +6039,8 @@ dependencies = [ [[package]] name = "serdes-ai-core" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "anyhow", "base64", @@ -6058,8 +6058,8 @@ dependencies = [ [[package]] name = "serdes-ai-macros" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "darling", "proc-macro2", @@ -6069,8 +6069,8 @@ dependencies = [ [[package]] name = "serdes-ai-mcp" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "async-trait", "base64", @@ -6091,8 +6091,8 @@ dependencies = [ [[package]] name = "serdes-ai-models" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "anyhow", "async-trait", @@ -6119,8 +6119,8 @@ dependencies = [ [[package]] name = "serdes-ai-output" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "anyhow", "async-trait", @@ -6139,8 +6139,8 @@ dependencies = [ [[package]] name = "serdes-ai-providers" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "async-trait", "base64", @@ -6160,8 +6160,8 @@ dependencies = [ [[package]] name = "serdes-ai-responses" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "async-trait", "base64", @@ -6184,8 +6184,8 @@ dependencies = [ [[package]] name = "serdes-ai-retries" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "anyhow", "async-trait", @@ -6199,8 +6199,8 @@ dependencies = [ [[package]] name = "serdes-ai-streaming" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "async-trait", "bytes", @@ -6219,8 +6219,8 @@ dependencies = [ [[package]] name = "serdes-ai-tools" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "anyhow", "async-trait", @@ -6238,8 +6238,8 @@ dependencies = [ [[package]] name = "serdes-ai-toolsets" -version = "0.3.0" -source = "git+https://github.com/acoliver/serdesAI.git?rev=e675674ef802a0ac90c03cd8bcdbd5114fabb44f#e675674ef802a0ac90c03cd8bcdbd5114fabb44f" +version = "0.4.0" +source = "git+https://github.com/acoliver/serdesAI.git?rev=aa0eb8715065af55e98d098116d14964b0fcd7b0#aa0eb8715065af55e98d098116d14964b0fcd7b0" dependencies = [ "async-trait", "indexmap", diff --git a/Cargo.toml b/Cargo.toml index a7cb903..6195e6f 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -43,16 +43,16 @@ reqwest = { version = "0.12", default-features = false, features = ["json", "rus # rev. When #66 lands, flip all of these back to # https://github.com/janfeddersen-wq/serdesAI.git at the merged rev in one # commit; nothing else in this crate needs to change. -serdes-ai = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai", features = ["openai", "anthropic", "mcp"] } -serdes-ai-agent = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai-agent" } -serdes-ai-tools = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai-tools" } -serdes-ai-mcp = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai-mcp", features = ["client", "reqwest"] } -serdes-ai-models = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai-models" } -serdes-ai-core = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai-core" } +serdes-ai = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai", features = ["openai", "anthropic", "mcp"] } +serdes-ai-agent = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai-agent" } +serdes-ai-tools = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai-tools" } +serdes-ai-mcp = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai-mcp", features = ["client", "reqwest"] } +serdes-ai-models = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai-models" } +serdes-ai-core = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai-core" } # Responses protocol client (codex / Open Responses) over WebSocket, plus the # PKCE OAuth flow used to authenticate against the ChatGPT backend. -serdes-ai-responses = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai-responses" } -serdes-ai-providers = { git = "https://github.com/acoliver/serdesAI.git", rev = "e675674ef802a0ac90c03cd8bcdbd5114fabb44f", package = "serdes-ai-providers" } +serdes-ai-responses = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai-responses" } +serdes-ai-providers = { git = "https://github.com/acoliver/serdesAI.git", rev = "aa0eb8715065af55e98d098116d14964b0fcd7b0", package = "serdes-ai-providers" } urlencoding = "2.1" tiny_http = "0.12" once_cell = "1.19" diff --git a/project-plans/steering-integration/plan.md b/project-plans/steering-integration/plan.md new file mode 100644 index 0000000..85c4d8e --- /dev/null +++ b/project-plans/steering-integration/plan.md @@ -0,0 +1,639 @@ +# Plan: Mid-turn steering delivery at the tool-call boundary + +Plan ID: PLAN-20260905-STEERINT +Generated: 2026-09-05 +Base: origin/main 92b5bc3 (branch `steering-integration`) +Total Phases: 5 +Requirements: REQ-SI-001 .. REQ-SI-008 + +## Problem + +Issue #222 shipped steering with LOCAL TURN CHAINING: a steer typed during a +running turn waits in `steering_queues` and is delivered by persisting it and +running a whole new `AgentStream` turn after the current one finishes +(`src/services/chat_impl/streaming/steering_delivery.rs:121-175`). The user +waits for the entire turn to end before the model hears the correction. + +Upstream `acoliver/serdesAI` now supports true mid-turn delivery. Commit +`c0e7327a880f8d3013bb47c56c2fb231e3e93264` (branch `feature/steering-input`, +read at `/Volumes/XS1000/acoliver/projects/serdesAI/branch-steering`) adds to +crate `serdes-ai-agent`: + +- `SteeringQueue` (`serdes-ai-agent/src/steering.rs`): `Clone + Default`, + `new()`, `steer(String) -> bool` (false only when every receiver is dropped + and none can be claimed again), `pending_len() -> usize`. Backed by a tokio + unbounded mpsc. A run claims the single receiver via + `take_receiver()`; `SteeringReceiver::Drop` PARKS IT BACK on the queue, so + undelivered texts survive a run and a later run from the SAME queue delivers + them at its first tool boundary. Drain is FIFO by channel order. +- `RunOptions::steering(queue)` (`serdes-ai-agent/src/run.rs`), mirroring + `message_history`. +- `AgentStreamEvent::SteeringDelivered { step: u32, text: String }` + (`serdes-ai-agent/src/stream.rs:100-112`), emitted once per delivered text by + `deliver_queued_steering` (`stream.rs:214-233`), AFTER that step's + `ToolExecuted` events and BEFORE the next `RequestStart`. Steps are 1-based. + The drained text is appended to the run's history as its own `ModelRequest` + carrying a user prompt part. +- Drain happens inside the spawned stream task at the tool-call boundary in + BOTH `AgentStream::new` (`stream.rs:913`) and `AgentStream::new_with_cancel` + (`stream.rs:1530`). Nothing drains before the first model request. Nothing + drains at a text-only end of turn; leftovers stay queued. + +This plan moves PA delivery from between-turns to that boundary. + +## PA call sites this touches (read, verified) + +- `src/llm/client_agent.rs:475-534` `handle_agent_stream_event` ends with + `other => tracing::debug!(...)` at line 532. `SteeringDelivered` will COMPILE + against the new rev and be SILENTLY SWALLOWED unless an explicit arm is + added. The compiler will not warn. Phase P02 writes a regression test that + fails if the arm is missing. +- `src/llm/client_agent.rs:558-562` builds `RunOptions`, then line 565 calls + `AgentStream::new(agent, UserContent::text(prompt), context, options)`. +- `src/llm/stream.rs:124-128` is a second, legacy path + (`send_message_stream`, `run_stream_with_options`). It has no production + caller (only the re-export at `src/llm/mod.rs:21`), no service context, no + event-bus access, and no queue to correlate against. It gets NO steering. + Justification: attaching a queue requires a delivery sink that can persist + and announce; this path has neither. With no queue attached the fork drains + nothing, so its behavior is byte-identical after the rev bump. Its + `_ => ChatStreamEvent::text(String::new())` catch-all (`stream.rs:139-142`) + is unreachable for `SteeringDelivered` for the same reason. +- `src/services/chat_impl.rs:66-76` `ActiveStream` (per-conversation slot: + stream_id, task handle, cancel token, lifecycle state). +- `src/services/chat_impl.rs:245-266` `begin_stream` creates the slot + atomically; `src/services/chat_impl.rs:320-372` `cancel_active_stream` fires + the token, ABORTS the task handle, drains the PA steering queue, and emits + `SteeringDiscarded` for each entry. +- `src/services/chat_impl/steering.rs` PA queue: cap at + `MAX_QUEUED_STEERING_MESSAGES = 5` (line 26), `queue_steering` (153-197), + `confirm_or_withdraw_steering` (237-249), `push_steering` (263-274), + `remove_steering` (293-304), `drain_steering` (311-313), discard emitters + (104-124). Lock discipline: `active_streams` and `steering_queues` never + held simultaneously, nothing emitted while either is held (37-56). +- `src/services/chat_impl/streaming.rs`: `stream_agent_response` (67-100) + calls `client.run_agent_stream(...)` with a sync `FnMut(StreamEvent)` + callback; `StreamTranscript` (31-43); `run_stream_task` (317-407) owns the + chain; `handle_llm_stream_event` (409-460) is sync; `clear_streaming_state` + (753-776) drains + discards PA-queue leftovers whenever the send's slot is + released. +- `src/services/chat_impl.rs:486-568` `build_llm_messages` rebuilds turn input + from persisted rows in insertion order. Insertion order IS reload order. +- `src/events/types.rs:470/480/496` `SteeringQueued` / `SteeringDelivered` / + `SteeringDiscarded`; the view removes the waiting bubble on + `SteeringDelivered` (`src/ui_gpui/views/chat_view/command.rs:191-204`). + +## THE CENTRAL DESIGN CONFLICT: when does a delivered steer reach the database + +Today `deliver_steering` (`steering_delivery.rs:205-238`) persists with +`add_message` FIRST and emits `SteeringDelivered` only after the write +succeeds; on failure it stops the chain and announces `SteeringDiscarded` for +that message and everything behind it (`steering_delivery.rs:211-227`). The +model is never shown text the database rejected. That ordering is available +only because PA controls the sequence between turns. + +Mid-turn, the ordering inverts. The fork appends the text to the model request +inside its spawned task (`stream.rs:214-233`) and PA learns about it +afterwards, via `SteeringDelivered`. By then the model has seen it. "Not +seeded, not announced" is structurally unachievable at this boundary. + +### Option (a): persist at steer-acceptance time, before `queue.steer(text)` + +The row is durable before the text can reach the model, and the model can +never see unpersisted text, restoring today's guarantee one level up. + +Rejected, for three reasons. + +1. It persists rows the model may never see. A steer accepted during a + text-only tail, or cancelled before any boundary, stays in the database + with its UI bubble withdrawn as discarded. #222 established that a + discarded steer leaves no trace; this option makes every discard a + deletion, and deleting a user row makes the transcript lie about what was + typed. +2. Delivery then needs "already persisted" bookkeeping so `add_message` is not + called twice, and the row sits at acceptance-order position even when the + model saw it later (or never). +3. Stop semantics rot: `cancel_active_stream` discards queued steers, but the + rows would already exist, so cancel would need row deletion to stay + consistent with the announced discard. + +### Option (b): persist on receipt of `SteeringDelivered` + +PA writes the row exactly when it learns the text reached the model. The +database reflects reality: the model saw it if and only if it is persisted, +modulo the failure window. Correlation and ordering stay with the delivery +event. Cost: a persist failure happens after the model saw the text. + +### Option (c): option (b), and the persist failure fails the turn (RECOMMENDED) + +(b) leaves the failure response undefined. The house style is fail fast, so +define the single behavior: when `add_message` fails for a delivered steer, PA +fails the stream immediately. Concretely: the sink returns `Err`, the event +loop maps it to `StreamEvent::Error` plus a returned `LlmError` (same shape as +the existing mid-stream failure path, `client_agent.rs:573-585` and +`streaming.rs:111-130`), `transcript.error` is set, and finalization runs the +interrupted path (`finalize_by_outcome`, `streaming.rs:263-277`, persisting +partial output marked interrupted). The failed steer's own bubble gets +`SteeringDiscarded` (its truthful terminal: not recorded), entries behind it +stay queued and are discarded by the send's teardown (`clear_streaming_state`, +`streaming.rs:753-776`). No retries, no swallowing, no guard layers. + +**Failure mode of the recommendation, stated plainly.** If the database fails +at the moment of delivery, the model has already seen text that is not in the +store, and that divergence is permanent: the aborted turn's continuation dies +with it, the next send re-seeds from the database without the lost text, and +nothing can recover it. PA announces the divergence twice (stream error, steer +discarded) and never pretends it did not happen. This is strictly narrower +than option (a)'s failure mode, where the database silently accumulates +instructions the model never received and cancellation must delete user rows +to hide them. + +### Reload ordering (the second half of the conflict) + +#222 kept this invariant: persisted history equals the history a chained turn +is seeded with, so a reload reconstructs what the model saw. + +Within one mid-turn-steered run the fork's history is `[user prompt, assistant +partial + tool calls, tool returns, steer text, continuation...]`. PA's +persistence model is already coarser than that: ONE assistant row per turn +holding the accumulated response text plus all tool calls and results +(`persist_assistant_response`, `streaming.rs:660`; reloaded by +`build_llm_messages:524-557`). The per-step interleaving of partial text and +tool calls inside a turn is already collapsed by that design and is not +reconstructed on reload today. + +The order PA can and must preserve is the order of PERSISTED ROWS, and +insertion order gives the right one with zero schema or `build_llm_messages` +changes: + +1. user row for the turn prompt: written at send start + (`prepare_message_context`, `chat_impl.rs:389-391`); +2. steer rows: written by the sink at `SteeringDelivered`, mid-turn; +3. assistant row for the turn: written at end of turn (chain persist or + finalization). + +A reloaded conversation reads `[turn prompt, steers that shaped the answer, +the answer]`, which is the order things actually happened at the granularity +PA persists. `build_llm_messages` (`chat_impl.rs:486-568`) iterates rows in +insertion order and maps user rows to `LlmMessage::user`, so reload and seed +agree with NO change to that function. + +To keep the seeded history equal to the reload on chained turns, the chain +must re-insert each turn's delivered steer texts into its in-memory `messages` +BEFORE the assistant message push (mirroring the row order). The sink records +delivered texts in order on the transcript; `run_steered_turns` splices them +in before `assistant_message(&transcript)` (the push currently at +`steering_delivery.rs:167-169`). REQ-SI-004 pins this. + +## Two queues, one relationship + +- PA's service-level queue (`SteeringQueues`, `VecDeque` with + `id` + `text`) remains the STATE MACHINE: acceptance cap (5), UI bubbles via + `SteeringQueued`, terminal events, withdrawal, cancel/teardown discard. It + is the source of truth for "accepted but not yet resolved". +- The fork's `SteeringQueue` is the TRANSPORT: it and only it decides when a + text reaches the model. PA never reads it back, never drains it, and never + seeds from it. + +Write path: acceptance pushes to the PA queue first, announces +`SteeringQueued`, passes `confirm_or_withdraw_steering`, and only then calls +`steer(text)` on the fork queue. The fork `steer()` call is the COMMIT POINT: +once it returns true, PA no longer withdraws the entry (it can only be +resolved by delivery, teardown discard, or cancel discard). Calling `steer()` +before the re-check would let a withdrawn text linger in the transport and +deliver as a ghost; the ordering above closes that. + +Read path: on `AgentStreamEvent::SteeringDelivered { text, .. }` the sink pops +the HEAD of the conversation's PA queue and treats it as the delivered entry. + +**Correlation is by FIFO position, never by text.** The fork drains FIFO from +a channel; PA pushes in the same order it pushes to its own deque; exactly one +`AgentStream` per send holds the receiver at any moment (the chain is +sequential), so deliveries arrive in exactly the PA queue's order. Two +identical texts occupy positions k and k+1 on both sides and resolve +correctly. A duplicate `steer_id` match or a text match would break here; a +position match cannot. + +Residual races, each with exactly one defined behavior: + +- Cancel/teardown drained the PA queue between the fork's drain and the sink's + pop (empty deque on delivery): the model DID see the text, so the sink + still persists the row, records it for chain seeding, and emits no UI + terminal (the discard already fired; the at-least-one-terminal guarantee of + #222 is intact). Bubble count: zero, correctly, because cancel announced it. +- `steer()` returns false at commit time (run gone between the re-check and + the commit): same refusal shape as `confirm_or_withdraw_steering`: remove + the entry, emit `SteeringDiscarded`, return the same + `no_active_turn()` validation error. +- A steer accepted while a delivery is being persisted: goes to the TAIL of + both queues; delivery order is unaffected. + +## Queue lifetime + +ONE `SteeringQueue` per SEND, not per turn. It is created in `begin_stream`, +stored on the conversation's `ActiveStream` entry, and dropped with that entry +in `clear_streaming_state` / `cancel_active_stream`. Every chained turn of the +send passes the same clone via `RunOptions::steering`. + +The park-back-on-drop rule makes this the only correct lifetime: a fresh queue +per turn would strand leftovers forever (turn N+1 would never see turn N's +undelivered texts), while reusing one queue delivers them at turn N+1's first +tool boundary, which is exactly the leftover path the chaining exists to +provide. + +**Double-delivery is resolved by deletion, not by guards.** Today the chain +drains the PA queue at the turn boundary and seeds the texts into the next +turn's history (`drain_steering_queue` at `steering_delivery.rs:143`, push at +`steering_delivery.rs:235`). If that survived alongside the transport, a +leftover would be BOTH seeded into turn N+1's history by PA AND delivered +inside turn N+1 by the fork. The PA drain-and-seed path is deleted. The fork +queue is the only thing that moves text into a model request. + +Cross-send leakage is impossible: the next user send creates a new +`ActiveStream` and a new queue; the old queue dies with the old slot, and any +texts still parked in it were already announced as discarded by the teardown +(`clear_streaming_state` drains the PA queue and emits discards; the parked +copies in the dropped transport vanish with it, model never having seen them). + +## What remains of end-of-turn chaining + +- STAYS: the loop in `run_steered_turns`; the clean-turn precondition + `reaches_delivery_boundary` (`steering_delivery.rs:186-188`: completed, no + error, not cancelled); intermediate-turn persistence + (`persist_assistant_response` at `steering_delivery.rs:159-166`) so a chained + turn's seed matches the reload; finalization once, on the last turn; the + turn cap; the discard announcements on every ending that leaves entries + queued. +- CHANGES: the loop's leftover check becomes a PEEK + (`has_queued_steering(queues, conversation_id)`), not + `drain_steering_queue`, because the transport still owns the texts and must + deliver them itself. The cap branch removes the entries and announces them + (they will never be delivered within this send; the dropped transport takes + its parked copies away at teardown). +- DELETED: `deliver_steering` (`steering_delivery.rs:205-238`) and the + `LlmMessage::user` push into the next turn's seed; the + `ChainOutcome::OutputPersisted` variant (`steering_delivery.rs:108-113`) + and the `deliver_steering -> false` exit that produced it + (`steering_delivery.rs:171-173`). With delivery inside the turn, there is no + step between a turn's persist and the next turn's start that can fail, so + `run_steered_turns_and_finalize` always ends in `finalize_by_outcome`. + `ChainOutcome` collapses to a boolean or disappears with the match. +- `MAX_STEERING_TURNS = 10` STAYS, with unchanged semantics: it counts + `run_turn` invocations (initial turn included, `steering_delivery.rs:131, + 137, 148`) and is still the only bound on the refill loop, since the queue + cap bounds depth but a user can refill during every follow-up turn. A + text-only leftover turn that triggers another text-only leftover turn still + consumes the cap and then discards, exactly like today. + +## Cancellation + +PA uses `AgentStream::new` (`client_agent.rs:565`), not `new_with_cancel`. +`create_stream_agent` receives the token as `_cancel` and ignores it +(`chat_impl.rs:761`). Stop works at the PA level: `cancel_active_stream` +(`chat_impl.rs:320-331`) fires the `CancellationToken` and ABORTS the spawned +stream task; the token's remaining uses are the chain preconditions +(`steering_delivery.rs:186-188`). + +No constructor change. Both fork constructors drain at the boundary, so +staying on `new` keeps delivery behavior identical while preserving the abort +semantics Stop depends on. Defined cancel interactions: + +- Steer still parked in the transport at abort: announced discarded by + `cancel_active_stream`, row never written, transport dropped with the send. + Same as today's discarded-but-never-delivered entries. +- Fork drained a text and the sink persisted it, THEN the user hits Stop: the + row stays. The model saw it mid-turn and it is in the store; only the + turn's assistant output is discarded, which is today's cancelled-turn + behavior for output. +- Abort lands between the fork's drain and the sink's persist: the text was + inside the fork task's in-memory history for a turn whose output is being + discarded anyway. PA announces the discard (the cancel path runs + unconditionally) and writes no row. No guard layer is added; a stopped turn + is the terminal state of everything it was holding. + +## Requirements + +### REQ-SI-001: Steering is delivered at the tool-call boundary inside the running turn + +**Behavior** +- GIVEN a running turn that executes at least one tool +- WHEN the user steers while a tool is in flight +- THEN the text reaches the model in that same turn, appended after that + step's tool returns and before the next model request +- AND PA observes it via `AgentStreamEvent::SteeringDelivered` +- AND no new user-initiated send and no chained turn is needed for it. + +### REQ-SI-002: One `SteeringQueue` per send, shared by all its chained turns + +**Behavior** +- GIVEN a send with queued steering that a turn ended without delivering +- WHEN the chain runs the next turn +- THEN the same queue is attached and the leftover is delivered at that + turn's first tool boundary +- AND the leftover is never also seeded into the turn's history by PA. + +### REQ-SI-003: Delivery persists, and persist failure fails the turn + +**Behavior** +- GIVEN a `SteeringDelivered` event carrying a text +- WHEN the sink receives it +- THEN PA persists the text as a user message and then emits + `ChatEvent::SteeringDelivered { steer_id }` for the correlated entry +- AND when `add_message` fails, the stream fails with a reported error, the + entry is announced `SteeringDiscarded`, and finalization takes the + interrupted path. + +### REQ-SI-004: Reload order equals seed order equals row order + +**Behavior** +- GIVEN a turn that received steering mid-turn and then produced output +- WHEN the conversation is reloaded and its next send is built +- THEN the history reads `[turn prompt, steer texts, turn assistant output]` +- AND the persisted rows are in exactly that insertion order +- AND `build_llm_messages` is unchanged. + +### REQ-SI-005: Correlation is by FIFO position, not by text + +**Behavior** +- GIVEN two queued steers with identical text +- WHEN both are delivered +- THEN each `SteeringDelivered` resolves the PA-queue head, so each `steer_id` + gets exactly one delivered terminal event, in queue order. + +### REQ-SI-006: Text-only ends of turn still chain for leftovers, bounded + +**Behavior** +- GIVEN a turn that finished cleanly without crossing a tool boundary while + steers are still queued +- WHEN the chain decides +- THEN it runs another turn on the same queue (the fork delivers there) +- AND when the chain hits `MAX_STEERING_TURNS` it announces every remaining + entry `SteeringDiscarded` and stops. + +### REQ-SI-007: The #222 invariants survive + +**Behavior** +- GIVEN any steering traffic during a turn +- THEN steering never cancels or disturbs the generation, never touches the + cancellation token, and never resolves an approval +- AND `MAX_QUEUED_STEERING_MESSAGES = 5` and `MAX_STEERING_TURNS = 10` still + bind +- AND every `SteeringQueued` is followed by at least one terminal event +- AND queued transcript entries remain chrome +- AND `active_streams` and `steering_queues` are never held simultaneously + and nothing is emitted while either is held. + +### REQ-SI-008: Stop semantics are unchanged + +**Behavior** +- GIVEN a running turn with or without steering traffic +- WHEN the user presses Stop +- THEN the task is aborted and the token fires exactly as today +- AND steers already persisted before the abort keep their rows +- AND steers never delivered are announced `SteeringDiscarded`. + +## Phases + +### P01: Move the dependency pins to the steering rev + +Plan ID `PLAN-20260905-STEERINT.P01` + +Files: `Cargo.toml`, `Cargo.lock` + +- Change `rev` on all EIGHT `serdes-ai-*` entries (`Cargo.toml:46-55`) from + `e675674ef802a0ac90c03cd8bcdbd5114fabb44f` to + `c0e7327a880f8d3013bb47c56c2fb231e3e93264` in ONE commit. Cargo resolves one + source per package name and `serdes-ai-responses` exists only on the fork, + so partial moves do not resolve. +- Verified upstream delta (`git log/diff e675674..c0e7327` in the fork): one + commit touching only `serdes-ai-agent` (steering.rs, stream.rs, run.rs, + lib.rs) plus workspace Cargo files. No other crate's behavior changes. +- No PA source changes in this phase. Nothing references `SteeringQueue` yet; + the fork drains nothing when no queue is attached. + +Tests: +- `cargo tree -i serdes-ai-agent` shows a single source at `c0e7327` +- full existing suites pass unchanged (empty-queue behavior is byte-identical) + +### P02: Steering channel in the LLM layer, with the explicit event arm (test-first) + +Plan ID `PLAN-20260905-STEERINT.P02` + +Prerequisites: P01 (the event variant and `RunOptions::steering` must exist). + +Files: new `src/llm/steering.rs`; `src/llm/mod.rs`; `src/llm/client_agent.rs` + +- `src/llm/steering.rs`: `SteeringDeliverySink` trait (`#[async_trait]`, one + method `delivered(&mut self, text: &str) -> StdResult<(), LlmError>`) and + `SteeringInput<'a> { queue: serdes_ai_agent::SteeringQueue, sink: &'a mut + dyn SteeringDeliverySink }`. The trait object keeps the storage layer out of + the LLM crate; the repo already uses `async-trait = "0.1"` + (`Cargo.toml:35`). +- `AgentClientExt::run_agent_stream` and `do_run_agent_stream` + (`client_agent.rs:361-392, 537-588`) gain one parameter + `steering: Option>`. When `Some`, chain + `.steering(channel.queue.clone())` onto the `RunOptions` built at + `client_agent.rs:558-562`. +- `handle_agent_stream_event` (`client_agent.rs:475-534`) becomes `async`, + takes `Option<&mut SteeringInput<'_>>`, and adds an EXPLICIT arm for + `AgentStreamEvent::SteeringDelivered { text, .. }` placed before the + catch-all at line 532. With a channel it awaits `sink.delivered(text)`; + on sink `Err` it emits `StreamEvent::Error` and the caller returns `Err` + (the existing mid-stream failure shape at `client_agent.rs:579-583`). With + `None` (impossible by construction: no queue, no drain, no event) it emits + `StreamEvent::Error` naming the bug rather than trusting the catch-all. +- Update the production caller (`stream_agent_response`, + `streaming.rs:78-88`) to pass `None` for now; real wiring lands in P04. + The legacy `src/llm/stream.rs` path is deliberately untouched (see call + sites section). + +Tests (`src/llm/client_agent.rs` `#[cfg(test)]` module or a sibling test +file, against a synthetic stream of `AgentStreamEvent`s): +- REGRESSION GUARD for the silent swallow: GIVEN a recording sink and a + synthetic `SteeringDelivered { step: 1, text }` WHEN + `handle_agent_stream_event` processes it THEN the sink recorded exactly + that text AND no `StreamEvent::Error` was emitted. This test FAILS if the + explicit arm is missing, because the catch-all at line 532 never calls the + sink. +- GIVEN a sink that returns `Err` WHEN `SteeringDelivered` is processed THEN + `StreamEvent::Error` is emitted and the function reports `Err` +- GIVEN `steering: None` and a `SteeringDelivered` event WHEN processed THEN + `StreamEvent::Error` is emitted (loud, not debug-logged) +- GIVEN no steering events WHEN processed THEN output events are identical to + today (regression). + +### P03: Acceptance commits through the fork queue (test-first) + +Plan ID `PLAN-20260905-STEERINT.P03` + +Prerequisites: P02. + +Files: `src/services/chat_impl.rs`, `src/services/chat_impl/steering.rs` + +- `ActiveStream` (`chat_impl.rs:66-76`) gains `steering: + serdes_ai_agent::SteeringQueue`, created in `begin_stream` + (`chat_impl.rs:245-266`) so its lifetime is the send's slot. +- `queue_steering` (`steering.rs:153-197`) extends its tail: after + `confirm_or_withdraw_steering` confirms, lock `active_streams`, clone the + conversation's queue handle, DROP the lock, call `steer(trimmed_text)`. On + `false`: remove the entry, emit `SteeringDiscarded`, return the same + `no_active_turn()` validation error. Lock discipline preserved: the two + locks are still never held together and nothing is emitted while either is + held. `steer()` takes the trimmed text, matching what the PA queue stores, + so transport and state machine hold identical bytes. +- `spawn_stream_task` / `run_stream_task` signatures gain the queue clone so + turns can attach it (used fully in P04). + +Tests (`src/services/chat_impl/tests/steering/`, same harness as +`withdrawal.rs`): +- GIVEN a running stream WHEN `steer` succeeds THEN the fork queue reports + `pending_len() == 1` AND the PA queue holds the entry +- GIVEN a stream torn down between the confirm re-check and the commit WHEN + `steer()` returns false THEN the entry is removed, `SteeringDiscarded` is + emitted, and `steer` returns `Err` +- GIVEN a withdrawn steer (confirm re-check fails) WHEN the transport is + inspected THEN the fork queue is empty (withdrawal never reaches it) +- GIVEN `MAX_QUEUED_STEERING_MESSAGES` entries WHEN a sixth steer arrives + THEN it is refused and the fork queue received nothing new +- GIVEN steering acceptance at any point THEN the cancel token is untouched + and no approval is resolved (existing REQ-222-006/008 tests keep passing). + +### P04: Mid-turn delivery: persistence, correlation, chaining rewrite (test-first) + +Plan ID `PLAN-20260905-STEERINT.P04` + +Prerequisites: P02, P03. + +Files: `src/services/chat_impl/streaming.rs`; +`src/services/chat_impl/streaming/steering_delivery.rs` (or a new +`steering_delivery/sink.rs` alongside `discard.rs`) + +- `MidTurnSteering` sink (services layer, implements `SteeringDeliverySink`): + owns `Arc`, `conversation_id`, the + `SteeringQueues` handle, and a delivered-text buffer. `delivered(text)`: + pop the PA-queue head under the `steering_queues` lock and release it; + `add_message(conversation_id, Message::user(text))`; on success record the + text in the buffer and emit `ChatEvent::SteeringDelivered { + conversation_id, steer_id }`; on failure emit `ChatEvent::SteeringDiscarded` + for that id and return `Err`. Empty deque (cancel raced the delivery): + persist the row and record the text, emit nothing (the discard already + fired). No text matching anywhere. +- `stream_agent_response` (`streaming.rs:67-100`) takes + `Option<(&serdes_ai_agent::SteeringQueue, &mut MidTurnSteering)>`, builds + the `SteeringInput`, and after `run_agent_stream` returns copies the sink + buffer into a new `StreamTranscript::delivered_steering: Vec` + (`streaming.rs:31-43`). +- `run_stream_task` (`streaming.rs:317-407`) constructs the sink from its + existing `conversation_service` + `steering_queues` and passes queue + sink + into the turn runner closure. +- `run_steered_turns` (`steering_delivery.rs:121-175`) rewrite: + - DELETE `deliver_steering` (205-238) and the `ChainOutcome` machinery + (98-113, 171-173); `run_steered_turns_and_finalize` always ends in + `finalize_by_outcome`. + - The leftover check becomes a peek: new `has_queued_steering(queues, id)` + in `steering.rs` (lock, `!queue.is_empty()`, drop). Nothing is removed on + the happy path; the transport owns the texts. + - Cap branch (`steering_delivery.rs:148-157`): drain the PA queue, announce + every entry `SteeringDiscarded`, return. + - Chaining seed: after the intermediate `persist_assistant_response`, + splice `transcript.delivered_steering` texts as `LlmMessage::user` BEFORE + the existing `assistant_message(&transcript)` push, reproducing row order + `[prompt, steers, assistant]` in the next turn's seed (REQ-SI-004). +- `MAX_STEERING_TURNS` stays at 10 with unchanged counting. + +Tests (`src/services/chat_impl/tests/steering_delivery/`, harness as +`discard.rs`; mock `ConversationService` recording `add_message` order): +- GIVEN a steer submitted while a tool runs WHEN the tool completes THEN the + model's next request in that turn contains the steer text AND a persisted + user row exists before the turn's assistant row (REQ-SI-001, REQ-SI-003) +- GIVEN two steers with identical text WHEN both are delivered THEN each gets + its own `SteeringDelivered` with distinct ids in queue order (REQ-SI-005) +- GIVEN a delivery WHEN `add_message` fails THEN the transcript records an + error, the entry is announced discarded, entries behind it are discarded at + teardown, and no follow-up turn runs (REQ-SI-003) +- GIVEN a delivered steer WHEN the next turn is chained THEN its seed is + `[prompt, steer, assistant]`, matching a fresh `build_llm_messages` reload + of the same rows (REQ-SI-004) +- GIVEN a text-only turn that left a steer queued WHEN the chain decides + THEN another turn runs on the same queue and the leftover delivers at its + first tool boundary (REQ-SI-006, REQ-SI-002) +- GIVEN `MAX_STEERING_TURNS` exhausted with entries queued WHEN the chain + stops THEN every remaining entry is announced discarded (REQ-SI-006) +- GIVEN a cancel that lands after a steer was persisted THEN the row remains + and no discard is announced for it (REQ-SI-008) +- GIVEN a cancel that lands before delivery THEN the entry is announced + discarded and no row is written (REQ-SI-008) +- GIVEN an empty queue WHEN a turn runs THEN behavior is identical to today + (regression) +- GIVEN a pending tool approval WHEN a steer arrives THEN it is delivered + only after the approval resolves and the boundary passes (REQ-222-008 + carried over; the fork drains after `ToolExecuted`). + +### P05: Verification sweep + +Plan ID `PLAN-20260905-STEERINT.P05` + +- `cargo fmt --all -- --check` +- `cargo clippy --all-targets -- -D warnings` +- `cargo test --lib --tests` +- structural/lizard checks used by the repo +- `@plan` / `@requirement` markers on every new item +- Manual smoke: run the app, steer during a tool-using turn, confirm the + bubble resolves mid-turn while the assistant is still streaming, confirm + the persisted order after reload, press Stop with a queued steer and + confirm the discard. + +## Test strategy summary + +Every phase above is test-first; the behavioral GIVEN/WHEN/THEN list lives in +the phases. The critical items: + +- The P02 regression guard that fails if the explicit `SteeringDelivered` + match arm is missing (the catch-all at `client_agent.rs:532` would + otherwise swallow the event with a debug log and zero compiler + diagnostics). +- The P04 order test that pins `[prompt, steer, assistant]` in rows, reload, + and seed simultaneously. +- The P04 FIFO test with duplicate texts, which fails under any + text-matching correlation. +- The P03 empty-transport test that fails if withdrawal order regresses. + +## Risks and rollback + +- Blast radius of the rev bump: all eight pinned crates move, but the + upstream delta is one commit confined to `serdes-ai-agent`; the other + crates are rebuilt from identical sources. Compile-time risk is contained + to `serdes-ai-agent`'s new exports (additive). +- Behavior risk concentrates in P04 (delivery, correlation, chaining). The + phase is one commit; its tests pin the ordering and correlation invariants. +- The two-queue relationship has one commit point (fork `steer()` at the tail + of acceptance) and one read rule (pop head on delivery). Both are pinned by + tests; there are no other couplings. +- Rollback: revert the branch's commits (or the whole branch) and pin + `Cargo.toml:46-55` back to `e675674`. Schema is untouched (`add_message` + user rows written by the new code are ordinary user messages the old code + reads happily), so a rollback never orphans data. Steering rows persisted + mid-turn remain valid transcript entries after rollback. + +## Out of scope + +- The legacy `send_message_stream` path (`src/llm/stream.rs`): no steering, + no changes; its removal is a separate cleanup decision. +- Delivering between two tool calls of one step, or before the first model + request: the fork does not drain there, by design. +- Any change to `Stop` partial-output behavior (issue #218) or the Responses + transport (#217). +- Compression interactions beyond what chaining already does; steering texts + are ordinary user rows to the compression pipeline. + +## Open questions for a human + +1. UI timing: a delivered steer's user row now appears in the transcript while + the assistant is still streaming (today it appeared at a turn boundary). + The view already removes the waiting bubble on `SteeringDelivered` + (`command.rs:191-204`) and renders persisted rows from the transcript, but + the visual behavior of a user message appearing above a streaming answer + mid-turn deserves a look before P04 ships. No code change expected. +2. Whether the unused legacy `send_message_stream` path should be deleted + outright now that it diverges further from the agent path. Not needed for + this feature. diff --git a/src/llm/client_agent.rs b/src/llm/client_agent.rs index 82be6ab..0465423 100644 --- a/src/llm/client_agent.rs +++ b/src/llm/client_agent.rs @@ -1,6 +1,7 @@ //! Agent-based LLM client with MCP tool integration for `PersonalAgent`. use crate::agent::tool_approval_policy::ToolApprovalPolicy; use crate::llm::error::debug_error_message; +use crate::llm::steering::SteeringInput; use crate::llm::{LlmError, Message, Role, StreamEvent}; use crate::presentation::view_command::ViewCommand; use futures::StreamExt; @@ -363,6 +364,7 @@ pub trait AgentClientExt { agent: &Agent, messages: &[Message], context: McpToolContext, + steering: Option>, on_event: F, ) -> impl std::future::Future> + Send where @@ -382,12 +384,13 @@ impl AgentClientExt for crate::llm::LlmClient { agent: &Agent, messages: &[Message], context: McpToolContext, + steering: Option>, mut on_event: F, ) -> StdResult<(), LlmError> where F: FnMut(StreamEvent) + Send, { - self.do_run_agent_stream(agent, messages, context, &mut on_event) + self.do_run_agent_stream(agent, messages, context, steering, &mut on_event) .await } @@ -472,11 +475,13 @@ impl crate::llm::LlmClient { (tool_calls, tool_results) } - fn handle_agent_stream_event( + async fn handle_agent_stream_event( event: AgentStreamEvent, history_request_count: usize, + steering: Option<&mut SteeringInput<'_>>, on_event: &mut F, - ) where + ) -> StdResult<(), LlmError> + where F: FnMut(StreamEvent) + Send, { match event { @@ -529,8 +534,47 @@ impl crate::llm::LlmClient { tracing::error!("run_agent_stream: Error: {}", message); on_event(StreamEvent::Error(message)); } + AgentStreamEvent::SteeringDelivered { text, .. } => { + Self::handle_steering_delivered(&text, steering, on_event).await?; + } other => tracing::debug!("run_agent_stream: other event: {:?}", other), } + Ok(()) + } + + /// Hand a delivered steering text to the caller's sink. + /// + /// A sink refusal emits a stream error and fails the run: the model has + /// already seen the text, so the divergence is reported, never retried or + /// swallowed. A delivery without steering input is impossible by + /// construction (the fork drains only a queue attached through + /// [`SteeringInput`]) and is reported as the wiring bug it is, loudly, + /// instead of falling into the catch-all's debug log. + /// + /// @plan PLAN-20260905-STEERINT.P02 + /// @requirement REQ-SI-001 + /// @requirement REQ-SI-003 + async fn handle_steering_delivered( + text: &str, + steering: Option<&mut SteeringInput<'_>>, + on_event: &mut F, + ) -> StdResult<(), LlmError> + where + F: FnMut(StreamEvent) + Send, + { + let Some(input) = steering else { + let message = + "internal bug: SteeringDelivered event received without steering input attached" + .to_string(); + on_event(StreamEvent::Error(message.clone())); + return Err(LlmError::Stream(message)); + }; + if let Err(error) = input.sink.delivered(text).await { + let message = format!("steering delivery failed: {error}"); + on_event(StreamEvent::Error(message.clone())); + return Err(LlmError::Stream(message)); + } + Ok(()) } #[allow(clippy::cognitive_complexity)] @@ -539,6 +583,7 @@ impl crate::llm::LlmClient { agent: &Agent, messages: &[Message], context: McpToolContext, + mut steering: Option>, on_event: &mut F, ) -> StdResult<(), LlmError> where @@ -555,11 +600,14 @@ impl crate::llm::LlmClient { ); let history_request_count = message_history.len(); - let options = if message_history.is_empty() { + let mut options = if message_history.is_empty() { RunOptions::default() } else { RunOptions::default().message_history(message_history) }; + if let Some(channel) = steering.as_ref() { + options = options.steering(channel.queue.clone()); + } tracing::info!("run_agent_stream: creating AgentStream..."); let mut stream = AgentStream::new(agent, UserContent::text(prompt), context, options) @@ -573,7 +621,13 @@ impl crate::llm::LlmClient { while let Some(event_result) = stream.next().await { match event_result { Ok(event) => { - Self::handle_agent_stream_event(event, history_request_count, on_event); + Self::handle_agent_stream_event( + event, + history_request_count, + steering.as_mut(), + on_event, + ) + .await?; } Err(e) => { diff --git a/src/llm/client_agent/tests.rs b/src/llm/client_agent/tests.rs index 7852d75..09fa2ef 100644 --- a/src/llm/client_agent/tests.rs +++ b/src/llm/client_agent/tests.rs @@ -361,8 +361,8 @@ fn collect_tool_transcript_uses_built_history_request_count_not_message_count() assert_eq!(tool_results[0].tool_use_id, "current-tool-call"); } -#[test] -fn tool_executed_emits_completion_without_interim_tool_transcript() { +#[tokio::test] +async fn tool_executed_emits_completion_without_interim_tool_transcript() { let mut events = Vec::new(); crate::llm::LlmClient::handle_agent_stream_event( AgentStreamEvent::ToolExecuted { @@ -372,8 +372,11 @@ fn tool_executed_emits_completion_without_interim_tool_transcript() { error: None, }, 0, + None, &mut |event| events.push(event), - ); + ) + .await + .expect("handling a non-steering event should not fail"); crate::llm::LlmClient::handle_agent_stream_event( AgentStreamEvent::ToolExecuted { tool_name: "shell_exec".to_string(), @@ -382,8 +385,11 @@ fn tool_executed_emits_completion_without_interim_tool_transcript() { error: Some("command failed".to_string()), }, 0, + None, &mut |event| events.push(event), - ); + ) + .await + .expect("handling a non-steering event should not fail"); assert_eq!( events.len(), @@ -430,8 +436,8 @@ fn tool_executed_emits_completion_without_interim_tool_transcript() { } } -#[test] -fn run_complete_emits_authoritative_tool_transcript_then_complete() { +#[tokio::test] +async fn run_complete_emits_authoritative_tool_transcript_then_complete() { let mut response = ModelResponse::new(); response.add_part(ModelResponsePart::ToolCall( ToolCallPart::new( @@ -456,8 +462,11 @@ fn run_complete_emits_authoritative_tool_transcript_then_complete() { usage: RunUsage::default(), }, 0, + None, &mut |event| events.push(event), - ); + ) + .await + .expect("handling a non-steering event should not fail"); assert_eq!( events.len(), @@ -608,3 +617,201 @@ async fn resolving_one_conversation_does_not_wake_another() { let result = tokio::time::timeout(Duration::from_millis(50), waiter_b.wait()).await; assert!(result.is_err(), "waiter B should still be pending"); } + +// --------------------------------------------------------------------------- +// Steering delivery contract of `handle_agent_stream_event` +// PLAN-20260905-STEERINT.P02 +// --------------------------------------------------------------------------- + +use crate::llm::steering::SteeringDeliverySink; +use async_trait::async_trait; +use std::sync::{Arc, Mutex}; + +/// Records every event the handler emits so tests can assert the exact +/// output sequence. +#[derive(Clone, Default)] +struct EventLog(Arc>>); + +impl EventLog { + fn record(&self, event: StreamEvent) { + self.0.lock().expect("event log lock").push(event); + } + + fn events(&self) -> Vec { + self.0.lock().expect("event log lock").clone() + } + + fn error_count(&self) -> usize { + self.events() + .iter() + .filter(|event| matches!(event, StreamEvent::Error(_))) + .count() + } +} + +/// Sink that records the texts it was handed; the buffer is shared so the +/// test can read it after the handler call returns. +#[derive(Default)] +struct RecordingSink { + texts: Arc>>, +} + +impl RecordingSink { + fn texts(&self) -> Vec { + self.texts.lock().expect("sink lock").clone() + } +} + +#[async_trait] +impl SteeringDeliverySink for RecordingSink { + async fn delivered(&mut self, text: &str) -> StdResult<(), LlmError> { + self.texts.lock().expect("sink lock").push(text.to_string()); + Ok(()) + } +} + +/// Sink that refuses every delivery. +struct FailingSink; + +#[async_trait] +impl SteeringDeliverySink for FailingSink { + async fn delivered(&mut self, _text: &str) -> StdResult<(), LlmError> { + Err(LlmError::Stream("sink refused the delivery".to_string())) + } +} + +/// REGRESSION GUARD: a `SteeringDelivered` event must reach the sink. If the +/// explicit match arm goes missing, the catch-all debug-logs the event and +/// swallows it silently, the sink sees nothing, and no error is emitted; this +/// test fails in exactly that state. +/// +/// @plan PLAN-20260905-STEERINT.P02 +/// @requirement REQ-SI-001 +/// @requirement REQ-SI-003 +#[tokio::test] +async fn steering_delivered_event_reaches_the_sink() { + let mut sink = RecordingSink::default(); + let log = EventLog::default(); + let mut input = SteeringInput { + queue: SteeringQueue::new(), + sink: &mut sink, + }; + + crate::llm::LlmClient::handle_agent_stream_event( + AgentStreamEvent::SteeringDelivered { + step: 1, + text: "focus on the error case".to_string(), + }, + 0, + Some(&mut input), + &mut |event: StreamEvent| log.record(event), + ) + .await + .expect("delivery should not fail the stream"); + + assert_eq!(sink.texts(), ["focus on the error case"]); + assert_eq!(log.error_count(), 0); +} + +/// A sink refusal is loud: the handler emits a stream error and reports the +/// failure to its caller instead of continuing as if the text had landed. +/// +/// @plan PLAN-20260905-STEERINT.P02 +/// @requirement REQ-SI-003 +#[tokio::test] +async fn failing_sink_emits_error_and_reports_err() { + let log = EventLog::default(); + let mut sink = FailingSink; + let mut input = SteeringInput { + queue: SteeringQueue::new(), + sink: &mut sink, + }; + + let result = crate::llm::LlmClient::handle_agent_stream_event( + AgentStreamEvent::SteeringDelivered { + step: 1, + text: "any".to_string(), + }, + 0, + Some(&mut input), + &mut |event: StreamEvent| log.record(event), + ) + .await; + + assert!(result.is_err()); + assert_eq!(log.error_count(), 1); +} + +/// A delivery with no steering input attached is an internal bug (no queue +/// attached means the fork can never drain, so the event is impossible by +/// construction). It must be reported as a stream error, not swallowed by the +/// catch-all's debug log. +/// +/// @plan PLAN-20260905-STEERINT.P02 +/// @requirement REQ-SI-001 +#[tokio::test] +async fn steering_delivered_without_input_is_a_loud_error() { + let log = EventLog::default(); + + let result = crate::llm::LlmClient::handle_agent_stream_event( + AgentStreamEvent::SteeringDelivered { + step: 1, + text: "ghost".to_string(), + }, + 0, + None, + &mut |event: StreamEvent| log.record(event), + ) + .await; + + assert_eq!(log.error_count(), 1); + assert!(result.is_err()); +} + +/// Without steering traffic the handler's output events are exactly what they +/// were before steering existed. +/// +/// @plan PLAN-20260905-STEERINT.P02 +#[tokio::test] +async fn non_steering_events_map_unchanged() { + let log = EventLog::default(); + let events = vec![ + AgentStreamEvent::TextDelta { + text: "hello".to_string(), + }, + AgentStreamEvent::RunComplete { + run_id: "run".to_string(), + messages: Vec::new(), + usage: serdes_ai_agent::RunUsage::default(), + }, + ]; + + for event in events { + crate::llm::LlmClient::handle_agent_stream_event( + event, + 0, + None, + &mut |event: StreamEvent| log.record(event), + ) + .await + .expect("non-steering events must not fail"); + } + + let emitted = log.events(); + assert_eq!(emitted.len(), 3); + assert!(matches!(&emitted[0], StreamEvent::TextDelta(t) if t == "hello")); + assert!(matches!( + &emitted[1], + StreamEvent::ToolTranscript { + tool_calls, + tool_results + } if tool_calls.is_empty() && tool_results.is_empty() + )); + assert!(matches!( + &emitted[2], + StreamEvent::Complete { + input_tokens: None, + output_tokens: None + } + )); +} diff --git a/src/llm/mod.rs b/src/llm/mod.rs index 3ce2a71..fdfc5b3 100644 --- a/src/llm/mod.rs +++ b/src/llm/mod.rs @@ -12,6 +12,7 @@ mod normalizing_model; pub mod open_responses; mod provider_quirks; pub(crate) mod sse_normalize; +pub mod steering; mod stream; pub mod tools; diff --git a/src/llm/steering.rs b/src/llm/steering.rs new file mode 100644 index 0000000..93a6707 --- /dev/null +++ b/src/llm/steering.rs @@ -0,0 +1,56 @@ +//! Mid-turn steering plumbing between the chat service layer and the agent +//! stream. +//! +//! The transport half lives in the fork (`serdes_ai_agent::SteeringQueue`): +//! it decides when queued text reaches the model, at the tool-call boundary +//! of a running turn. This module carries the two pieces the LLM layer needs +//! to participate without knowing about persistence or the UI: the input +//! bundle attached to a run, and the sink that observes what the model +//! actually received. +//! +//! @plan PLAN-20260905-STEERINT.P02 +//! @requirement REQ-SI-001 +//! @requirement REQ-SI-003 + +use async_trait::async_trait; + +use crate::llm::LlmError; + +// Use std Result to avoid conflict with serdes_ai::prelude::Result +type StdResult = std::result::Result; + +/// Observes steering texts the model actually received mid-turn. +/// +/// The LLM layer calls [`SteeringDeliverySink::delivered`] once per +/// [`AgentStreamEvent::SteeringDelivered`], in delivery order. Implementors +/// live in the service layer so storage stays out of the LLM crate; the trait +/// object keeps the dependency pointed this direction only. +/// +/// @plan PLAN-20260905-STEERINT.P02 +/// @requirement REQ-SI-003 +#[async_trait] +pub trait SteeringDeliverySink: Send { + /// Record a steering text that reached the model. + /// + /// # Errors + /// + /// Returns `Err` when the delivered text could not be handled (today: + /// not persisted). The stream fails on the error; the model has already + /// seen the text, so the divergence is reported, never retried. + async fn delivered(&mut self, text: &str) -> StdResult<(), LlmError>; +} + +/// Everything a run needs to accept mid-turn steering. +/// +/// The queue is cloned onto the run's `RunOptions`; the sink observes the +/// resulting deliveries. Owned by the caller of +/// [`AgentClientExt::run_agent_stream`] for the duration of the send. +/// +/// @plan PLAN-20260905-STEERINT.P02 +/// @requirement REQ-SI-002 +pub struct SteeringInput<'a> { + /// Transport queue the run drains at tool-call boundaries. + pub queue: serdes_ai_agent::SteeringQueue, + /// Observer invoked for every text the run delivers to the model. + pub sink: &'a mut dyn SteeringDeliverySink, +} diff --git a/src/services/chat_impl.rs b/src/services/chat_impl.rs index 37e7f68..bd22954 100644 --- a/src/services/chat_impl.rs +++ b/src/services/chat_impl.rs @@ -32,7 +32,10 @@ mod streaming; mod titling; use prompt::{build_system_prompt, filter_emoji_setting}; -use steering::{drain_steering_queue, emit_steering_discarded, QueuedSteering, SteeringQueues}; +use steering::{ + drain_steering_queue, emit_steering_discarded, has_queued_steering, pop_steering_head, + SteeringQueues, +}; use streaming::{ build_stream_error_diagnostics, clear_streaming_state, emit_stream_error, run_stream_task, StreamDiagnosticContext, StreamTranscript, STREAM_ERROR_MESSAGE, @@ -40,7 +43,7 @@ use streaming::{ use titling::{generate_and_apply_title, TitleGenerationRequest}; #[cfg(test)] -use steering::MAX_QUEUED_STEERING_MESSAGES; +use steering::{QueuedSteering, MAX_QUEUED_STEERING_MESSAGES}; // Re-export for tests #[cfg(test)] @@ -73,6 +76,13 @@ pub(super) struct ActiveStream { pub(super) cancel: CancellationToken, /// Current lifecycle state of the stream. pub(super) state: StreamLifecycle, + /// The send's steering transport: one fork queue per send, shared by + /// every chained turn, dropped with this entry. The acceptance path + /// commits to it; the turns attach it so delivered texts reach the sink. + /// + /// @plan PLAN-20260905-STEERINT.P03 + /// @requirement REQ-SI-002 + pub(super) steering: serdes_ai_agent::SteeringQueue, } pub struct ChatServiceImpl { @@ -259,6 +269,12 @@ impl ChatServiceImpl { task: None, cancel: cancel.clone(), state: StreamLifecycle::Starting, + // Created with the slot so the send's lifetime bounds the + // transport: texts committed to it die with the send. + // + // @plan PLAN-20260905-STEERINT.P03 + // @requirement REQ-SI-002 + steering: serdes_ai_agent::SteeringQueue::new(), }, ); } @@ -614,6 +630,22 @@ impl ChatServiceImpl { let approval_gate = self.approval_gate.clone(); let policy = self.policy.clone(); + // The send's transport is read from the reservation so the turn + // runner can attach the same queue the acceptance path commits to. + // The lock is released before the spawn; a superseded or removed + // reservation means there is nothing to run for, exactly the state + // the post-spawn guard below aborts. + // + // @plan PLAN-20260905-STEERINT.P03 + // @requirement REQ-SI-002 + let steering_queue = { + let map = self.active_streams.lock().expect("active_streams poisoned"); + match map.get(&conversation_id) { + Some(entry) if entry.stream_id == stream_id => entry.steering.clone(), + _ => return, + } + }; + let handle = tokio::spawn(async move { run_stream_task( prepared, @@ -628,6 +660,7 @@ impl ChatServiceImpl { view_tx, approval_gate, policy, + steering_queue, ) .await; }); @@ -886,100 +919,7 @@ impl ChatService for ChatServiceImpl { } #[cfg(test)] -impl ChatServiceImpl { - /// Test-only shim to begin a stream for a specific conversation. - /// - /// Exercises the real `begin_stream` guard path (same synchronous - /// reservation used in production `send_message`), then promotes the - /// reservation from `Starting` to `Running` by attaching a long-sleeping - /// mock task. This matches what `spawn_stream_task` does for real - /// streams, so tests see `is_streaming_for` return `true` without - /// driving an actual LLM stream. - /// - /// Returns `Err` with the real error message if `begin_stream` - /// rejects the reservation (e.g. duplicate conversation). - /// - /// @plan PLAN-20260416-ISSUE173.P14-CR5 - /// @requirement REQ-173-001.1 - /// @requirement REQ-173-001.2 - pub(crate) fn begin_stream_for_test(&self, conversation_id: Uuid) -> ServiceResult<()> { - let (stream_id, cancel) = self.begin_stream(conversation_id)?; - - // Mock long-running task. When the stream is cancelled or cleared, - // the `CancellationToken` is cancelled (and `task.abort()` is called - // by `cancel_active_stream` / `clear_all_streams_for_test`), which - // wakes this sleep and lets the task exit cleanly. - let mut task = Some(tokio::spawn(async move { - tokio::select! { - () = cancel.cancelled() => {} - () = tokio::time::sleep(tokio::time::Duration::from_hours(1)) => {} - } - })); - - let promoted = { - let mut map = self.active_streams.lock().expect("active_streams poisoned"); - if let Some(entry) = map.get_mut(&conversation_id) { - if entry.stream_id == stream_id { - entry.task = task.take(); - entry.state = StreamLifecycle::Running; - true - } else { - false - } - } else { - false - } - }; - if promoted { - Ok(()) - } else { - // Entry was replaced or removed between reserve and promotion — - // drop the spawned task to avoid leaking it. - if let Some(handle) = task { - handle.abort(); - } - Err(ServiceError::Internal( - "Stream reservation lost before promotion".to_string(), - )) - } - } - - /// Test-only helper to read back the `stream_id` of the active entry for - /// a conversation, if any. Used by CR #4 regression tests that verify - /// a stale task cannot evict a newer stream. - /// - /// @plan PLAN-20260416-ISSUE173.P14-CR4 - /// @requirement REQ-173-001.3 - pub(crate) fn stream_id_for_test(&self, conversation_id: Uuid) -> Option { - let map = self.active_streams.lock().expect("active_streams poisoned"); - map.get(&conversation_id).map(|a| a.stream_id) - } - - /// Test-only handles to the two registries a spawned stream task is - /// given, so tests can drive the steering delivery loop against the same - /// state the real service reads through `is_streaming_for` and `steer`. - /// - /// @plan PLAN-20260903-ISSUE222.P02 - /// @requirement REQ-222-005 - pub(in crate::services::chat_impl) fn stream_registries_for_test( - &self, - ) -> (Arc>>, SteeringQueues) { - (self.active_streams.clone(), self.steering_queues.clone()) - } - - /// Test-only helper to clear all mock streams. - /// @plan PLAN-20260416-ISSUE173.P03 - /// @requirement REQ-173-001.3 - pub(crate) fn clear_all_streams_for_test(&self) { - let mut map = self.active_streams.lock().expect("active_streams poisoned"); - for (_, active) in map.drain() { - active.cancel.cancel(); - if let Some(handle) = active.task { - handle.abort(); - } - } - } -} +mod stream_test_hooks; #[cfg(test)] mod tests; diff --git a/src/services/chat_impl/steering.rs b/src/services/chat_impl/steering.rs index b5bb83c..7603647 100644 --- a/src/services/chat_impl/steering.rs +++ b/src/services/chat_impl/steering.rs @@ -8,7 +8,7 @@ //! @requirement REQ-222-004 //! @requirement REQ-222-006 -use super::{ChatService, ChatServiceImpl, ServiceError, ServiceResult}; +use super::{ChatServiceImpl, ServiceError, ServiceResult, StreamLifecycle}; use crate::events::types::ChatEvent; use crate::events::{emit, AppEvent}; use std::collections::{HashMap, VecDeque}; @@ -31,6 +31,8 @@ pub(super) const MAX_QUEUED_STEERING_MESSAGES: usize = 5; /// @requirement REQ-222-004 pub(super) struct QueuedSteering { pub(super) id: Uuid, + /// The send the user was steering, never a later send in the same conversation. + pub(super) stream_id: Uuid, pub(super) text: String, } @@ -89,6 +91,49 @@ pub(super) fn drain_steering_queue( .unwrap_or_default() } +/// Whether any steering message is queued for a conversation, without +/// taking any of them. +/// +/// The chaining loop only needs to know whether a follow-up turn could still +/// deliver something. The texts stay owned by the transport, which delivers +/// them and resolves their entries through the sink; draining them here +/// would strand a text the transport still holds. An entry popped by the +/// sink can leave an empty deque behind, so an empty queue counts as no. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-002 +/// @requirement REQ-SI-006 +pub(super) fn has_queued_steering(queues: &SteeringQueues, conversation_id: Uuid) -> bool { + queues + .lock() + .expect("steering_queues poisoned") + .get(&conversation_id) + .is_some_and(|entries| !entries.is_empty()) +} + +/// Take the oldest steering message queued for a conversation, if any. +/// +/// The mid-turn sink resolves deliveries by FIFO position: the fork drains +/// its transport in the order acceptance pushed, so the head is always the +/// entry whose text was just delivered, whatever that text says. This is the +/// single-entry counterpart of [`drain_steering_queue`], and it holds the +/// same lock discipline: take the entry, release the lock, then let the +/// caller emit. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-003 +/// @requirement REQ-SI-005 +pub(super) fn pop_steering_head( + queues: &SteeringQueues, + conversation_id: Uuid, +) -> Option { + queues + .lock() + .expect("steering_queues poisoned") + .get_mut(&conversation_id) + .and_then(VecDeque::pop_front) +} + /// Announce that queued steering messages are never going to be delivered. /// /// A queued entry is rendered as waiting until something reports what became @@ -131,8 +176,8 @@ impl ChatServiceImpl { /// never resolves a pending approval. `cancel` stays reachable only from /// an explicit user stop (REQ-222-006). /// - /// Lock discipline: the `active_streams` guard taken by - /// `is_streaming_for` is released before `steering_queues` is locked, so + /// Lock discipline: the `active_streams` guard capturing the originating + /// stream id is released before `steering_queues` is locked, so /// the two locks are never held at once. See [`SteeringQueues`]. That /// also means the check and the insert are not one atomic step, which is /// why the entry is confirmed against the stream state again once it is @@ -158,12 +203,18 @@ impl ChatServiceImpl { )); } - if !ChatService::is_streaming_for(self, conversation_id) { - return Err(no_active_turn()); - } + let stream_id = self + .active_streams + .lock() + .expect("active_streams poisoned") + .get(&conversation_id) + .filter(|entry| matches!(entry.state, StreamLifecycle::Running)) + .map(|entry| entry.stream_id) + .ok_or_else(no_active_turn)?; let entry = QueuedSteering { id: Uuid::new_v4(), + stream_id, text: trimmed.to_string(), }; let steer_id = entry.id; @@ -193,7 +244,58 @@ impl ChatServiceImpl { text: queued_text, })); - self.confirm_or_withdraw_steering(conversation_id, steer_id) + self.confirm_or_withdraw_steering(conversation_id, steer_id)?; + + // The transport commit is the acceptance's point of no return, so it + // runs only after the re-check confirmed the entry: a committed text + // reaches a model request whether or not anything else ever refuses + // it, which is why nothing that can refuse may run after this. + // + // @plan PLAN-20260905-STEERINT.P03 + // @requirement REQ-SI-002 + self.commit_steering_to_transport(conversation_id, steer_id, trimmed)?; + Ok(steer_id) + } + + /// Commit an accepted steer's text to its send's transport queue. + /// + /// The fork queue lives on the send's `ActiveStream` slot; cloning the + /// handle and dropping the lock before the commit keeps the registry + /// discipline intact (the two locks are still never held together, and + /// nothing is emitted under either). + /// + /// A `false` from `steer` means the transport is closed: the send went + /// away between the confirm re-check and this commit, so the text can + /// never reach a model request. The entry is taken back off the queue + /// and announced as discarded, with the same `no_active_turn` refusal + /// the re-check returns, so a caller cannot tell the two windows apart. + /// + /// Lock discipline: the `active_streams` guard is released before + /// `remove_steering` locks `steering_queues`, and both are released + /// before anything is emitted. See [`SteeringQueues`]. + /// + /// # Errors + /// + /// Returns the same `ServiceError::Validation` the confirm re-check + /// returns when the send is gone and the commit cannot happen. + /// + /// @plan PLAN-20260905-STEERINT.P03 + /// @requirement REQ-SI-002 + pub(super) fn commit_steering_to_transport( + &self, + conversation_id: Uuid, + steer_id: Uuid, + text: &str, + ) -> ServiceResult<()> { + let transport = self.steering_transport_for_entry(conversation_id, steer_id); + + if transport.is_some_and(|queue| queue.steer(text.to_string())) { + return Ok(()); + } + + let _ = self.remove_steering(conversation_id, steer_id); + emit_steering_discarded_id(conversation_id, steer_id); + Err(no_active_turn()) } /// Confirm a steer that is now on the queue, or take it back. @@ -221,7 +323,10 @@ impl ChatServiceImpl { /// the view removes by id, so the second one finds no entry and takes /// nothing, which `a_discard_of_an_unknown_id_withdraws_nothing` pins. /// - /// Lock discipline: `is_streaming_for` releases `active_streams` before + /// Confirmation requires the queued entry's originating stream id, not + /// merely another running turn in the same conversation. + /// + /// Lock discipline: `steering_transport_for_entry` releases both registries before /// `remove_steering` locks `steering_queues`, and both are released /// before anything is emitted. See [`SteeringQueues`]. /// @@ -239,7 +344,10 @@ impl ChatServiceImpl { conversation_id: Uuid, steer_id: Uuid, ) -> ServiceResult { - if ChatService::is_streaming_for(self, conversation_id) { + if self + .steering_transport_for_entry(conversation_id, steer_id) + .is_some() + { return Ok(steer_id); } @@ -248,6 +356,36 @@ impl ChatServiceImpl { Err(no_active_turn()) } + /// Resolve only the running transport belonging to this queued entry's send. + /// + /// Release the queue lock before reading the stream registry. If teardown + /// or replacement occurs between reads, the captured stream id still prevents + /// committing this instruction to the replacement's transport. + fn steering_transport_for_entry( + &self, + conversation_id: Uuid, + steer_id: Uuid, + ) -> Option { + let stream_id = { + let queues = self + .steering_queues + .lock() + .expect("steering_queues poisoned"); + queues + .get(&conversation_id)? + .iter() + .find(|entry| entry.id == steer_id)? + .stream_id + }; + let streams = self.active_streams.lock().expect("active_streams poisoned"); + streams + .get(&conversation_id) + .filter(|entry| { + entry.stream_id == stream_id && matches!(entry.state, StreamLifecycle::Running) + }) + .map(|entry| entry.steering.clone()) + } + /// Append `entry` to a conversation's steering queue. /// /// Returns `false` without queuing when the conversation already holds diff --git a/src/services/chat_impl/stream_test_hooks.rs b/src/services/chat_impl/stream_test_hooks.rs new file mode 100644 index 0000000..1ecfdc4 --- /dev/null +++ b/src/services/chat_impl/stream_test_hooks.rs @@ -0,0 +1,126 @@ +//! Test-only hooks on `ChatServiceImpl` for driving real stream state. +//! +//! Moved here from `chat_impl.rs` so that file stays under the repo's +//! 1000-line ceiling. Pure relocation: the methods are unchanged and remain +//! compiled only under `cfg(test)`. +//! +//! @plan PLAN-20260905-STEERINT.P04 + +use std::collections::HashMap; +use std::sync::{Arc, Mutex as StdMutex}; + +use uuid::Uuid; + +use super::{ + ActiveStream, ChatServiceImpl, ServiceError, ServiceResult, SteeringQueues, StreamLifecycle, +}; + +#[cfg(test)] +impl ChatServiceImpl { + /// Test-only shim to begin a stream for a specific conversation. + /// + /// Exercises the real `begin_stream` guard path (same synchronous + /// reservation used in production `send_message`), then promotes the + /// reservation from `Starting` to `Running` by attaching a long-sleeping + /// mock task. This matches what `spawn_stream_task` does for real + /// streams, so tests see `is_streaming_for` return `true` without + /// driving an actual LLM stream. + /// + /// Returns `Err` with the real error message if `begin_stream` + /// rejects the reservation (e.g. duplicate conversation). + /// + /// @plan PLAN-20260416-ISSUE173.P14-CR5 + /// @requirement REQ-173-001.1 + /// @requirement REQ-173-001.2 + pub(crate) fn begin_stream_for_test(&self, conversation_id: Uuid) -> ServiceResult<()> { + let (stream_id, cancel) = self.begin_stream(conversation_id)?; + + // Mock long-running task. When the stream is cancelled or cleared, + // the `CancellationToken` is cancelled (and `task.abort()` is called + // by `cancel_active_stream` / `clear_all_streams_for_test`), which + // wakes this sleep and lets the task exit cleanly. + let mut task = Some(tokio::spawn(async move { + tokio::select! { + () = cancel.cancelled() => {} + () = tokio::time::sleep(tokio::time::Duration::from_hours(1)) => {} + } + })); + + let promoted = { + let mut map = self.active_streams.lock().expect("active_streams poisoned"); + if let Some(entry) = map.get_mut(&conversation_id) { + if entry.stream_id == stream_id { + entry.task = task.take(); + entry.state = StreamLifecycle::Running; + true + } else { + false + } + } else { + false + } + }; + if promoted { + Ok(()) + } else { + // Entry was replaced or removed between reserve and promotion — + // drop the spawned task to avoid leaking it. + if let Some(handle) = task { + handle.abort(); + } + Err(ServiceError::Internal( + "Stream reservation lost before promotion".to_string(), + )) + } + } + + /// Test-only helper to read back the `stream_id` of the active entry for + /// a conversation, if any. Used by CR #4 regression tests that verify + /// a stale task cannot evict a newer stream. + /// + /// @plan PLAN-20260416-ISSUE173.P14-CR4 + /// @requirement REQ-173-001.3 + pub(crate) fn stream_id_for_test(&self, conversation_id: Uuid) -> Option { + let map = self.active_streams.lock().expect("active_streams poisoned"); + map.get(&conversation_id).map(|a| a.stream_id) + } + + /// Test-only handles to the two registries a spawned stream task is + /// given, so tests can drive the steering delivery loop against the same + /// state the real service reads through `is_streaming_for` and `steer`. + /// + /// @plan PLAN-20260903-ISSUE222.P02 + /// @requirement REQ-222-005 + pub(in crate::services::chat_impl) fn stream_registries_for_test( + &self, + ) -> (Arc>>, SteeringQueues) { + (self.active_streams.clone(), self.steering_queues.clone()) + } + + /// Test-only handle to a conversation's steering transport: the fork + /// queue the send's slot holds, so acceptance tests can observe the + /// commit that puts committed text where a run will deliver it. + /// + /// @plan PLAN-20260905-STEERINT.P03 + /// @requirement REQ-SI-002 + pub(in crate::services::chat_impl) fn steering_transport_for_test( + &self, + conversation_id: Uuid, + ) -> Option { + let map = self.active_streams.lock().expect("active_streams poisoned"); + map.get(&conversation_id).map(|a| a.steering.clone()) + } + + /// Test-only helper to clear all mock streams. + /// @plan PLAN-20260416-ISSUE173.P03 + /// @requirement REQ-173-001.3 + pub(crate) fn clear_all_streams_for_test(&self) { + let mut map = self.active_streams.lock().expect("active_streams poisoned"); + for (_, active) in map.drain() { + active.cancel.cancel(); + if let Some(handle) = active.task { + handle.abort(); + } + } + } +} diff --git a/src/services/chat_impl/streaming.rs b/src/services/chat_impl/streaming.rs index bb4f30f..9041567 100644 --- a/src/services/chat_impl/streaming.rs +++ b/src/services/chat_impl/streaming.rs @@ -9,13 +9,14 @@ use crate::events::{emit, AppEvent}; use crate::llm::error::{debug_error_message, LlmError}; use crate::llm::{LlmClient, StreamEvent as LlmStreamEvent}; use crate::models::{ContextState, Message}; -use crate::services::ConversationService; +use crate::services::{ConversationService, SkillsService}; use crate::ui_gpui::error_log::{ base_url_host, sanitize_text, ErrorLogDiagnosticContext, ErrorLogRunStatus, ErrorLogStreamLifecycle, ErrorLogToolContext, }; use std::collections::HashMap; +use std::pin::Pin; use std::sync::Arc; use tokio_util::sync::CancellationToken; use uuid::Uuid; @@ -23,8 +24,12 @@ use uuid::Uuid; use super::{build_stream_context, create_stream_agent}; pub(super) mod steering_delivery; +pub(super) mod steering_sink; use steering_delivery::{run_steered_turns_and_finalize, SteeringDeliveryContext}; +use steering_sink::MidTurnSteering; + +use crate::llm::steering::SteeringInput; pub(super) const STREAM_ERROR_MESSAGE: &str = "An error interrupted the chat stream."; @@ -63,6 +68,13 @@ impl StreamDiagnosticContext { } } +/// Run one turn of a send, with the send's steering transport attached: texts +/// the fork delivers mid-turn reach `steering_sink`, which persists them, +/// resolves the queue head by FIFO position, and announces the delivery. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-001 +/// @requirement REQ-SI-003 #[allow(clippy::too_many_arguments)] pub(super) async fn stream_agent_response( client: &LlmClient, @@ -72,11 +84,22 @@ pub(super) async fn stream_agent_response( diagnostics_context: &StreamDiagnosticContext, conversation_id: Uuid, tx: &tokio::sync::mpsc::UnboundedSender, + steering_queue: &serdes_ai_agent::SteeringQueue, + steering_sink: &Arc>, ) -> StreamTranscript { let mut transcript = StreamTranscript::default(); + // The guard lives for the whole run because the fork calls the sink from + // inside it. The mutex exists for the type system only: one sink per + // send, and a send's turns run sequentially, so nothing ever contends. + let mut steering_guard = steering_sink.lock().await; + let steering_input = SteeringInput { + queue: steering_queue.clone(), + sink: &mut *steering_guard, + }; + if let Err(error) = client - .run_agent_stream(agent, messages, context, |event| { + .run_agent_stream(agent, messages, context, Some(steering_input), |event| { handle_llm_stream_event( diagnostics_context, event, @@ -96,6 +119,12 @@ pub(super) async fn stream_agent_response( ); } + tracing::debug!( + conversation_id = %conversation_id, + delivered_steering = steering_guard.delivered, + "agent stream finished" + ); + transcript } @@ -301,6 +330,59 @@ impl StreamFinalizeContext<'_> { } } +/// The future one steered turn runs on. +/// +/// Boxed so the runner closure has a single callable type across turns while +/// carrying the send's borrowed pieces; one allocation per user-visible turn. +type SteeredTurnFuture<'a> = + Pin + Send + 'a>>; + +/// Build the turn runner `run_steered_turns_and_finalize` drives: each turn +/// runs with the send's transport and mid-turn sink attached. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-001 +#[allow(clippy::too_many_arguments)] +fn steered_turn_runner<'a>( + client: &'a LlmClient, + agent: &'a serdes_ai_agent::Agent, + diagnostics_context: &'a StreamDiagnosticContext, + conversation_id: Uuid, + tx: &'a tokio::sync::mpsc::UnboundedSender, + view_tx: tokio::sync::mpsc::Sender, + approval_gate: Arc, + policy: Arc>, + skills_service: Arc, + filter_emoji: bool, + steering_queue: &'a serdes_ai_agent::SteeringQueue, + mid_turn_steering: &'a Arc>, +) -> impl FnMut(Vec) -> SteeredTurnFuture<'a> + 'a { + move |turn_messages: Vec| { + let context = build_stream_context( + conversation_id, + view_tx.clone(), + approval_gate.clone(), + policy.clone(), + skills_service.clone(), + filter_emoji, + ); + Box::pin(async move { + stream_agent_response( + client, + agent, + &turn_messages, + context, + diagnostics_context, + conversation_id, + tx, + steering_queue, + mid_turn_steering, + ) + .await + }) + } +} + /// Run a stream task for a conversation. /// /// One user send runs one turn, plus any further turns queued steering @@ -327,6 +409,12 @@ pub(super) async fn run_stream_task( view_tx: tokio::sync::mpsc::Sender, approval_gate: Arc, policy: Arc>, + // The send's steering transport, cloned from its `ActiveStream` slot. + // Attached to every chained turn so delivered texts reach the sink. + // + // @plan PLAN-20260905-STEERINT.P03 + // @requirement REQ-SI-002 + steering_queue: serdes_ai_agent::SteeringQueue, ) { let PreparedMessageContext { profile, @@ -357,6 +445,19 @@ pub(super) async fn run_stream_task( return; }; + // The send's transport is attached to every turn's run: texts the fork + // delivers mid-turn reach this sink, which persists them, resolves the + // queue head, and announces the delivery. + // + // @plan PLAN-20260905-STEERINT.P04 + // @requirement REQ-SI-001 + // @requirement REQ-SI-003 + let mid_turn_steering = Arc::new(AsyncMutex::new(MidTurnSteering::new( + conversation_service.clone(), + conversation_id, + steering_queues.clone(), + ))); + let delivery_ctx = SteeringDeliveryContext { finalize: StreamFinalizeContext { conversation_service: &conversation_service, @@ -371,37 +472,24 @@ pub(super) async fn run_stream_task( // Bound outside the turn runner so the future it returns has one type for // every turn, independent of the call that produced it. - let client = &client; - let agent = &agent; - let diagnostics_context = &diagnostics_context; - let tx = &tx; - run_steered_turns_and_finalize( &delivery_ctx, compression_result, messages, - move |turn_messages: Vec| { - let context = build_stream_context( - conversation_id, - view_tx.clone(), - approval_gate.clone(), - policy.clone(), - skills_service.clone(), - filter_emoji, - ); - async move { - stream_agent_response( - client, - agent, - &turn_messages, - context, - diagnostics_context, - conversation_id, - tx, - ) - .await - } - }, + steered_turn_runner( + &client, + &agent, + &diagnostics_context, + conversation_id, + &tx, + view_tx, + approval_gate, + policy, + skills_service, + filter_emoji, + &steering_queue, + &mid_turn_steering, + ), ) .await; } diff --git a/src/services/chat_impl/streaming/steering_delivery.rs b/src/services/chat_impl/streaming/steering_delivery.rs index 92d2672..b2bc7ff 100644 --- a/src/services/chat_impl/streaming/steering_delivery.rs +++ b/src/services/chat_impl/streaming/steering_delivery.rs @@ -1,12 +1,15 @@ -//! Delivery of queued steering messages at the turn boundary (issue #222). +//! Chaining of a send's follow-up turns while steering stays queued. //! -//! A user send runs one `AgentStream` turn. When that turn finishes cleanly -//! and the conversation has steering messages waiting, this module runs -//! another turn seeded with the conversation so far plus the steering text — -//! no new user-initiated send, and nothing cancelled. The chain ends when the -//! queue is empty, the turn did not finish cleanly, the turn cap is hit, or a -//! steering message cannot be recorded; finalization then runs once, on the -//! last turn. +//! A user send runs one `AgentStream` turn. The transport delivers queued +//! steering itself, mid-turn at each tool-call boundary, and PA's sink +//! (`steering_sink`) persists and resolves what was delivered. When a turn +//! still finishes cleanly with entries left queued — a turn that never +//! crossed a tool boundary, or steers that arrived after its last one — this +//! module chains another turn, so the transport gets a boundary to deliver +//! them at. The texts are never written into the history here: the transport +//! owns them, and taking them away from it would strand both copies. The +//! chain ends when the queue is empty, the turn did not finish cleanly, or +//! the turn cap is hit; finalization then runs once, on the last turn. //! //! Every ending that leaves messages queued announces them as discarded. A //! queued entry is on screen until something says what became of it, and @@ -14,21 +17,23 @@ //! //! @plan PLAN-20260903-ISSUE222.P02 //! @plan PLAN-20260903-ISSUE222.P06 +//! @plan PLAN-20260905-STEERINT.P04 //! @requirement REQ-222-003 //! @requirement REQ-222-005 //! @requirement REQ-222-006 //! @requirement REQ-222-007 +//! @requirement REQ-SI-002 +//! @requirement REQ-SI-006 use super::{ - finalize_by_outcome, finalize_completed_turn, has_assistant_output, persist_assistant_response, - StreamFinalizeContext, StreamTranscript, + finalize_by_outcome, has_assistant_output, persist_assistant_response, StreamFinalizeContext, + StreamTranscript, }; use crate::compression::pipeline::CompressionResult; -use crate::events::types::ChatEvent; -use crate::events::{emit, AppEvent}; use crate::llm::Message as LlmMessage; -use crate::models::Message; -use crate::services::chat_impl::{drain_steering_queue, emit_steering_discarded, QueuedSteering}; +use crate::services::chat_impl::{ + drain_steering_queue, emit_steering_discarded, has_queued_steering, +}; use std::future::Future; use tokio_util::sync::CancellationToken; @@ -59,16 +64,18 @@ pub(in crate::services::chat_impl) struct SteeringDeliveryContext<'a> { /// /// `run_turn` is handed the history for a turn and yields that turn's /// transcript. Intermediate turns persist their own assistant output here so -/// it is ordered before the steering message that follows it; the last turn's -/// output is normally persisted by `finalize_by_outcome`, so nothing is -/// written twice. The one chain that ends between those two writes reports -/// itself with [`ChainOutcome::OutputPersisted`] and finalizes without the -/// second, which is what keeps that guarantee true. +/// it is ordered before the steering text the transport delivers during the +/// turn that follows; the last turn's output is persisted by +/// `finalize_by_outcome`, so nothing is written twice. Queued steering texts +/// are never seeded into the history: the transport delivers them mid-turn +/// and the sink records them (REQ-SI-002). /// /// @plan PLAN-20260903-ISSUE222.P02 /// @plan PLAN-20260903-ISSUE222.P06 +/// @plan PLAN-20260905-STEERINT.P04 /// @requirement REQ-222-005 /// @requirement REQ-222-007 +/// @requirement REQ-SI-002 pub(in crate::services::chat_impl) async fn run_steered_turns_and_finalize( ctx: &SteeringDeliveryContext<'_>, compression_result: CompressionResult, @@ -78,51 +85,37 @@ pub(in crate::services::chat_impl) async fn run_steered_turns_and_finalize) -> Fut, Fut: Future, { - let (transcript, outcome) = run_steered_turns(ctx, messages, run_turn).await; - match outcome { - ChainOutcome::Unfinalized => { - finalize_by_outcome( - &ctx.finalize, - compression_result, - transcript, - ctx.model_label, - ) - .await; - } - ChainOutcome::OutputPersisted => { - finalize_completed_turn(&ctx.finalize, compression_result, &transcript).await; - } - } -} - -/// What the chain has left for finalization to do with the last turn. -/// -/// @plan PLAN-20260903-ISSUE222.P06 -/// @requirement REQ-222-007 -enum ChainOutcome { - /// Nothing about the last turn has been recorded, so finalization decides - /// what to persist from the transcript itself. This is every ordinary - /// ending: the queue ran dry, the turn did not finish cleanly, or the - /// chain hit its turn cap. - Unfinalized, - /// The last turn's assistant output is already recorded, because the - /// chain stopped after writing it and before the follow-up turn it was - /// written for could start. The rest of finalization still applies; - /// writing that output again would record it twice. - OutputPersisted, + let transcript = run_steered_turns(ctx, messages, run_turn).await; + finalize_by_outcome( + &ctx.finalize, + compression_result, + transcript, + ctx.model_label, + ) + .await; } /// Run turns until the conversation stops steering, returning the last -/// turn's transcript and what finalization still owes it. +/// turn's transcript. +/// +/// The leftover check between turns is a peek, not a drain: the queued texts +/// are the transport's to deliver at the next turn's first tool boundary, +/// and taking them here would strand a text the transport still holds +/// (REQ-SI-006). Hitting the turn cap is the one ending that does take the +/// entries — no turn of this send is going to deliver them — so it announces +/// them discarded. /// /// @plan PLAN-20260903-ISSUE222.P02 +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-222-003 /// @requirement REQ-222-005 /// @requirement REQ-222-006 +/// @requirement REQ-SI-006 async fn run_steered_turns( ctx: &SteeringDeliveryContext<'_>, mut messages: Vec, mut run_turn: R, -) -> (StreamTranscript, ChainOutcome) +) -> StreamTranscript where R: FnMut(Vec) -> Fut, Fut: Future, @@ -137,23 +130,23 @@ where turns += 1; if !reaches_delivery_boundary(&transcript, ctx.cancel) { - return (transcript, ChainOutcome::Unfinalized); + return transcript; } - let queued = drain_steering_queue(ctx.finalize.steering_queues, conversation_id); - if queued.is_empty() { - return (transcript, ChainOutcome::Unfinalized); + if !has_queued_steering(ctx.finalize.steering_queues, conversation_id) { + return transcript; } if turns >= MAX_STEERING_TURNS { + let discarded = drain_steering_queue(ctx.finalize.steering_queues, conversation_id); tracing::warn!( conversation_id = %conversation_id, turns, - dropped = queued.len(), + dropped = discarded.len(), "Steering chain reached its turn cap; dropping the messages still queued" ); - emit_steering_discarded(conversation_id, &queued); - return (transcript, ChainOutcome::Unfinalized); + emit_steering_discarded(conversation_id, &discarded); + return transcript; } persist_assistant_response( @@ -167,10 +160,6 @@ where if has_assistant_output(&transcript) { messages.push(assistant_message(&transcript)); } - - if !deliver_steering(ctx, &queued, &mut messages).await { - return (transcript, ChainOutcome::OutputPersisted); - } } } @@ -178,8 +167,8 @@ where /// delivered at. /// /// A turn that failed, that never completed, or whose send has been stopped -/// is over: it drains nothing and chains nothing (REQ-222-006). This is also -/// the re-check that runs before every follow-up turn starts. +/// is over: it delivers nothing and chains nothing (REQ-222-006). This is +/// also the re-check that runs before every follow-up turn starts. /// /// @plan PLAN-20260903-ISSUE222.P02 /// @requirement REQ-222-006 @@ -187,56 +176,6 @@ fn reaches_delivery_boundary(transcript: &StreamTranscript, cancel: &Cancellatio transcript.completed && transcript.error.is_none() && !cancel.is_cancelled() } -/// Persist, announce, and seed each drained steering message in FIFO order. -/// -/// Returns `false` when a steering message could not be persisted, which -/// stops the chain. Recording the steer is what keeps the history a chained -/// turn is seeded with equal to the history a reload rebuilds; a turn run -/// over text the store rejected would break that for the rest of the -/// conversation. So the failed message is not seeded, is not announced as -/// delivered, and neither is anything behind it: they are announced as -/// discarded instead, because no later turn is going to pick them up. -/// -/// @plan PLAN-20260903-ISSUE222.P02 -/// @plan PLAN-20260903-ISSUE222.P06 -/// @requirement REQ-222-003 -/// @requirement REQ-222-005 -/// @requirement REQ-222-007 -async fn deliver_steering( - ctx: &SteeringDeliveryContext<'_>, - queued: &[QueuedSteering], - messages: &mut Vec, -) -> bool { - let conversation_id = ctx.finalize.conversation_id; - for (index, entry) in queued.iter().enumerate() { - if let Err(error) = ctx - .finalize - .conversation_service - .add_message(conversation_id, Message::user(entry.text.clone())) - .await - { - // ServiceError's Display never carries user content; the steering - // text itself is intentionally not logged. - tracing::warn!( - conversation_id = %conversation_id, - steer_id = %entry.id, - error = %error, - "Failed to persist a delivered steering message" - ); - emit_steering_discarded(conversation_id, &queued[index..]); - return false; - } - - let _ = emit(AppEvent::Chat(ChatEvent::SteeringDelivered { - conversation_id, - steer_id: entry.id, - })); - - messages.push(LlmMessage::user(entry.text.clone())); - } - true -} - /// The finished turn's assistant output, shaped the way /// `ChatServiceImpl::build_llm_messages` shapes a persisted assistant /// message, so the chained turn sees the history a reload would rebuild. diff --git a/src/services/chat_impl/streaming/steering_sink.rs b/src/services/chat_impl/streaming/steering_sink.rs new file mode 100644 index 0000000..2b83b3c --- /dev/null +++ b/src/services/chat_impl/streaming/steering_sink.rs @@ -0,0 +1,99 @@ +//! The services-layer sink for steering texts a turn delivered mid-run. +//! +//! The fork hands a delivered text to [`SteeringDeliverySink::delivered`] the +//! moment the model actually received it. This type is that sink: it persists +//! the text as a user row, resolves the conversation's queue entry by FIFO +//! position, and announces the resolution. Storage and the UI stay out of the +//! LLM layer; the trait object points the dependency this way only. +//! +//! @plan PLAN-20260905-STEERINT.P04 +//! @requirement REQ-SI-003 +//! @requirement REQ-SI-005 + +use std::sync::Arc; + +use async_trait::async_trait; +use uuid::Uuid; + +use crate::events::types::ChatEvent; +use crate::events::{emit, AppEvent}; +use crate::llm::error::LlmError; +use crate::llm::steering::SteeringDeliverySink; +use crate::models::Message; +use crate::services::ConversationService; + +use super::super::{emit_steering_discarded, pop_steering_head, SteeringQueues}; + +/// Persists and correlates the steering texts one send's turns delivered. +/// +/// Correlation is by FIFO position, never by matching text: the fork drains +/// its transport in the order acceptance pushed, so the queue head is always +/// the entry whose text was just delivered, and two identical texts still +/// resolve to their own distinct entries. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-003 +/// @requirement REQ-SI-005 +pub(in crate::services::chat_impl) struct MidTurnSteering { + conversation_service: Arc, + conversation_id: Uuid, + steering_queues: SteeringQueues, + /// Number of delivered texts persisted during this send. + pub(in crate::services::chat_impl) delivered: usize, +} + +impl MidTurnSteering { + /// A sink for one send, over its conversation's registries. + pub(in crate::services::chat_impl) fn new( + conversation_service: Arc, + conversation_id: Uuid, + steering_queues: SteeringQueues, + ) -> Self { + Self { + conversation_service, + conversation_id, + steering_queues, + delivered: 0, + } + } +} + +#[async_trait] +impl SteeringDeliverySink for MidTurnSteering { + async fn delivered(&mut self, text: &str) -> Result<(), LlmError> { + if let Err(error) = self + .conversation_service + .add_message(self.conversation_id, Message::user(text.to_string())) + .await + { + // ServiceError's Display never carries user content; the steering + // text itself is intentionally not logged. + tracing::warn!( + conversation_id = %self.conversation_id, + error = %error, + "Failed to persist a steering text the model already received" + ); + // The entry this delivery was for is the head by FIFO position. + // The turn is about to fail, so it is taken and announced here + // rather than left for teardown to announce a second time. + if let Some(entry) = pop_steering_head(&self.steering_queues, self.conversation_id) { + emit_steering_discarded(self.conversation_id, &[entry]); + } + return Err(LlmError::Stream(format!( + "failed to persist delivered steering: {error}" + ))); + } + + self.delivered += 1; + // An empty deque means teardown raced the delivery and already + // announced the discard: the row is persisted and recorded, but no + // terminal event is emitted and nothing is treated as an error. + if let Some(entry) = pop_steering_head(&self.steering_queues, self.conversation_id) { + let _ = emit(AppEvent::Chat(ChatEvent::SteeringDelivered { + conversation_id: self.conversation_id, + steer_id: entry.id, + })); + } + Ok(()) + } +} diff --git a/src/services/chat_impl/support.rs b/src/services/chat_impl/support.rs index f1c0cbb..317d5df 100644 --- a/src/services/chat_impl/support.rs +++ b/src/services/chat_impl/support.rs @@ -4,6 +4,7 @@ use crate::models::{AuthConfig, Message, ModelParameters}; use crate::services::{AppSettingsService, ServiceError, ServiceResult}; use futures::StreamExt; use std::collections::HashMap; +use std::sync::atomic::{AtomicUsize, Ordering}; use std::sync::Arc; use tokio::sync::RwLock; @@ -21,6 +22,9 @@ pub(super) struct MockConversationService { /// When set, `add_message` returns a storage error instead of recording /// the message, exercising persistence-failure logging paths. pub(super) add_message_fails: Arc>, + /// How many times `add_message` was called, failed attempts included, so + /// tests can tell a single refused write from a retry loop. + pub(super) add_message_attempts: Arc, } pub(super) struct InMemoryAppSettingsService { @@ -226,9 +230,15 @@ impl MockConversationService { rename_calls: Arc::new(RwLock::new(Vec::new())), load_missing: Arc::new(RwLock::new(false)), add_message_fails: Arc::new(RwLock::new(false)), + add_message_attempts: Arc::new(AtomicUsize::new(0)), } } + /// The number of `add_message` calls so far, failed attempts included. + pub(super) fn add_message_attempts(&self) -> usize { + self.add_message_attempts.load(Ordering::SeqCst) + } + pub(super) async fn set_load_missing(&self, missing: bool) { *self.load_missing.write().await = missing; } @@ -288,6 +298,7 @@ impl super::super::ConversationService for MockConversationService { _conversation_id: Uuid, message: Message, ) -> Result { + self.add_message_attempts.fetch_add(1, Ordering::SeqCst); if *self.add_message_fails.read().await { return Err(crate::services::ServiceError::Storage( "simulated add_message persistence failure".to_string(), diff --git a/src/services/chat_impl/tests/steering.rs b/src/services/chat_impl/tests/steering.rs index f7ee387..6af7b1e 100644 --- a/src/services/chat_impl/tests/steering.rs +++ b/src/services/chat_impl/tests/steering.rs @@ -22,6 +22,7 @@ use uuid::Uuid; // Import test support utilities from parent tests module (support.rs) use super::chat_test_support::{MockConversationService, MockProfileService}; +mod transport; mod withdrawal; fn make_test_chat_service() -> ChatServiceImpl { diff --git a/src/services/chat_impl/tests/steering/transport.rs b/src/services/chat_impl/tests/steering/transport.rs new file mode 100644 index 0000000..76c0652 --- /dev/null +++ b/src/services/chat_impl/tests/steering/transport.rs @@ -0,0 +1,285 @@ +//! A steer's acceptance commits through the send's transport queue. +//! +//! The service-level queue stays the state machine for steering: cap, +//! announcements, withdrawal, teardown. The fork's `SteeringQueue` is the +//! transport, and the only thing that moves text into a model request. The +//! two are written in a fixed order — PA queue, `SteeringQueued`, the +//! confirm re-check, and only then the transport commit — because the +//! commit is the point of no return: once the transport holds the text, a +//! later run on the same queue will deliver it, so anything that can still +//! refuse the steer must happen first. +//! +//! @plan PLAN-20260905-STEERINT.P03 +//! @requirement REQ-SI-002 +//! @requirement REQ-SI-007 + +use super::*; +use crate::services::chat_impl::steering::QueuedSteering; + +/// The transport queue a running send's slot holds. +/// +/// @plan PLAN-20260905-STEERINT.P03 +/// @requirement REQ-SI-002 +fn transport_of( + service: &ChatServiceImpl, + conversation_id: Uuid, +) -> serdes_ai_agent::SteeringQueue { + service + .steering_transport_for_test(conversation_id) + .expect("a running send must hold a transport queue") +} + +/// An accepted steer lands on both sides of the two-queue relationship: the +/// state machine keeps the entry, and the transport holds the text a run +/// will deliver at its next tool boundary. +/// +/// @plan PLAN-20260905-STEERINT.P03 +/// @requirement REQ-SI-002 +#[tokio::test] +async fn an_accepted_steer_is_committed_to_the_send_transport() { + let _steering_bus_guard = lock_steering_bus().await; + let service = make_test_chat_service(); + let conversation_id = Uuid::new_v4(); + + service + .begin_stream_for_test(conversation_id) + .expect("begin_stream should succeed"); + let transport = transport_of(&service, conversation_id); + + let steer_id = ChatService::steer( + &service, + conversation_id, + " mid-turn correction ".to_string(), + ) + .await + .expect("steering a running turn must be accepted"); + + assert_eq!( + transport.pending_len(), + 1, + "the transport must hold the committed text for the next run to deliver" + ); + assert_eq!( + drained_pairs(&service, conversation_id), + vec![(steer_id, "mid-turn correction".to_string())], + "the state machine must keep the accepted entry, trimmed, until delivery" + ); + + service.clear_all_streams_for_test(); +} + +/// The steer past the queue cap is refused before any commit: the transport +/// holds exactly the accepted entries, nothing more. +/// +/// @plan PLAN-20260905-STEERINT.P03 +/// @requirement REQ-SI-002 +#[tokio::test] +async fn steer_past_the_cap_never_reaches_the_transport() { + let _steering_bus_guard = lock_steering_bus().await; + let service = make_test_chat_service(); + let conversation_id = Uuid::new_v4(); + + service + .begin_stream_for_test(conversation_id) + .expect("begin_stream should succeed"); + let transport = transport_of(&service, conversation_id); + + for index in 0..MAX_QUEUED_STEERING_MESSAGES { + ChatService::steer(&service, conversation_id, format!("steer {index}")) + .await + .expect("steers up to the cap must be accepted"); + } + + ChatService::steer(&service, conversation_id, "one too many".to_string()) + .await + .expect_err("a steer past the cap must be rejected"); + + assert_eq!( + transport.pending_len(), + MAX_QUEUED_STEERING_MESSAGES, + "the refused steer must never reach the transport" + ); + + service.clear_all_streams_for_test(); +} + +/// A steer withdrawn by the confirm re-check never reaches the transport. +/// +/// The window the re-check exists for: the entry was inserted, the view was +/// told, and then the turn ended before the re-check ran. If acceptance +/// committed before that re-check, the text would sit in a transport a +/// later run of the same send could deliver — a ghost instruction the user +/// withdrew. The transport clone here is captured before the teardown so +/// the assertion observes the real queue object rather than its absence. +/// +/// @plan PLAN-20260905-STEERINT.P03 +/// @requirement REQ-SI-002 +#[tokio::test] +async fn a_withdrawn_steer_is_never_committed_to_the_transport() { + let _steering_bus_guard = lock_steering_bus().await; + let service = make_test_chat_service(); + let conversation_id = Uuid::new_v4(); + let steer_id = Uuid::new_v4(); + + service + .begin_stream_for_test(conversation_id) + .expect("begin_stream should succeed"); + let transport = transport_of(&service, conversation_id); + + // queue_steering has inserted the entry; nothing has been committed yet. + assert!( + service.push_steering( + conversation_id, + QueuedSteering { + id: steer_id, + stream_id: service.stream_id_for_test(conversation_id).unwrap(), + text: "take the other branch".to_string(), + }, + ), + "the queue has room, so the insert must take" + ); + + // The turn ends before the re-check runs. + service.clear_all_streams_for_test(); + service + .confirm_or_withdraw_steering(conversation_id, steer_id) + .expect_err("a steer with no turn left to deliver it must be refused"); + + assert_eq!( + transport.pending_len(), + 0, + "the withdrawal must never commit the text to the transport" + ); +} + +/// A commit whose send is gone between the re-check and the commit is +/// refused, takes its entry back off the queue, and announces the discard. +/// +/// The re-check passed a moment earlier; by the time the commit reads the +/// slot, the teardown has removed it and its transport with it. The model +/// can never be handed the text, so the entry must not stay queued, and the +/// view must be told what became of it. The refusal is the same +/// `no_active_turn` the re-check gives, so a caller cannot tell which side +/// of the window it landed on. +/// +/// @plan PLAN-20260905-STEERINT.P03 +/// @requirement REQ-SI-002 +/// @requirement REQ-SI-007 +#[tokio::test] +async fn a_commit_after_the_transport_is_gone_is_withdrawn_and_announced() { + let _steering_bus_guard = lock_steering_bus().await; + let service = make_test_chat_service(); + let conversation_id = Uuid::new_v4(); + let steer_id = Uuid::new_v4(); + let mut event_rx = subscribe(); + + service + .begin_stream_for_test(conversation_id) + .expect("begin_stream should succeed"); + + // The state the commit is called in: entry queued, re-check passed. + assert!( + service.push_steering( + conversation_id, + QueuedSteering { + id: steer_id, + stream_id: service.stream_id_for_test(conversation_id).unwrap(), + text: "too late".to_string(), + }, + ), + "the queue has room, so the insert must take" + ); + + // The send tears down before the commit runs, taking the slot and its + // transport with it. + service.clear_all_streams_for_test(); + + let error = service + .commit_steering_to_transport(conversation_id, steer_id, "too late") + .expect_err("a commit whose transport is gone must be refused"); + + assert!( + matches!(&error, ServiceError::Validation(message) + if message.contains("No active turn to steer")), + "the refusal must be the one the re-check gives, got: {error}" + ); + assert!( + drained_pairs(&service, conversation_id).is_empty(), + "the refused commit must take the entry back off the queue" + ); + + let events = + collect_chat_events(&mut event_rx, conversation_id, Duration::from_millis(100)).await; + assert_eq!( + discarded_ids(&events), + vec![steer_id], + "the view must be told the entry will never be delivered, got {events:?}" + ); +} + +/// A late insert from a finished send must not be confirmed against its replacement. +#[tokio::test] +async fn replacement_stream_cannot_confirm_an_old_steer() { + let _steering_bus_guard = lock_steering_bus().await; + let service = make_test_chat_service(); + let conversation_id = Uuid::new_v4(); + service.begin_stream_for_test(conversation_id).unwrap(); + let steer_id = service + .queue_steering(conversation_id, "old instruction") + .unwrap(); + let entry = service.drain_steering(conversation_id).pop().unwrap(); + ChatService::cancel(&service, conversation_id); + service.begin_stream_for_test(conversation_id).unwrap(); + assert!(service.push_steering(conversation_id, entry)); + let mut event_rx = subscribe(); + + let result = service.confirm_or_withdraw_steering(conversation_id, steer_id); + let remaining = drained_pairs(&service, conversation_id); + let pending = transport_of(&service, conversation_id).pending_len(); + service.clear_all_streams_for_test(); + + assert!( + result.is_err(), + "a replacement is not the turn the user steered" + ); + assert!(remaining.is_empty()); + assert_eq!(pending, 0); + let events = + collect_chat_events(&mut event_rx, conversation_id, Duration::from_millis(100)).await; + assert_eq!(discarded_ids(&events), vec![steer_id]); +} + +/// Replacement after confirmation must not redirect a commit into the new send. +#[tokio::test] +async fn replacement_stream_cannot_receive_an_old_steer_commit() { + let _steering_bus_guard = lock_steering_bus().await; + let service = make_test_chat_service(); + let conversation_id = Uuid::new_v4(); + service.begin_stream_for_test(conversation_id).unwrap(); + let steer_id = service + .queue_steering(conversation_id, "old instruction") + .unwrap(); + service + .confirm_or_withdraw_steering(conversation_id, steer_id) + .unwrap(); + let entry = service.drain_steering(conversation_id).pop().unwrap(); + ChatService::cancel(&service, conversation_id); + service.begin_stream_for_test(conversation_id).unwrap(); + assert!(service.push_steering(conversation_id, entry)); + let mut event_rx = subscribe(); + + let result = service.commit_steering_to_transport(conversation_id, steer_id, "old instruction"); + let pending = transport_of(&service, conversation_id).pending_len(); + let remaining = drained_pairs(&service, conversation_id); + service.clear_all_streams_for_test(); + + assert_eq!( + pending, 0, + "the new model must never receive the old instruction" + ); + assert!(result.is_err()); + assert!(remaining.is_empty()); + let events = + collect_chat_events(&mut event_rx, conversation_id, Duration::from_millis(100)).await; + assert_eq!(discarded_ids(&events), vec![steer_id]); +} diff --git a/src/services/chat_impl/tests/steering/withdrawal.rs b/src/services/chat_impl/tests/steering/withdrawal.rs index 9b535b9..5c12fac 100644 --- a/src/services/chat_impl/tests/steering/withdrawal.rs +++ b/src/services/chat_impl/tests/steering/withdrawal.rs @@ -345,6 +345,7 @@ async fn a_steer_drained_before_it_was_announced_still_reaches_a_terminal_event( conversation_id, QueuedSteering { id: steer_id, + stream_id, text: "take the other branch".to_string(), }, ), diff --git a/src/services/chat_impl/tests/steering_delivery.rs b/src/services/chat_impl/tests/steering_delivery.rs index 32bcce5..5dd972e 100644 --- a/src/services/chat_impl/tests/steering_delivery.rs +++ b/src/services/chat_impl/tests/steering_delivery.rs @@ -1,24 +1,32 @@ -//! Delivery of queued steering messages at the turn boundary. +//! Chaining of a send's follow-up turns while steering stays queued. //! -//! Covers issue #222: when a turn finishes with steering messages waiting, -//! the send chains another turn seeded with the conversation so far plus the -//! steering text. A cancelled or failed turn chains nothing, finalization -//! still happens exactly once, and the chain is bounded. +//! Covers issue #222 and its mid-turn steering integration: the transport +//! delivers a queued steer itself, at a tool-call boundary inside a running +//! turn, and the sink records it. When a turn still ends with entries +//! queued, the send chains a follow-up turn so the transport gets a boundary +//! to deliver them at. A cancelled or failed turn chains nothing, +//! finalization still happens exactly once, and the chain is bounded. //! -//! The turn itself is the only double here: each test hands the delivery loop +//! The turn itself is the only double here: each test hands the chaining loop //! a runner that records the history it was seeded with and replies with a //! scripted transcript, which is the same level the approval-gate tests in -//! `three_stream_concurrency.rs` operate at. +//! `three_stream_concurrency.rs` operate at. Where a test simulates the +//! delivery side, it hands the text to the real mid-turn sink, so the row +//! and the resolved queue entry are the production ones. //! //! @plan PLAN-20260903-ISSUE222.P02 +//! @plan PLAN-20260905-STEERINT.P04 //! @requirement REQ-222-005 //! @requirement REQ-222-006 //! @requirement REQ-222-007 //! @requirement REQ-222-008 +//! @requirement REQ-SI-002 +//! @requirement REQ-SI-006 use super::*; use crate::events::subscribe; use crate::events::types::ChatEvent; +use crate::llm::steering::SteeringDeliverySink; use crate::llm::Role as LlmRole; use crate::services::chat_impl::streaming::steering_delivery::{ run_steered_turns_and_finalize, SteeringDeliveryContext, MAX_STEERING_TURNS, @@ -28,7 +36,9 @@ use std::sync::atomic::{AtomicBool, Ordering}; use tokio::sync::Notify; use tokio::time::Duration; +mod approval; mod discard; +mod sink; /// The future a scripted turn runner hands back. /// @@ -351,6 +361,67 @@ fn recorded(log: &TurnLog) -> Vec> { log.lock().expect("turn log poisoned").clone() } +/// The mid-turn sink of a fixture's send, shared behind a lock so a turn's +/// future can hand it texts. +/// +/// @plan PLAN-20260905-STEERINT.P04 +type SharedSink = Arc>; + +/// A sink wired to the fixture's conversation service and steering queues. +fn sink_for(fixture: &DeliveryFixture) -> streaming::steering_sink::MidTurnSteering { + streaming::steering_sink::MidTurnSteering::new( + fixture.conversation_service.clone(), + fixture.conversation_id, + fixture.steering_queues.clone(), + ) +} + +/// Wrap a sink for handoff into turn futures. +fn shared_sink(sink: streaming::steering_sink::MidTurnSteering) -> SharedSink { + Arc::new(tokio::sync::Mutex::new(sink)) +} + +/// A runner that logs each turn's seed history, replies with the next +/// scripted completed transcript, and on turn `deliver_on` (1-based) first +/// hands `texts` to the sink — what the transport does with the queued steers +/// at that turn's tool boundary. +/// +/// @plan PLAN-20260905-STEERINT.P04 +fn boundary_runner( + log: &TurnLog, + replies: &'static [&'static str], + sink: SharedSink, + deliver_on: usize, + texts: &'static [&'static str], +) -> impl FnMut(Vec) -> TurnFuture { + let log = log.clone(); + move |messages: Vec| { + let index = { + let mut turns = log.lock().expect("turn log poisoned"); + turns.push(messages); + turns.len() - 1 + }; + let sink = Arc::clone(&sink); + Box::pin(async move { + if index + 1 == deliver_on { + for text in texts { + sink.lock() + .await + .delivered(text) + .await + .expect("the boundary delivery must persist and resolve its entry"); + } + } + completed_turn( + replies + .get(index) + .copied() + .expect("the turn runner ran out of scripted replies"), + ) + }) as TurnFuture + } +} + /// An empty queue must leave a send exactly as it is today: one turn, one /// completion, and the assistant output persisted once. /// @@ -411,21 +482,37 @@ async fn empty_steering_queue_runs_a_single_turn_and_finalizes_once() { ); } -/// A queued steer chains a second turn whose history carries the steer as a -/// user message after the first turn's assistant output. +/// A text-only turn never crosses a tool boundary, so its steer is still +/// queued when the turn ends; the chain peeks at it and runs a follow-up +/// turn, and the transport's boundary in that turn delivers it. /// -/// @plan PLAN-20260903-ISSUE222.P02 -/// @requirement REQ-222-005 +/// The steer is never seeded into the follow-up turn's history: the seed +/// carries only the finished turn's assistant output, and the text reaches +/// the model through the sink at the boundary, persisted after that output. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-002 +/// @requirement REQ-SI-006 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn queued_steer_chains_a_second_turn_carrying_the_steer_text() { +async fn a_text_only_turn_leaves_its_steer_queued_and_the_chained_turn_delivers_it() { let _steering_bus_guard = lock_steering_bus().await; let fixture = DeliveryFixture::new(); - fixture.steer("focus on the parser").await; + let steer_id = fixture.steer("focus on the parser").await; let log = new_turn_log(); + let sink = shared_sink(sink_for(&fixture)); - fixture - .run_turns(scripted_runner(&log, &["first answer", "second answer"])) - .await; + let ((), events) = events_during(fixture.conversation_id, async { + fixture + .run_turns(boundary_runner( + &log, + &["first answer", "second answer"], + sink, + 2, + &["focus on the parser"], + )) + .await; + }) + .await; let turns = recorded(&log); assert_eq!(turns.len(), 2, "a queued steer must chain a second turn"); @@ -435,27 +522,59 @@ async fn queued_steer_chains_a_second_turn_carrying_the_steer_text() { (LlmRole::System, "be helpful".to_string()), (LlmRole::User, "original prompt".to_string()), (LlmRole::Assistant, "first answer".to_string()), - (LlmRole::User, "focus on the parser".to_string()), ], - "the second turn must be seeded with the finished turn plus the steer" + "PA must never seed the steer into the chained turn's history" + ); + assert_eq!( + delivered_ids(&events), + vec![steer_id], + "the chained turn's boundary must deliver the leftover steer, got {events:?}" + ); + assert_eq!( + persisted_shape(&fixture.conversations).await, + vec![ + (MessageRole::Assistant, "first answer".to_string()), + (MessageRole::User, "focus on the parser".to_string()), + (MessageRole::Assistant, "second answer".to_string()), + ], + "the delivered steer must sit between the two assistant outputs" + ); + assert_eq!( + completion_count(&events), + 1, + "the chained send must finalize once, got {events:?}" ); } -/// Two steers queued against the same boundary arrive in submission order. +/// Two steers queued behind a text-only turn are delivered in submission +/// order at the chained turn's boundary, and neither is seeded into the +/// turn's history by PA. /// /// @plan PLAN-20260903-ISSUE222.P02 -/// @requirement REQ-222-005 +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-002 +/// @requirement REQ-SI-005 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn two_queued_steers_are_delivered_in_fifo_order() { let _steering_bus_guard = lock_steering_bus().await; let fixture = DeliveryFixture::new(); - fixture.steer("first steer").await; - fixture.steer("second steer").await; + let first_id = fixture.steer("first steer").await; + let second_id = fixture.steer("second steer").await; let log = new_turn_log(); + let sink = shared_sink(sink_for(&fixture)); - fixture - .run_turns(scripted_runner(&log, &["first answer", "second answer"])) - .await; + let ((), events) = events_during(fixture.conversation_id, async { + fixture + .run_turns(boundary_runner( + &log, + &["first answer", "second answer"], + sink, + 2, + &["first steer", "second steer"], + )) + .await; + }) + .await; let turns = recorded(&log); assert_eq!( @@ -469,10 +588,13 @@ async fn two_queued_steers_are_delivered_in_fifo_order() { (LlmRole::System, "be helpful".to_string()), (LlmRole::User, "original prompt".to_string()), (LlmRole::Assistant, "first answer".to_string()), - (LlmRole::User, "first steer".to_string()), - (LlmRole::User, "second steer".to_string()), ], - "both steers must be seeded in FIFO order" + "the queued steers must be delivered by the transport, never seeded by PA" + ); + assert_eq!( + delivered_ids(&events), + vec![first_id, second_id], + "both steers must resolve their own entries, in FIFO order, got {events:?}" ); assert_eq!( persisted_shape(&fixture.conversations).await, @@ -486,34 +608,6 @@ async fn two_queued_steers_are_delivered_in_fifo_order() { ); } -/// Every delivered steer announces itself once, under the id its -/// `SteeringQueued` reported. -/// -/// @plan PLAN-20260903-ISSUE222.P02 -/// @requirement REQ-222-005 -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn steering_delivered_is_emitted_once_per_steer_with_the_queued_id() { - let _steering_bus_guard = lock_steering_bus().await; - let fixture = DeliveryFixture::new(); - let log = new_turn_log(); - - let ((first_id, second_id), events) = events_during(fixture.conversation_id, async { - let first_id = fixture.steer("first steer").await; - let second_id = fixture.steer("second steer").await; - fixture - .run_turns(scripted_runner(&log, &["first answer", "second answer"])) - .await; - (first_id, second_id) - }) - .await; - - assert_eq!( - delivered_ids(&events), - vec![first_id, second_id], - "each steer must be delivered once, under the id steer() returned, got {events:?}" - ); -} - /// A cancelled turn never reaches a delivery boundary: it drains nothing and /// chains nothing, even though it finished cleanly. /// @@ -622,14 +716,21 @@ async fn failed_turn_drains_nothing_and_chains_no_follow_up() { /// preceded it and the assistant output it produced, so a reload replays the /// turn in the order the user experienced it. /// +/// The chained turn's boundary delivers the steer through the sink, the loop +/// persists the finished turn's output before it, and finalization persists +/// the last turn's output after it. +/// /// @plan PLAN-20260903-ISSUE222.P02 +/// @plan PLAN-20260905-STEERINT.P04 /// @requirement REQ-222-007 +/// @requirement REQ-SI-002 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn steering_message_persists_between_the_assistant_outputs() { let _steering_bus_guard = lock_steering_bus().await; let fixture = DeliveryFixture::new(); fixture.steer("adjust course").await; let log = new_turn_log(); + let sink = shared_sink(sink_for(&fixture)); let log_for_runner = log.clone(); fixture @@ -639,23 +740,32 @@ async fn steering_message_persists_between_the_assistant_outputs() { turns.push(messages); turns.len() - 1 }; - let transcript = if index == 0 { - streaming::StreamTranscript { - response_text: "before the steer".to_string(), - thinking_text: "weighing options".to_string(), - tool_calls: vec![crate::llm::tools::ToolUse::new( - "call-1", - "web_search", - serde_json::json!({"query": "rust"}), - )], - tool_results: vec![crate::llm::tools::ToolResult::success("call-1", "ok")], - completed: true, - ..streaming::StreamTranscript::default() + let sink = Arc::clone(&sink); + Box::pin(async move { + if index == 0 { + streaming::StreamTranscript { + response_text: "before the steer".to_string(), + thinking_text: "weighing options".to_string(), + tool_calls: vec![crate::llm::tools::ToolUse::new( + "call-1", + "web_search", + serde_json::json!({"query": "rust"}), + )], + tool_results: vec![crate::llm::tools::ToolResult::success("call-1", "ok")], + completed: true, + ..streaming::StreamTranscript::default() + } + } else { + // The chained turn's boundary delivers the leftover steer + // through the sink, exactly as the transport would. + sink.lock() + .await + .delivered("adjust course") + .await + .expect("the boundary delivery must succeed"); + completed_turn("after the steer") } - } else { - completed_turn("after the steer") - }; - Box::pin(std::future::ready(transcript)) as TurnFuture + }) as TurnFuture }) .await; @@ -677,7 +787,13 @@ async fn steering_message_persists_between_the_assistant_outputs() { let turns = recorded(&log); let second = &turns[1]; - let preceding = &second[second.len() - 2]; + assert!( + !shape(second) + .iter() + .any(|(_, content)| content == "adjust course"), + "the steer must be delivered by the transport, never seeded by PA" + ); + let preceding = second.last().expect("the chained seed is never empty"); assert_eq!( preceding.thinking_content.as_deref(), Some("weighing options"), @@ -698,17 +814,25 @@ async fn steering_message_persists_between_the_assistant_outputs() { /// A chained send completes once, not once per turn. /// /// @plan PLAN-20260903-ISSUE222.P02 +/// @plan PLAN-20260905-STEERINT.P04 /// @requirement REQ-222-005 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn stream_completed_is_emitted_once_across_a_chained_send() { let _steering_bus_guard = lock_steering_bus().await; let fixture = DeliveryFixture::new(); let log = new_turn_log(); + let sink = shared_sink(sink_for(&fixture)); let ((), events) = events_during(fixture.conversation_id, async { fixture.steer("keep going").await; fixture - .run_turns(scripted_runner(&log, &["first answer", "second answer"])) + .run_turns(boundary_runner( + &log, + &["first answer", "second answer"], + sink, + 2, + &["keep going"], + )) .await; }) .await; @@ -733,12 +857,16 @@ async fn stream_completed_is_emitted_once_across_a_chained_send() { /// A user who keeps steering cannot chain turns forever: the send stops at /// `MAX_STEERING_TURNS`. /// -/// Every follow-up turn steers again through the real service path, which -/// only succeeds while the conversation still holds its stream slot — so this -/// also pins that the slot survives the chain. +/// Every turn delivers the steer queued before it at its boundary — what the +/// transport does — and then the user steers again through the real service +/// path, which only succeeds while the conversation still holds its stream +/// slot. So this also pins that the slot survives the chain, and that only +/// the entry still queued at the cap is announced discarded. /// /// @plan PLAN-20260903-ISSUE222.P02 +/// @plan PLAN-20260905-STEERINT.P04 /// @requirement REQ-222-005 +/// @requirement REQ-SI-006 #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn chain_stops_at_the_turn_cap_when_steering_keeps_refilling() { let _steering_bus_guard = lock_steering_bus().await; @@ -748,6 +876,7 @@ async fn chain_stops_at_the_turn_cap_when_steering_keeps_refilling() { let log_for_runner = log.clone(); let service = fixture.service.clone(); let conversation_id = fixture.conversation_id; + let sink = shared_sink(sink_for(&fixture)); let ((), events) = events_during(conversation_id, async { fixture @@ -758,7 +887,15 @@ async fn chain_stops_at_the_turn_cap_when_steering_keeps_refilling() { turns.len() }; let service = service.clone(); + let sink = Arc::clone(&sink); Box::pin(async move { + // The boundary delivers the steer the previous turn left + // queued, then the user steers again mid-turn. + sink.lock() + .await + .delivered("the steer the previous turn left queued") + .await + .expect("the boundary delivery must succeed"); ChatService::steer(service.as_ref(), conversation_id, format!("steer {index}")) .await .expect("the conversation must still be steerable mid-chain"); @@ -779,139 +916,9 @@ async fn chain_stops_at_the_turn_cap_when_steering_keeps_refilling() { 1, "a capped chain must still emit exactly one StreamCompleted, got {events:?}" ); -} - -/// A turn that stops at a real tool approval, and what it was told. -/// -/// The first turn takes the waiter, announces that it has reached the gate, -/// and blocks until someone decides. `decided` therefore distinguishes "the -/// steer resolved the approval" from "the user did". -/// -/// @plan PLAN-20260903-ISSUE222.P02 -/// @requirement REQ-222-008 -struct ApprovalProbe { - waiter: StdMutex>, - reached: tokio::sync::Notify, - decided: AtomicBool, - approved: AtomicBool, -} - -impl ApprovalProbe { - fn new(waiter: crate::llm::client_agent::ApprovalWaiter) -> Arc { - Arc::new(Self { - waiter: StdMutex::new(Some(waiter)), - reached: tokio::sync::Notify::new(), - decided: AtomicBool::new(false), - approved: AtomicBool::new(false), - }) - } - - /// A turn runner whose first turn blocks on the approval decision. - fn runner(self: &Arc, log: &TurnLog) -> impl FnMut(Vec) -> TurnFuture { - let probe = self.clone(); - let log = log.clone(); - move |messages: Vec| { - let index = { - let mut turns = log.lock().expect("turn log poisoned"); - turns.push(messages); - turns.len() - 1 - }; - let probe = probe.clone(); - Box::pin(async move { - if index == 0 { - probe.await_decision().await; - completed_turn("ran the tool") - } else { - completed_turn("followed the steer") - } - }) as TurnFuture - } - } - - async fn await_decision(&self) { - let pending = self - .waiter - .lock() - .expect("waiter slot poisoned") - .take() - .expect("the first turn owns the approval waiter"); - self.reached.notify_one(); - let decision = pending.wait().await.expect("the approval must resolve"); - self.approved.store(decision, Ordering::SeqCst); - self.decided.store(true, Ordering::SeqCst); - } -} - -/// A steer submitted while a tool approval is pending neither resolves nor -/// cancels that approval, and lands only after the decision lets the turn -/// reach its boundary. -/// -/// @plan PLAN-20260903-ISSUE222.P02 -/// @requirement REQ-222-008 -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn steer_during_a_pending_approval_waits_for_the_decision() { - let _steering_bus_guard = lock_steering_bus().await; - let fixture = DeliveryFixture::new(); - let log = new_turn_log(); - let request_id = Uuid::new_v4().to_string(); - let probe = ApprovalProbe::new(fixture.service.approval_gate.wait_for_approval( - request_id.clone(), - "WriteFile".to_string(), - fixture.conversation_id, - )); - let steer_id = StdMutex::new(None); - - let (((), ()), events) = events_during(fixture.conversation_id, async { - tokio::join!(fixture.run_turns(probe.runner(&log)), async { - probe.reached.notified().await; - *steer_id.lock().expect("steer slot poisoned") = - Some(fixture.steer("write the other file instead").await); - - // Let the waiting turn observe a decision it must not have been - // given. - tokio::task::yield_now().await; - assert!( - !probe.decided.load(Ordering::SeqCst), - "a steer must not resolve a pending tool approval" - ); - - fixture - .service - .resolve_tool_approval(request_id.clone(), ToolApprovalResponseAction::ProceedOnce) - .await - .expect("resolving the approval must succeed"); - }) - }) - .await; - - assert!( - probe.approved.load(Ordering::SeqCst), - "the approval must resolve from the user's decision, approved" - ); - assert!( - events - .iter() - .all(|event| !matches!(event, ChatEvent::StreamCancelled { .. })), - "a steer must not cancel the turn holding the approval, got {events:?}" - ); - - let turns = recorded(&log); - assert_eq!( - turns.len(), - 2, - "the steer must be delivered after the approved turn reaches its boundary" - ); assert_eq!( - shape(&turns[1]).last().cloned(), - Some((LlmRole::User, "write the other file instead".to_string())), - "the chained turn must carry the steer queued during the approval" - ); - assert_eq!( - delivered_ids(&events), - vec![steer_id - .lock() - .expect("steer slot poisoned") - .expect("the driver must have queued a steer")], - "the steer must be announced as delivered exactly once, got {events:?}" + discarded_ids(&events).len(), + 1, + "only the entry still queued at the cap must be announced discarded, got {events:?}" ); } diff --git a/src/services/chat_impl/tests/steering_delivery/approval.rs b/src/services/chat_impl/tests/steering_delivery/approval.rs new file mode 100644 index 0000000..1f1fad8 --- /dev/null +++ b/src/services/chat_impl/tests/steering_delivery/approval.rs @@ -0,0 +1,168 @@ +//! A steer accepted while a tool approval is pending, and the chained +//! turn that delivers it once the turn reaches its boundary. +//! +//! @plan PLAN-20260905-STEERINT.P04 +//! @requirement REQ-222-008 +//! @requirement REQ-SI-002 + +use super::*; + +/// A turn that stops at a real tool approval, and what it was told. +/// +/// The first turn takes the waiter, announces that it has reached the gate, +/// and blocks until someone decides. `decided` therefore distinguishes "the +/// steer resolved the approval" from "the user did". +/// +/// @plan PLAN-20260903-ISSUE222.P02 +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-222-008 +struct ApprovalProbe { + waiter: StdMutex>, + reached: tokio::sync::Notify, + decided: AtomicBool, + approved: AtomicBool, +} + +impl ApprovalProbe { + fn new(waiter: crate::llm::client_agent::ApprovalWaiter) -> Arc { + Arc::new(Self { + waiter: StdMutex::new(Some(waiter)), + reached: tokio::sync::Notify::new(), + decided: AtomicBool::new(false), + approved: AtomicBool::new(false), + }) + } + + /// A turn runner whose first turn blocks on the approval decision, and + /// whose chained turn delivers the queued steer at its boundary. + fn runner( + self: &Arc, + log: &TurnLog, + boundary_delivery: Option<(SharedSink, &'static str)>, + ) -> impl FnMut(Vec) -> TurnFuture { + let probe = self.clone(); + let log = log.clone(); + move |messages: Vec| { + let index = { + let mut turns = log.lock().expect("turn log poisoned"); + turns.push(messages); + turns.len() - 1 + }; + let probe = probe.clone(); + let boundary_delivery = boundary_delivery.clone(); + Box::pin(async move { + if index == 0 { + probe.await_decision().await; + completed_turn("ran the tool") + } else { + if let Some((sink, text)) = &boundary_delivery { + sink.lock() + .await + .delivered(text) + .await + .expect("the boundary delivery must succeed"); + } + completed_turn("followed the steer") + } + }) as TurnFuture + } + } + + async fn await_decision(&self) { + let pending = self + .waiter + .lock() + .expect("waiter slot poisoned") + .take() + .expect("the first turn owns the approval waiter"); + self.reached.notify_one(); + let decision = pending.wait().await.expect("the approval must resolve"); + self.approved.store(decision, Ordering::SeqCst); + self.decided.store(true, Ordering::SeqCst); + } +} + +/// A steer submitted while a tool approval is pending neither resolves nor +/// cancels that approval, and lands only after the decision lets the turn +/// reach its boundary. +/// +/// @plan PLAN-20260903-ISSUE222.P02 +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-222-008 +/// @requirement REQ-SI-002 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn steer_during_a_pending_approval_waits_for_the_decision() { + let _steering_bus_guard = lock_steering_bus().await; + let fixture = DeliveryFixture::new(); + let log = new_turn_log(); + let request_id = Uuid::new_v4().to_string(); + let probe = ApprovalProbe::new(fixture.service.approval_gate.wait_for_approval( + request_id.clone(), + "WriteFile".to_string(), + fixture.conversation_id, + )); + let steer_id = StdMutex::new(None); + let sink = shared_sink(sink_for(&fixture)); + + let (((), ()), events) = events_during(fixture.conversation_id, async { + tokio::join!( + fixture.run_turns(probe.runner(&log, Some((sink, "write the other file instead")),)), + async { + probe.reached.notified().await; + *steer_id.lock().expect("steer slot poisoned") = + Some(fixture.steer("write the other file instead").await); + + // Let the waiting turn observe a decision it must not have + // been given. + tokio::task::yield_now().await; + assert!( + !probe.decided.load(Ordering::SeqCst), + "a steer must not resolve a pending tool approval" + ); + + fixture + .service + .resolve_tool_approval( + request_id.clone(), + ToolApprovalResponseAction::ProceedOnce, + ) + .await + .expect("resolving the approval must succeed"); + } + ) + }) + .await; + + assert!( + probe.approved.load(Ordering::SeqCst), + "the approval must resolve from the user's decision, approved" + ); + assert!( + events + .iter() + .all(|event| !matches!(event, ChatEvent::StreamCancelled { .. })), + "a steer must not cancel the turn holding the approval, got {events:?}" + ); + + let turns = recorded(&log); + assert_eq!( + turns.len(), + 2, + "the steer must be delivered after the approved turn reaches its boundary" + ); + assert!( + !shape(&turns[1]) + .iter() + .any(|(_, content)| content == "write the other file instead"), + "the steer queued during the approval must be delivered by the transport, \ + never seeded by PA" + ); + assert_eq!( + delivered_ids(&events), + vec![steer_id + .lock() + .expect("steer slot poisoned") + .expect("the driver must have queued a steer")], + "the steer must be announced as delivered exactly once, got {events:?}" + ); +} diff --git a/src/services/chat_impl/tests/steering_delivery/discard.rs b/src/services/chat_impl/tests/steering_delivery/discard.rs index 707a0a4..a7a0b30 100644 --- a/src/services/chat_impl/tests/steering_delivery/discard.rs +++ b/src/services/chat_impl/tests/steering_delivery/discard.rs @@ -5,11 +5,11 @@ //! says that. So must every path that ends a turn with entries still queued, //! or the user is left watching an instruction that will never be acted on. //! -//! Two of those paths are driven here: a steer accepted in the window between -//! the delivery loop's last drain and the release of the stream slot, and a -//! steer whose own persistence fails. Both are driven from inside a -//! conversation-service write, so the window is a fact of the call order -//! rather than something a sleep hopes to hit. +//! The path driven here: a steer accepted in the window between the chaining +//! loop's last peek and the release of the stream slot. It is driven from +//! inside a conversation-service write, so the window is a fact of the call +//! order rather than something a sleep hopes to hit. (A steer whose own +//! persistence fails is the sink's to announce; see the sink tests.) //! //! @plan PLAN-20260903-ISSUE222.P06 //! @requirement REQ-222-003 @@ -79,36 +79,22 @@ impl LateSteer { } } -/// What the interposed conversation service does to the turn writing through -/// it. -enum Interference { - /// Refuse to persist user messages. Assistant output still lands, so the - /// chain is intact right up to the steering message that cannot be - /// recorded. - RefuseUserMessages, - /// Steer the conversation from inside `update_context_state`, which - /// finalization calls after the delivery loop's final drain and before it - /// releases the stream slot. - SteerDuringTeardown(Arc), -} - -/// A conversation service that delegates to the mock and interferes with one -/// write on the way through. +/// A conversation service that delegates to the mock and steers the +/// conversation from inside `update_context_state`, which finalization calls +/// after the chaining loop's final peek and before it releases the stream +/// slot. struct InterferingConversationService { inner: Arc, - interference: Interference, + late: Arc, } impl InterferingConversationService { /// Wrap `inner`, ready to hand to a fixture as its conversation service. fn interposed( inner: Arc, - interference: Interference, + late: Arc, ) -> Arc { - Arc::new(Self { - inner, - interference, - }) + Arc::new(Self { inner, late }) } } @@ -139,13 +125,6 @@ impl crate::services::ConversationService for InterferingConversationService { conversation_id: Uuid, message: Message, ) -> Result { - if matches!(self.interference, Interference::RefuseUserMessages) - && message.role == MessageRole::User - { - return Err(ServiceError::Storage( - "simulated steering persistence failure".to_string(), - )); - } self.inner.add_message(conversation_id, message).await } @@ -167,9 +146,7 @@ impl crate::services::ConversationService for InterferingConversationService { id: Uuid, state: &crate::models::ContextState, ) -> Result<(), ServiceError> { - if let Interference::SteerDuringTeardown(late) = &self.interference { - late.fire().await; - } + self.late.fire().await; self.inner.update_context_state(id, state).await } @@ -214,8 +191,9 @@ impl crate::services::ConversationService for InterferingConversationService { /// still `Running`, is announced as discarded rather than dropped. /// /// The steer is submitted from inside `update_context_state`: finalization -/// calls that after the final drain and before it releases the stream slot, -/// which is precisely the window the entry used to vanish in. The service +/// calls that after the chaining loop's final peek and before it releases +/// the stream slot, which is precisely the window the entry used to vanish +/// in. The service /// accepts it — `SteeringQueued` proves the view was told to render it — and /// then nothing is ever going to deliver it, so something has to say so. /// @@ -228,10 +206,7 @@ async fn a_steer_accepted_during_teardown_is_announced_as_discarded() { let late = LateSteer::new("too late to be delivered"); let fixture = DeliveryFixture::with_conversation_service( conversations.clone(), - InterferingConversationService::interposed( - conversations, - Interference::SteerDuringTeardown(late.clone()), - ), + InterferingConversationService::interposed(conversations, late.clone()), ); late.attach(&fixture); let log = new_turn_log(); @@ -272,68 +247,3 @@ async fn a_steer_accepted_during_teardown_is_announced_as_discarded() { "finalization must still release the conversation's stream slot" ); } - -/// A steering message the store refuses is not chained over. -/// -/// Persisting the steer is what keeps the chained history equal to what a -/// reload rebuilds. When that write fails, seeding the model with the text -/// anyway would run a turn over a conversation the database does not have, so -/// the chain stops here. The entry is announced as discarded, not delivered: -/// the view withdraws it either way, and reporting delivery would claim the -/// instruction reached the model when it did not. -/// -/// @plan PLAN-20260903-ISSUE222.P06 -/// @requirement REQ-222-003 -/// @requirement REQ-222-007 -#[tokio::test(flavor = "multi_thread", worker_threads = 2)] -async fn a_steer_that_cannot_be_persisted_is_discarded_and_chains_nothing() { - let _steering_bus_guard = lock_steering_bus().await; - let conversations = Arc::new(MockConversationService::new(Uuid::new_v4())); - let fixture = DeliveryFixture::with_conversation_service( - conversations.clone(), - InterferingConversationService::interposed(conversations, Interference::RefuseUserMessages), - ); - let log = new_turn_log(); - - // One scripted reply: a follow-up turn would run out and panic, so - // "chains nothing" is enforced by the runner as well as asserted. - let ((first, second), events) = events_during(fixture.conversation_id, async { - let first = fixture.steer("first steer").await; - let second = fixture.steer("second steer").await; - fixture - .run_turns(scripted_runner(&log, &["only answer"])) - .await; - (first, second) - }) - .await; - - assert_eq!( - recorded(&log).len(), - 1, - "a steer the store refused must not seed a follow-up turn" - ); - assert!( - delivered_ids(&events).is_empty(), - "a steer that was never recorded must not be reported delivered, got {events:?}" - ); - assert_eq!( - discarded_ids(&events), - vec![first, second], - "the refused steer and every entry behind it must be announced as discarded, \ - got {events:?}" - ); - assert_eq!( - persisted_shape(&fixture.conversations).await, - vec![(MessageRole::Assistant, "only answer".to_string())], - "only the assistant output the turn actually produced may be recorded, and only once" - ); - assert_eq!( - completion_count(&events), - 1, - "the turn itself finished, so the send must still finalize once, got {events:?}" - ); - assert!( - !fixture.is_streaming(), - "finalization must still release the conversation's stream slot" - ); -} diff --git a/src/services/chat_impl/tests/steering_delivery/sink.rs b/src/services/chat_impl/tests/steering_delivery/sink.rs new file mode 100644 index 0000000..fcba7ac --- /dev/null +++ b/src/services/chat_impl/tests/steering_delivery/sink.rs @@ -0,0 +1,230 @@ +//! The `MidTurnSteering` sink: what happens to a text the model received. +//! +//! A delivery is a promise the view is already rendering: the steer leaves the +//! queue only when the row is persisted, and the `SteeringDelivered` it emits +//! names the entry resolved by FIFO position, never by matching text. The +//! duplicate-text test is the one that guards that rule: under any correlation +//! that looks entries up by text it passes, if at all, only by luck. +//! +//! @plan PLAN-20260905-STEERINT.P04 +//! @requirement REQ-SI-003 +//! @requirement REQ-SI-005 + +use super::*; + +use crate::llm::steering::SteeringDeliverySink; + +/// Queue an entry directly, without announcing it, under a caller-chosen id. +fn queue_entry(fixture: &DeliveryFixture, id: Uuid, text: &str) { + assert!( + fixture.service.push_steering( + fixture.conversation_id, + QueuedSteering { + id, + stream_id: fixture.stream_id, + text: text.to_string(), + }, + ), + "the fixture's queue must accept the entry" + ); +} + +/// Two deliveries of byte-identical text resolve the two distinct entries they +/// were queued under, in push order. +/// +/// The ids are the assertion: only position-based correlation can return them +/// in the order the entries were pushed, because the texts cannot tell the +/// entries apart. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-005 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn duplicate_texts_resolve_to_distinct_entries_in_push_order() { + let _steering_bus_guard = lock_steering_bus().await; + let fixture = DeliveryFixture::new(); + let first = Uuid::new_v4(); + let second = Uuid::new_v4(); + queue_entry(&fixture, first, "same text"); + queue_entry(&fixture, second, "same text"); + let mut sink = sink_for(&fixture); + + let ((), events) = events_during(fixture.conversation_id, async { + sink.delivered("same text") + .await + .expect("the first delivery must persist and resolve the head"); + sink.delivered("same text") + .await + .expect("the second delivery must persist and resolve the next entry"); + }) + .await; + + assert_eq!( + delivered_ids(&events), + vec![first, second], + "each delivery must resolve its own queue entry, in push order, got {events:?}" + ); + assert!( + discarded_ids(&events).is_empty(), + "a successful delivery must discard nothing, got {events:?}" + ); + assert_eq!( + persisted_shape(&fixture.conversations).await, + vec![ + (MessageRole::User, "same text".to_string()), + (MessageRole::User, "same text".to_string()), + ], + "both delivered texts must be persisted as user rows, in delivery order" + ); +} + +/// A delivery the store refuses returns an error, announces that entry as +/// discarded, writes no row, and never tries the write again. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-003 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_delivery_that_cannot_be_persisted_fails_and_discards_without_retry() { + let _steering_bus_guard = lock_steering_bus().await; + let fixture = DeliveryFixture::new(); + let steer_id = Uuid::new_v4(); + queue_entry(&fixture, steer_id, "do the thing"); + fixture.conversations.set_add_message_failure(true).await; + let mut sink = sink_for(&fixture); + + let (result, events) = events_during(fixture.conversation_id, async { + sink.delivered("do the thing").await + }) + .await; + + assert!( + result.is_err(), + "a refused persistence must fail the delivery, and with it the turn" + ); + assert_eq!( + discarded_ids(&events), + vec![steer_id], + "the refused entry must be announced as never-to-deliver, got {events:?}" + ); + assert!( + delivered_ids(&events).is_empty(), + "a steer no store ever recorded must not be reported delivered, got {events:?}" + ); + assert!( + fixture.conversations.messages.read().await.is_empty(), + "the failed write must not have recorded a row" + ); + assert_eq!( + fixture.conversations.add_message_attempts(), + 1, + "the sink must give up after the one failed write, not retry" + ); +} + +/// An ordinary delivery persists the text and announces the entry it popped. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-003 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_delivery_persists_the_text_and_announces_the_popped_entry() { + let _steering_bus_guard = lock_steering_bus().await; + let fixture = DeliveryFixture::new(); + let steer_id = Uuid::new_v4(); + queue_entry(&fixture, steer_id, "focus on the parser"); + let mut sink = sink_for(&fixture); + + let ((), events) = events_during(fixture.conversation_id, async { + sink.delivered("focus on the parser") + .await + .expect("the delivery must succeed"); + }) + .await; + + assert_eq!( + delivered_ids(&events), + vec![steer_id], + "the delivery must announce the queued entry's id, got {events:?}" + ); + assert_eq!( + persisted_shape(&fixture.conversations).await, + vec![(MessageRole::User, "focus on the parser".to_string())], + "the delivered text must be persisted as a user row" + ); +} + +/// A delivery whose queue entry is already gone (teardown raced it) still +/// persists the text, records it, and stays silent: the discard already fired. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-003 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn a_delivery_with_no_queue_entry_persists_without_a_terminal_event() { + let _steering_bus_guard = lock_steering_bus().await; + let fixture = DeliveryFixture::new(); + let mut sink = sink_for(&fixture); + + let ((), events) = events_during(fixture.conversation_id, async { + sink.delivered("landed during teardown") + .await + .expect("an empty queue is not an error; the row is still persisted"); + }) + .await; + + assert!( + events.is_empty(), + "the teardown race already announced the discard, so the sink must emit nothing, \ + got {events:?}" + ); + assert_eq!( + persisted_shape(&fixture.conversations).await, + vec![(MessageRole::User, "landed during teardown".to_string())], + "the text must be persisted even though its queue entry is gone" + ); + assert_eq!( + sink.delivered, 1, + "the delivery must be recorded for the turn that delivered it" + ); +} + +/// Correlation must survive a delivery whose text matches a non-head entry: +/// the head is resolved whatever the delivered text says. +/// +/// The mutation experiment showed the duplicate-text test alone passes under +/// a text-matcher that removes the entry it matched, because a FIFO fork +/// happens to feed texts back in queue order. This test removes that luck: +/// a text naming the second entry must still resolve the first, which is +/// what "never match by text" means as an implementation rule. The fork's +/// FIFO discipline means production never feeds texts out of order; this +/// pin exists so that a future correlation change fails here first. +/// +/// @plan PLAN-20260905-STEERINT.P04 +/// @requirement REQ-SI-005 +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn correlation_is_positional_even_when_the_text_matches_a_later_entry() { + let _steering_bus_guard = lock_steering_bus().await; + let fixture = DeliveryFixture::new(); + let head = Uuid::new_v4(); + let later = Uuid::new_v4(); + queue_entry(&fixture, head, "first text"); + queue_entry(&fixture, later, "second text"); + let mut sink = sink_for(&fixture); + + let ((), events) = events_during(fixture.conversation_id, async { + sink.delivered("second text") + .await + .expect("the delivery must succeed"); + }) + .await; + + assert_eq!( + delivered_ids(&events), + vec![head], + "the delivery must resolve the queue head, never the entry its text matches, \ + got {events:?}" + ); + let remaining = fixture.service.drain_steering(fixture.conversation_id); + assert_eq!( + remaining.iter().map(|entry| entry.id).collect::>(), + vec![later], + "the entry the text matched must still be queued, waiting for its own delivery" + ); +} diff --git a/src/services/chat_impl/tests/stream_failure/finalization.rs b/src/services/chat_impl/tests/stream_failure/finalization.rs index c026ce5..08bbf39 100644 --- a/src/services/chat_impl/tests/stream_failure/finalization.rs +++ b/src/services/chat_impl/tests/stream_failure/finalization.rs @@ -12,6 +12,7 @@ fn active_stream_entry(stream_id: Uuid) -> ActiveStream { task: None, cancel: CancellationToken::new(), state: StreamLifecycle::Running, + steering: serdes_ai_agent::SteeringQueue::new(), } } diff --git a/tests/e2e_agent_tool_execution.rs b/tests/e2e_agent_tool_execution.rs index 093316f..7467c2d 100644 --- a/tests/e2e_agent_tool_execution.rs +++ b/tests/e2e_agent_tool_execution.rs @@ -59,6 +59,7 @@ async fn test_agent_mode_basic() { &agent, &messages, McpToolContext::default(), + None, |event| match &event { StreamEvent::TextDelta(text) => { print!("{text}"); @@ -153,6 +154,7 @@ async fn test_agent_tool_events() { &agent, &messages, McpToolContext::default(), + None, |event| match &event { StreamEvent::TextDelta(text) => { print!("{text}"); diff --git a/tests/e2e_codex_stream.rs b/tests/e2e_codex_stream.rs index 4585346..ff448f5 100644 --- a/tests/e2e_codex_stream.rs +++ b/tests/e2e_codex_stream.rs @@ -276,6 +276,7 @@ async fn a_live_turn_through_the_agent_streams_text() { &agent, &[Message::user("Reply with exactly: agent ok")], McpToolContext::default(), + None, move |event| { sink.lock().expect("event sink").push(event); }, diff --git a/tests/e2e_install_and_search.rs b/tests/e2e_install_and_search.rs index 07534fa..d9e1fef 100644 --- a/tests/e2e_install_and_search.rs +++ b/tests/e2e_install_and_search.rs @@ -152,6 +152,7 @@ async fn test_install_exa_and_search() { &agent, &messages, McpToolContext::default(), + None, |event| match &event { StreamEvent::TextDelta(text) => { print!("{text}"); diff --git a/tests/e2e_real_tool_execution.rs b/tests/e2e_real_tool_execution.rs index 2b3830d..09a2b26 100644 --- a/tests/e2e_real_tool_execution.rs +++ b/tests/e2e_real_tool_execution.rs @@ -104,11 +104,17 @@ async fn test_real_exa_search() { )]; client - .run_agent_stream(&agent, &messages, McpToolContext::default(), |event| { - if let StreamEvent::TextDelta(text) = event { - print!("{text}"); - } - }) + .run_agent_stream( + &agent, + &messages, + McpToolContext::default(), + None, + |event| { + if let StreamEvent::TextDelta(text) = event { + print!("{text}"); + } + }, + ) .await .expect("Agent stream failed"); @@ -154,6 +160,7 @@ async fn test_real_exa_search() { &agent, &messages, McpToolContext::default(), + None, |event| match &event { StreamEvent::TextDelta(text) => { print!("{text}"); diff --git a/tests/kimi_agent_path_tests.rs b/tests/kimi_agent_path_tests.rs index 2c18b30..88124f7 100644 --- a/tests/kimi_agent_path_tests.rs +++ b/tests/kimi_agent_path_tests.rs @@ -99,6 +99,7 @@ async fn agent_path_sends_user_agent_header_to_kimi() { &agent, &[LlmMessage::user("ping")], McpToolContext::default(), + None, |event| match event { StreamEvent::TextDelta(text) => { collected_text.push_str(&text); @@ -208,6 +209,7 @@ async fn agent_path_without_kimi_provider_misses_user_agent() { &agent, &[LlmMessage::user("hi")], McpToolContext::default(), + None, |event| { if let StreamEvent::Error(_) = event { saw_error = true; @@ -309,6 +311,7 @@ data: [DONE] &agent, &[LlmMessage::user("hi")], McpToolContext::default(), + None, |event| match event { StreamEvent::TextDelta(text) => collected_text.push_str(&text), StreamEvent::Complete { .. } => saw_complete = true, @@ -376,6 +379,7 @@ async fn agent_path_kimi_live_e2e() { &agent, &[LlmMessage::user("Say exactly: pong")], McpToolContext::default(), + None, |event| match event { StreamEvent::TextDelta(text) => { print!("{text}"); diff --git a/tests/open_responses_wire_tests.rs b/tests/open_responses_wire_tests.rs index a9b229b..3504c89 100644 --- a/tests/open_responses_wire_tests.rs +++ b/tests/open_responses_wire_tests.rs @@ -544,6 +544,7 @@ async fn the_agent_path_also_asks_for_reasoning() { &agent, &[user("think about it")], McpToolContext::default(), + None, |_event| {}, ) .await;