From bca585e904548aca5bf7775da79e3b5b47f1c450 Mon Sep 17 00:00:00 2001 From: ymote <151983+ymote@users.noreply.github.com> Date: Mon, 14 Sep 2026 17:57:07 -0700 Subject: [PATCH] octos-app: a turn completes on the kernel's durable text, not on what streamed With the kernel pointed at DeepSeek V4 Flash (a thinking model, ~200 deltas/s) the hosted AppCard drew its weather card as a code block, cut off mid-line, while the same flow rendered with a slow local Qwen. Measured on the OnePlus (session 6aa89116-00000001): the kernel's ledger holds all 1840 text deltas (7194 chars, a clean ```runl0 block) but the app finalized on 897 of them (3251 chars, unclosed fence -> "still streaming" -> shown as code), and it never saw the durable assistant row before turn/completed. Reproduced on the desktop against a mock replaying that exact stream: "stream/persisted MISMATCH streamed=1937 persisted=7194". Why. `message/delta` is ephemeral: octos drops it when its stdio writer queue (1024 frames) is full (`WsConnection::try_enqueue` -> BackpressureDrop, debug-level only). The durable `assistant_persisted` row rides the ledger forwarder, which lags under the same backpressure, while the legacy `turn/completed` is direct-sent and overtakes it - the kernel's own comment in `send_notification_lifecycle` names this ordering as "the real first-turn soak failure" and fixes it only for v2-negotiated clients. The reasoning text itself never reached the message: `reasoning_delta` was already bridged to the thinking strip. Thinking merely made the model fast. Fix, backend/octos_ui.rs: a successful terminal (legacy or v2) finalizes only once the turn's durable text is in hand. Otherwise the turn is held, `session/hydrate` is requested, and the turn completes on whichever lands first - the live `assistant_persisted` row or the hydrate reply's assistant row for that turn - bridged as `TextAuthoritative` right before `TurnComplete`, which lib.rs already prefers over the streamed accumulation. A hold expires after 8 s onto the streamed text (an empty reply, a kernel without the v2 lane), so nothing hangs. The hydrate replies a hold requested are consumed by it even when the live row resolved the hold first; otherwise they would replay the session over the chat. The persisted row is routed by the envelope's turn id, falling back to the session as before. Terminal metrics are recorded once per turn. Also, app/l0_card.rs: `split_l0_blocks` treated the first "```runl0" substring as the opener and the next "```" substring as the close. Fences are lines now: the opener is a line that is ```runl0 (Markdown's three leading spaces allowed), the close is the next line that is exactly ```. DeepSeek's visible reasoning names the fence in prose and drafts the card inside a bare ``` block; substring scanning would have made a ledger out of the model's sentence had that text ever reached the message. Tests: the held-terminal paths (live row, hydrate reply, expiry, swallowed late reply, ordered lane) and the fence scanner against the recorded reasoning-plus-card shape. Verified on the desktop against the recorded stream and on the phone against DeepSeek. Co-Authored-By: Claude Fable 5.1 --- app/app/src/app/l0_card.rs | 169 ++++++++++-- app/app/src/backend/octos_ui.rs | 459 ++++++++++++++++++++++++++++---- 2 files changed, 564 insertions(+), 64 deletions(-) diff --git a/app/app/src/app/l0_card.rs b/app/app/src/app/l0_card.rs index 288f4781..221c00c5 100644 --- a/app/app/src/app/l0_card.rs +++ b/app/app/src/app/l0_card.rs @@ -1565,33 +1565,81 @@ pub enum Piece<'a> { /// end of the pipeline downcast the VM host to one — and a unit test has no /// `Cx` to give. This half is this module's own logic and is testable; the other /// half is covered by `octoscript-ui-l0`'s lowering tests and by the device harness. +/// +/// FENCES ARE LINES. An opener is a line that is ```` ```runl0 ```` (Markdown's +/// up-to-three leading spaces allowed, nothing else on the line); the close is +/// the next line that is exactly ```` ``` ````. This scanned for the SUBSTRINGS +/// instead — the first "```runl0" anywhere and the next "```" anywhere — and a +/// thinking model showed why that is wrong: DeepSeek V4 Flash's visible +/// reasoning says, in prose, "I must emit exactly one ```runl0 fenced block", +/// and then drafts the card inside a bare ```` ``` ```` block before writing the +/// real one. Substring scanning takes the prose mention as the opener and the +/// draft's fence as its close, and what renders is a "ledger" made of the +/// model's sentence. Reasoning is kept off the message text upstream (the +/// thinking strip), and this scanner no longer depends on that. pub fn split_l0_blocks(text: &str) -> Vec> { - const FENCE: &str = "```runl0"; let mut out = Vec::new(); - let mut rest = text; - while let Some(open) = rest.find(FENCE) { - out.push(Piece::Prose(&rest[..open])); - let after = &rest[open + FENCE.len()..]; - let body_start = after.find('\n').map(|n| n + 1).unwrap_or(after.len()); - let body_and_rest = &after[body_start..]; - match body_and_rest.find("```") { - Some(close) => { - out.push(Piece::Ledger(&body_and_rest[..close])); - rest = &body_and_rest[close + 3..]; + let mut prose_from = 0; + let mut pos = 0; + while pos < text.len() { + let line_end = text[pos..].find('\n').map(|n| pos + n + 1).unwrap_or(text.len()); + let line = &text[pos..line_end]; + if !is_opening_fence(line) { + pos = line_end; + continue; + } + let body_start = line_end; + // The close: the next line that is only ```. + let mut scan = body_start; + let mut close: Option<(usize, usize)> = None; + while scan < text.len() { + let end = text[scan..].find('\n').map(|n| scan + n + 1).unwrap_or(text.len()); + if let Some(after_ticks) = closing_fence_end(&text[scan..end]) { + close = Some((scan, scan + after_ticks)); + break; + } + scan = end; + } + match close { + Some((close_line, after_ticks)) => { + out.push(Piece::Prose(&text[prose_from..pos])); + out.push(Piece::Ledger(&text[body_start..close_line])); + // Prose resumes right after the three backticks, as before: + // the rest of that line and everything below is the caller's. + prose_from = after_ticks; + pos = after_ticks; } // An unclosed block is still streaming. Leave it whole rather than // rendering half a ledger, which looks like a card the model got // wrong rather than one that has not finished arriving. None => { - out.push(Piece::Prose(&rest[open..])); + out.push(Piece::Prose(&text[prose_from..])); return out; } } } - out.push(Piece::Prose(rest)); + out.push(Piece::Prose(&text[prose_from..])); out } +/// A line (with or without its `\n`) that opens an L0 ledger: ```` ```runl0 ```` +/// alone, after at most three spaces of indentation. +fn is_opening_fence(line: &str) -> bool { + let body = line.trim_end_matches(['\n', '\r']); + let indent = body.len() - body.trim_start_matches(' ').len(); + indent <= 3 && body[indent..].trim_end() == "```runl0" +} + +/// A line (with or without its `\n`) that closes a ledger: ```` ``` ```` alone, +/// whitespace around it allowed. The offset just past the backticks, so the +/// caller can resume prose there. +fn closing_fence_end(line: &str) -> Option { + let body = line.trim_end_matches(['\n', '\r']); + let indent = body.len() - body.trim_start().len(); + let rest = &body[indent..]; + (rest.trim_end() == "```").then_some(indent + 3) +} + /// What each source's fetch is actually doing, keyed by the ledger it belongs to. /// /// Observed, not asserted. See `live_source_status`. @@ -2094,12 +2142,97 @@ mod resolve_tests { /// Two ledgers in one message are both found. #[test] fn two_ledgers_are_both_found() { - let msg = "```runl0\na\n```mid```runl0\nb\n```"; - let n = split_l0_blocks(msg) + let msg = "```runl0\na\n```\nmid\n```runl0\nb\n```"; + let pieces = split_l0_blocks(msg); + let ledgers: Vec<&str> = pieces .iter() - .filter(|p| matches!(p, Piece::Ledger(_))) - .count(); - assert_eq!(n, 2, "both ledgers must be found"); + .filter_map(|p| match p { + Piece::Ledger(t) => Some(*t), + _ => None, + }) + .collect(); + assert_eq!(ledgers, vec!["a\n", "b\n"], "both ledgers must be found: {pieces:?}"); + assert_eq!( + pieces.iter().filter(|p| matches!(p, Piece::Prose(_))).count(), + 3, + "prose before, between and after: {pieces:?}" + ); + } + + /// Fences are lines. A "```runl0" inside a sentence opens nothing, and a + /// bare ``` block is prose, not a ledger. + #[test] + fn a_fence_mentioned_in_prose_opens_nothing() { + let msg = "I must emit exactly one ```runl0 fenced block as the entire answer.\n\n\ + ```\n# level: L0\ndraft\n```\n"; + let pieces = split_l0_blocks(msg); + assert!( + !pieces.iter().any(|p| matches!(p, Piece::Ledger(_))), + "prose that names the fence is not a fence: {pieces:?}" + ); + assert_eq!(rejoin(&pieces), msg); + } + + /// A close is a line that is exactly ```; a longer run of backticks or a + /// ``` with trailing text on its line does not end the ledger. + #[test] + fn a_close_is_a_line_of_three_backticks() { + let msg = "```runl0\nview root Col { Text(\"```not a close\") }\n```\n"; + assert_eq!( + split_l0_blocks(msg), + vec![ + Piece::Prose(""), + Piece::Ledger("view root Col { Text(\"```not a close\") }\n"), + Piece::Prose("\n"), + ] + ); + // Whitespace around either fence is tolerated (Markdown's three spaces). + let indented = " ```runl0 \nview root Rule()\n ``` \n"; + let pieces = split_l0_blocks(indented); + assert!(matches!(pieces.as_slice(), [Piece::Prose(""), Piece::Ledger("view root Rule()\n"), Piece::Prose(" \n")]), "{pieces:?}"); + } + + /// The exact shape a thinking model produced (DeepSeek V4 Flash, "tokyo + /// weather", ledger 6aa88d76): 98 lines of visible reasoning that (a) name + /// the fence in prose and (b) draft the card inside a bare ``` block, + /// followed by the real answer — one ```runl0 block. If the reasoning ever + /// reaches the message text, exactly the answer must be the ledger. + #[test] + fn reasoning_before_the_card_yields_exactly_the_card() { + let reasoning = "The user request is \"tokyo weather\" — a bare place name + weather. \ + Pick the weather app.\n\n\ + I must emit exactly one ```runl0 fenced block as the entire answer. No prose.\n\n\ + Let me carefully transcribe.\n\n\ + ```\n# level: L0\n# model: weather\ntheme atro_light\n\n\ + source place sys.geocode(name: state.city)\n\ + copy feels ... etc.\n```\n\n\ + Wait — catalog says `sys.moonphase` answers `phase`. Fix that.\n\n"; + let card = "# level: L0\n# model: weather\ntheme atro_light\n\n\ + source place sys.geocode(name: state.city)\n\ + state city { shape: text, initial: \"Tokyo\" }\n\n\ + view root Surface(pad: .page) {\n\ + \x20 when now.$state == .pending { TextBody(text: copy.loading) }\n\ + \x20 when now.$state == .failed { TextBody(text: copy.offline) }\n\ + }\n"; + let msg = format!("{reasoning}```runl0\n{card}```"); + let pieces = split_l0_blocks(&msg); + let ledgers: Vec<&str> = pieces + .iter() + .filter_map(|p| match p { + Piece::Ledger(t) => Some(*t), + _ => None, + }) + .collect(); + assert_eq!(ledgers, vec![card], "the answer, whole, and nothing from the reasoning: {pieces:?}"); + assert!( + matches!(pieces.as_slice(), [Piece::Prose(p), Piece::Ledger(_), Piece::Prose("")] if *p == reasoning), + "the reasoning is prose, untouched, and nothing follows the card: {pieces:?}" + ); + // And the card alone — the message as the thinking strip keeps it. + assert_eq!( + split_l0_blocks(&format!("```runl0\n{card}```")), + vec![Piece::Prose(""), Piece::Ledger(card), Piece::Prose("")] + ); } } diff --git a/app/app/src/backend/octos_ui.rs b/app/app/src/backend/octos_ui.rs index 201bd23f..28e3464b 100644 --- a/app/app/src/backend/octos_ui.rs +++ b/app/app/src/backend/octos_ui.rs @@ -9,7 +9,7 @@ //! Crate boundary: `OctosUiAgent` is the *only* place inside `app/` that //! talks to `octos-app-transport`. UI code goes through the `Agent` trait. -use std::collections::HashMap; +use std::collections::{HashMap, HashSet}; use makepad_ai::{Agent, AgentEvent, PromptId, SessionConfig, SessionId, StopReason}; use makepad_widgets::*; @@ -101,8 +101,49 @@ pub struct OctosUiAgent { /// rather than WebSocket. Stdio carries no `X-Profile-Id` header, so /// `session/open` must name the profile in its params instead. stdio_transport: bool, + /// Turns whose durable `assistant_persisted` row has already been bridged + /// as `TextAuthoritative`. A terminal for one of these can finalize at + /// once; a terminal for any other turn must wait (see `finish_turn`). + persisted_seen: HashSet, + /// Turns that have reached their terminal but whose authoritative text + /// has not arrived. Each waits for the durable row on the live lane, or + /// for the `session/hydrate` reply requested on its behalf, or for + /// [`AUTHORITATIVE_WAIT`] to expire — whichever comes first. + pending_completion: HashMap, + /// `session/hydrate` requests issued on behalf of held turns, per session, + /// not yet answered. Their replies are the hold's business even when the + /// live row has resolved the hold first — a reply that fell through to + /// the resume flow would replace the chat with the session's history + /// mid-turn. + hold_hydrates: HashMap, } +/// A turn held open at its terminal until the kernel's durable assistant row +/// is in hand — or the wait expires. +/// +/// WHY THE WAIT EXISTS. `message/delta` is ephemeral: the kernel drops it on +/// the floor when its stdio writer backs up (a 1024-frame queue; a fast model +/// such as DeepSeek V4 Flash at ~200 deltas/s fills it on a phone), and the +/// live ledger forwarder that carries the durable `assistant_persisted` row +/// lags behind the DIRECT-sent legacy `turn/completed`. Measured on the +/// OnePlus: 897 of 1840 text deltas arrived, the persisted row had not, and +/// the terminal came anyway — so the turn finalized on 3251 of 7194 chars, +/// an unclosed ```runl0 fence, and the card drew as a code block. The kernel +/// keeps the whole reply in its ledger the entire time; `session/hydrate` +/// reads it back. Holding the terminal until that text is here is what makes +/// the card the model wrote the card that renders. +#[derive(Debug)] +struct PendingCompletion { + session_id: SessionKey, + since: std::time::Instant, +} + +/// Upper bound on how long a completed turn waits for its durable text. +/// A kernel that never persists a row (an empty reply; a build without the +/// v2 lane) must not leave the spinner running forever, so after this the +/// turn completes on the streamed text with a warning in the log. +const AUTHORITATIVE_WAIT: std::time::Duration = std::time::Duration::from_secs(8); + impl OctosUiAgent { /// Construct a new agent. Spawns the WebSocket task immediately so /// `create_session` can ship `session/open` on the first call. @@ -165,6 +206,9 @@ impl OctosUiAgent { fallback_profile, thinking: false, stdio_transport, + persisted_seen: HashSet::new(), + pending_completion: HashMap::new(), + hold_hydrates: HashMap::new(), } } @@ -232,6 +276,165 @@ impl OctosUiAgent { } } + /// Forget every routing entry for `turn`. The prompt it mapped to, if + /// it was tracked. + fn retire_turn(&mut self, turn: &TurnId) -> Option { + let pid = self.prompt_ids.remove(turn)?; + self.turn_ids.remove(&pid); + self.prompt_sessions.remove(&pid); + self.generation_started.remove(turn); + self.persisted_seen.remove(turn); + self.pending_completion.remove(turn); + Some(pid) + } + + /// A turn reached a successful terminal on `lane`. It completes now if + /// its durable assistant row has already been bridged; otherwise it is + /// held and the row is fetched with `session/hydrate`, and the turn + /// completes when either that reply or the live row lands — see + /// [`PendingCompletion`] for why a terminal alone is not enough. + fn finish_turn(&mut self, turn: &TurnId, lane: &str) -> Vec { + let Some(&pid) = self.prompt_ids.get(turn) else { + return Vec::new(); + }; + if self.persisted_seen.contains(turn) { + self.retire_turn(turn); + return vec![AgentEvent::TurnComplete { + prompt_id: pid, + stop_reason: StopReason::EndTurn, + }]; + } + if self.pending_completion.contains_key(turn) { + // The other lane's terminal for a turn already held. + return Vec::new(); + } + let Some(session_id) = self.prompt_sessions.get(&pid).cloned() else { + self.retire_turn(turn); + return vec![AgentEvent::TurnComplete { + prompt_id: pid, + stop_reason: StopReason::EndTurn, + }]; + }; + log::warn!( + "octos-ui-agent: {lane} for turn {} arrived before its persisted row; \ + holding completion and hydrating {}", + turn.0, + session_id.0 + ); + self.pending_completion.insert( + turn.clone(), + PendingCompletion { + session_id: session_id.clone(), + since: std::time::Instant::now(), + }, + ); + *self.hold_hydrates.entry(session_id.clone()).or_insert(0) += 1; + self.post(OutboundCommand::HydrateSession { + session_id: session_id.0, + }); + Vec::new() + } + + /// The durable text for `turn` is in hand. Bridged as authoritative; a + /// turn held at its terminal completes right behind it. + fn authoritative_arrived(&mut self, turn: &TurnId, text: String) -> Vec { + let Some(&pid) = self.prompt_ids.get(turn) else { + return Vec::new(); + }; + self.persisted_seen.insert(turn.clone()); + let mut out = vec![AgentEvent::TextAuthoritative { + prompt_id: pid, + text, + }]; + if self.pending_completion.contains_key(turn) { + self.retire_turn(turn); + out.push(AgentEvent::TurnComplete { + prompt_id: pid, + stop_reason: StopReason::EndTurn, + }); + } + out + } + + /// The `session/hydrate` reply requested for a held `turn`. Its + /// assistant row for that turn — or, failing a turn-tagged row, the + /// newest assistant row — is the text the turn completes on. A reply + /// with no assistant text (an empty reply; a decode failure) completes + /// the turn on what streamed, so it never hangs on a bad reply. + fn complete_from_hydrate(&mut self, turn: &TurnId, result: serde_json::Value) -> Vec { + let text = match serde_json::from_value::(result) { + Ok(r) => { + let rows = r.messages.unwrap_or_default(); + let tagged = rows + .iter() + .rev() + .find(|row| row.role == "assistant" && row.turn_id.as_ref() == Some(turn)); + let row = tagged.or_else(|| { + log::warn!( + "octos-ui-agent: hydrate carries no assistant row tagged for turn {}; \ + taking the newest", + turn.0 + ); + rows.iter().rev().find(|row| row.role == "assistant") + }); + row.map(|row| row.content.clone()).filter(|text| !text.trim().is_empty()) + } + Err(e) => { + log::warn!("octos-ui-agent: decode session/hydrate for held turn: {e}"); + None + } + }; + match text { + Some(text) => { + log::info!( + "octos-ui-agent: held turn {} completes on its hydrated row ({} chars)", + turn.0, + text.chars().count() + ); + self.authoritative_arrived(turn, text) + } + None => self + .retire_turn(turn) + .map(|pid| { + log::warn!( + "octos-ui-agent: no assistant text hydrated for turn {}; completing on the streamed text", + turn.0 + ); + vec![AgentEvent::TurnComplete { + prompt_id: pid, + stop_reason: StopReason::EndTurn, + }] + }) + .unwrap_or_default(), + } + } + + /// Complete every held turn whose [`AUTHORITATIVE_WAIT`] has run out, on + /// whatever text streamed. + fn expire_pending_completions(&mut self) -> Vec { + let expired: Vec = self + .pending_completion + .iter() + .filter(|(_, p)| p.since.elapsed() >= AUTHORITATIVE_WAIT) + .map(|(turn, _)| turn.clone()) + .collect(); + let mut out = Vec::new(); + for turn in expired { + if let Some(pid) = self.retire_turn(&turn) { + log::warn!( + "octos-ui-agent: no persisted row for turn {} within {:?}; completing on the streamed text", + turn.0, + AUTHORITATIVE_WAIT + ); + out.push(AgentEvent::TurnComplete { + prompt_id: pid, + stop_reason: StopReason::EndTurn, + }); + } + } + out + } + /// W05 — extract a cheap, send-safe handle for issuing /// `approval/respond` commands without cloning the whole agent. The /// handle wraps a `Sender` plus a runtime handle, so @@ -342,9 +545,29 @@ impl OctosUiAgent { Vec::new() } TransportEvent::SessionHydrated { session_id, result } => { + let key = SessionKey(session_id); + // A turn held at its terminal asked for this reply: the + // durable assistant row it carries is the text the turn + // finalizes on. Consumed here — it must not replay the whole + // session over a chat that is mid-turn — and consumed even + // when the live row already resolved the hold. + if let Some(outstanding) = self.hold_hydrates.get_mut(&key) { + *outstanding -= 1; + if *outstanding == 0 { + self.hold_hydrates.remove(&key); + } + let held = self + .pending_completion + .iter() + .find(|(_, p)| p.session_id == key) + .map(|(turn, _)| turn.clone()); + return match held { + Some(turn) => self.complete_from_hydrate(&turn, result), + None => Vec::new(), + }; + } // Resume flow: decode the chat rows and hand them to the App // via a posted action (same pattern as `SessionListAction`). - let key = SessionKey(session_id); let Some(&sid) = self.session_ids.get(&key) else { log::warn!( "octos-ui-agent: session/hydrate reply for unmapped {}", @@ -469,22 +692,25 @@ impl OctosUiAgent { let turn = self.prompt_ids.keys() .find(|turn| turn.0.to_string() == ev.envelope.turn_id).cloned(); if let Some(turn) = turn { - let pid = self.prompt_ids.remove(&turn).unwrap(); - self.turn_ids.remove(&pid); - self.prompt_sessions.remove(&pid); - let elapsed = self.generation_started.remove(&turn) - .map(|(started, _)| started.elapsed().as_millis()); let success = matches!(outcome, octos_core::ui_protocol::TurnTerminalOutcome::Completed); - self.app_prompt_cache.complete(&turn, success, - token_usage.as_ref().map(|usage| usage.output_tokens as usize)); - log::info!("generation-metric {}", serde_json::json!({ - "event": if success { "completed" } else { "failed" }, - "turn_id": turn, "elapsed_ms": elapsed, - "outcome": outcome, "token_usage": token_usage - })); + // The legacy `turn/completed` may already have reached + // this turn and be holding it for its durable text; its + // metrics were recorded then. Record them once. + if !self.pending_completion.contains_key(&turn) { + let elapsed = self.generation_started.remove(&turn) + .map(|(started, _)| started.elapsed().as_millis()); + self.app_prompt_cache.complete(&turn, success, + token_usage.as_ref().map(|usage| usage.output_tokens as usize)); + log::info!("generation-metric {}", serde_json::json!({ + "event": if success { "completed" } else { "failed" }, + "turn_id": turn, "elapsed_ms": elapsed, + "outcome": outcome, "token_usage": token_usage + })); + } return if success { - vec![AgentEvent::TurnComplete { prompt_id: pid, stop_reason: StopReason::EndTurn }] + self.finish_turn(&turn, "v2 terminal") } else { + let pid = self.retire_turn(&turn).expect("turn was tracked above"); vec![AgentEvent::PromptError { prompt_id: pid, error: error.as_ref().map(|e| e.message.clone()) .unwrap_or_else(|| format!("Generation ended: {outcome:?}")) }] @@ -537,7 +763,9 @@ impl OctosUiAgent { } } } - UiNotification::TurnCompleted(ev) => { + // A turn the v2 terminal already reached (and is holding for its + // durable text) has had its metrics recorded; do not record twice. + UiNotification::TurnCompleted(ev) if !self.pending_completion.contains_key(&ev.turn_id) => { self.app_prompt_cache.complete(&ev.turn_id, true, ev.tokens_out.map(|tokens| tokens as usize)); if let Some((started, _)) = self.generation_started.remove(&ev.turn_id) { log::info!("generation-metric {}", serde_json::json!({ @@ -621,42 +849,36 @@ impl OctosUiAgent { // bridging them here would double every one. UiNotification::EnvelopeV2(ev) => match ev.envelope.payload { PayloadV2::AssistantPersisted { text, .. } if !text.trim().is_empty() => { - // Route by SESSION, not turn. The envelope does carry a - // `turn_id`, but as a bare String against our `TurnId(Uuid)` - // keys; `prompt_sessions` holds exactly the prompts still in - // flight, so the session's entry is the turn this row - // belongs to — the same routing the old lane used. - self.prompt_sessions - .iter() - .find(|(_, sess)| **sess == ev.session_id) - .map(|(prompt_id, _)| { - vec![AgentEvent::TextAuthoritative { - prompt_id: *prompt_id, - text, - }] - }) + // Route by the envelope's turn when it names one we track + // (the wire carries it as a bare String against our + // `TurnId(Uuid)` keys, so compare the rendering), else by + // SESSION: `prompt_sessions` holds exactly the prompts + // still in flight, so the session's entry is the turn this + // row belongs to — the routing the old lane used. + let turn = self + .prompt_ids + .keys() + .find(|turn| turn.0.to_string() == ev.envelope.turn_id) + .cloned() + .or_else(|| { + self.prompt_sessions + .iter() + .find(|(_, sess)| **sess == ev.session_id) + .and_then(|(pid, _)| self.turn_ids.get(pid).cloned()) + }); + turn.map(|turn| self.authoritative_arrived(&turn, text)) .unwrap_or_default() } _ => Vec::new(), }, - UiNotification::TurnCompleted(ev) => self - .prompt_ids - .remove(&ev.turn_id) - .map(|pid| { - self.turn_ids.remove(&pid); - self.prompt_sessions.remove(&pid); - vec![AgentEvent::TurnComplete { - prompt_id: pid, - stop_reason: StopReason::EndTurn, - }] - }) - .unwrap_or_default(), + // The legacy terminal is DIRECT-sent and can overtake the ledger + // lane carrying the durable row (the kernel says so itself, in + // `send_notification_lifecycle`); `finish_turn` holds it until + // that row — or its `session/hydrate` stand-in — has landed. + UiNotification::TurnCompleted(ev) => self.finish_turn(&ev.turn_id, "legacy turn/completed"), UiNotification::TurnError(ev) => self - .prompt_ids - .remove(&ev.turn_id) + .retire_turn(&ev.turn_id) .map(|pid| { - self.turn_ids.remove(&pid); - self.prompt_sessions.remove(&pid); vec![AgentEvent::PromptError { prompt_id: pid, error: format!("{}: {}", ev.code, ev.message), @@ -895,6 +1117,10 @@ impl Agent for OctosUiAgent { } } } + // A held terminal whose durable text never came completes on what + // streamed. Swept here because this runs on every UI event — the + // stream tick keeps them coming while a turn is open. + out.extend(self.expire_pending_completions()); out } @@ -1053,6 +1279,8 @@ mod generation_terminal_tests { app_prompt_cache: Default::default(), capabilities: None, workspace_cwd: None, fallback_profile: "_main".into(), thinking: false, stdio_transport: true, + persisted_seen: HashSet::new(), pending_completion: HashMap::new(), + hold_hydrates: HashMap::new(), } } @@ -1067,6 +1295,142 @@ mod generation_terminal_tests { }) } + /// The durable assistant row, as the ledger forwarder delivers it. + fn persisted(turn: &TurnId, text: &str) -> UiNotification { + UiNotification::EnvelopeV2(EnvelopeV2Notification { + session_id: SessionKey("_main:test".into()), topic: None, + envelope: EnvelopeV2 { + thread_id: "shared-thread".into(), seq: 2, cursor: None, + turn_id: turn.0.to_string(), client_message_id: None, + payload: PayloadV2::AssistantPersisted { + text: text.into(), + assistant_segment_id: format!("{}:assistant:iteration:1", turn.0), + meta: octos_core::ui_protocol::MessageMeta { + message_id: "row-1".into(), + persisted_at: chrono::Utc::now(), + media: Vec::new(), + }, + }, + }, + }) + } + + /// The direct-sent legacy terminal — the one that overtakes the ledger lane. + fn legacy_completed(turn: &TurnId) -> UiNotification { + UiNotification::TurnCompleted(TurnCompletedEvent { + session_id: SessionKey("_main:test".into()), topic: None, turn_id: turn.clone(), + cursor: None, tokens_in: None, tokens_out: None, session_result: None, + }) + } + + /// A `session/hydrate` reply carrying the turn's assistant row. + fn hydrated(turn: &TurnId, text: &str) -> TransportEvent { + TransportEvent::SessionHydrated { + session_id: "_main:test".into(), + result: serde_json::json!({ + "session_id": "_main:test", + "cursor": {"stream": "_main:test", "seq": 3}, + "messages": [ + {"seq": 1, "role": "user", "content": "tokyo weather", + "turn_id": turn, "persisted_at": "2026-09-14T09:28:19Z"}, + {"seq": 2, "role": "assistant", "content": text, + "turn_id": turn, "persisted_at": "2026-09-14T09:28:36Z"} + ] + }), + } + } + + const CARD: &str = "```runl0\n# level: L0\n# model: weather\nview root Col {}\n```"; + + /// The failure as measured: 897 of 1840 deltas, then the legacy + /// terminal, then (late) the persisted row. The terminal alone must not + /// finalize; the row does, as authoritative text. + #[test] + fn legacy_terminal_waits_for_the_persisted_row() { + let mut agent = agent(); + let (turn, prompt) = track(&mut agent); + assert!(agent.translate_notification(legacy_completed(&turn)).is_empty(), + "a terminal without the durable row must hold the turn"); + assert!(agent.prompt_ids.contains_key(&turn), "a held turn keeps routing late frames"); + assert!(agent.pending_completion.contains_key(&turn)); + let events = agent.translate_notification(persisted(&turn, CARD)); + assert!(matches!(events.as_slice(), + [AgentEvent::TextAuthoritative { prompt_id: p1, text }, AgentEvent::TurnComplete { prompt_id: p2, .. }] + if *p1 == prompt && *p2 == prompt && text == CARD), "{events:?}"); + assert!(agent.prompt_ids.is_empty() && agent.pending_completion.is_empty() + && agent.persisted_seen.is_empty()); + assert!(agent.translate_notification(terminal(&turn, TurnTerminalOutcome::Completed)).is_empty(), + "the ledger lane's own terminal, arriving after, does nothing"); + } + + /// When even the live row is lost (the ledger forwarder lagged), the + /// `session/hydrate` reply the hold requested completes the turn. + #[test] + fn held_turn_completes_on_the_hydrated_row() { + let mut agent = agent(); + let (turn, prompt) = track(&mut agent); + assert!(agent.translate_notification(legacy_completed(&turn)).is_empty()); + let events = agent.translate(hydrated(&turn, CARD)); + assert!(matches!(events.as_slice(), + [AgentEvent::TextAuthoritative { text, .. }, AgentEvent::TurnComplete { prompt_id, .. }] + if *prompt_id == prompt && text == CARD), "{events:?}"); + assert!(agent.pending_completion.is_empty()); + } + + /// The hydrate reply for a held turn is consumed by it and never replays + /// the session over the chat (no `SessionResumeHydrated`); a reply whose + /// assistant text is empty still completes the turn. + #[test] + fn empty_hydrated_row_completes_on_the_streamed_text() { + let mut agent = agent(); + let (turn, prompt) = track(&mut agent); + assert!(agent.translate_notification(legacy_completed(&turn)).is_empty()); + let events = agent.translate(hydrated(&turn, " ")); + assert!(matches!(events.as_slice(), [AgentEvent::TurnComplete { prompt_id, .. }] if *prompt_id == prompt), + "{events:?}"); + } + + /// The hydrate a hold requested still belongs to the hold when the live + /// row resolved it first: its reply is swallowed, never replayed over the + /// chat as a session resume. + #[test] + fn a_late_hold_hydrate_reply_is_swallowed() { + let mut agent = agent(); + let (turn, _) = track(&mut agent); + assert!(agent.translate_notification(legacy_completed(&turn)).is_empty()); + assert_eq!(agent.hold_hydrates.get(&SessionKey("_main:test".into())), Some(&1)); + assert!(matches!(agent.translate_notification(persisted(&turn, CARD)).as_slice(), + [AgentEvent::TextAuthoritative { .. }, AgentEvent::TurnComplete { .. }])); + assert!(agent.translate(hydrated(&turn, CARD)).is_empty()); + assert!(agent.hold_hydrates.is_empty()); + } + + /// The ordered lane: row, then terminal — completes at the terminal. + #[test] + fn persisted_row_first_completes_at_the_terminal() { + let mut agent = agent(); + let (turn, prompt) = track(&mut agent); + let events = agent.translate_notification(persisted(&turn, CARD)); + assert!(matches!(events.as_slice(), [AgentEvent::TextAuthoritative { .. }])); + let events = agent.translate_notification(legacy_completed(&turn)); + assert!(matches!(events.as_slice(), [AgentEvent::TurnComplete { prompt_id, .. }] if *prompt_id == prompt)); + assert!(agent.prompt_ids.is_empty()); + } + + /// A kernel that never persists a row must not hang the turn. + #[test] + fn held_turn_expires_onto_the_streamed_text() { + let mut agent = agent(); + let (turn, prompt) = track(&mut agent); + assert!(agent.translate_notification(legacy_completed(&turn)).is_empty()); + assert!(agent.expire_pending_completions().is_empty(), "not yet"); + agent.pending_completion.get_mut(&turn).unwrap().since = + std::time::Instant::now() - AUTHORITATIVE_WAIT; + let events = agent.expire_pending_completions(); + assert!(matches!(events.as_slice(), [AgentEvent::TurnComplete { prompt_id, .. }] if *prompt_id == prompt)); + assert!(agent.prompt_ids.is_empty() && agent.pending_completion.is_empty()); + } + #[test] fn l0_migration_first_reference_survives_startup_but_not_reconnect_or_resume() { let mut agent = agent(); @@ -1118,6 +1482,9 @@ mod generation_terminal_tests { let (turn, prompt) = track(&mut agent); assert!(agent.translate_notification(terminal(&TurnId::new(), TurnTerminalOutcome::Completed)).is_empty()); assert!(agent.prompt_ids.contains_key(&turn)); + // The ledger lane is ordered: the durable row precedes its terminal. + assert!(matches!(agent.translate_notification(persisted(&turn, CARD)).as_slice(), + [AgentEvent::TextAuthoritative { .. }])); let events = agent.translate_notification(terminal(&turn, TurnTerminalOutcome::Completed)); assert!(matches!(events.as_slice(), [AgentEvent::TurnComplete { prompt_id, .. }] if *prompt_id == prompt)); assert!(agent.prompt_ids.is_empty() && agent.turn_ids.is_empty()