diff --git a/PLAN.md b/PLAN.md index 06cf294..8f1cacb 100644 --- a/PLAN.md +++ b/PLAN.md @@ -113,6 +113,10 @@ seq N", so there is no separate history path to drift from the live one. - `UserMessage { text }` — echoed into the transcript **by the manager, not by drivers**, so every device renders the conversation from one stream. - `AssistantText { delta }` — streaming text, rendered as markdown. +- `AssistantTextFinal { text }` — the provider's authoritative value for the + open assistant message. It replaces its preceding provisional deltas while + remaining an append-only transcript event, so live SSE, replay and paging + converge on the same words. - `ToolStart / ToolUpdate / ToolEnd { tool, input, output }`. The tool vocabulary is common too (2026-09-09), not just the envelope: Codex's `/usr/bin/bash -lc` argv and Claude's Bash call are both @@ -247,13 +251,13 @@ remains the turn's usage. If an older common transcript has no such event yet, the server seeds the same measurement from the last `token_count` in Codex's own rollout, including when that rollout is on an SSH setup. -App-server assistant text comes only from its durable -`item/agentMessage/delta` notifications; the full text on `item/completed` is -always the consolidated copy and is ignored. This is decided from the dialect, -not an in-memory set of ids: after a backend restart, the previous deltas can -be behind the persisted stdout cursor while the completion is still ahead, -and forgetting which ids streamed used to append the complete message after -its already-recorded prefix. +App-server's `item/agentMessage/delta` notifications are provisional: safety +buffering can revise their words before `item/completed` supplies the durable +text. The driver records that completion as `AssistantTextFinal`; the phone +replaces the open message both live and on replay. A distinct append-only event +also makes adoption safe: if a backend restart falls between the deltas and the +completion, the correction is still meaningful without process-local memory +of which item ids streamed. ### The llama driver diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt index 203892e..d8415d5 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt @@ -61,6 +61,9 @@ sealed class SessionEvent { data class AssistantText(val delta: String) : SessionEvent() + /** The durable value of the open assistant message, replacing its provisional deltas. */ + data class AssistantTextFinal(val text: String) : SessionEvent() + data class ToolStart(val id: String, val tool: String, val input: String) : SessionEvent() data class ToolUpdate(val id: String, val output: String) : SessionEvent() @@ -225,6 +228,7 @@ fun parseSeqEvent(json: String): SeqEvent { ) "messageDropped" -> SessionEvent.MessageDropped(body.getString("id")) "assistantText" -> SessionEvent.AssistantText(body.getString("delta")) + "assistantTextFinal" -> SessionEvent.AssistantTextFinal(body.getString("text")) "toolStart" -> SessionEvent.ToolStart( id = body.getString("id"), diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt index 6e3896f..a387a6b 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -83,6 +83,8 @@ import androidx.lifecycle.repeatOnLifecycle import java.util.concurrent.atomic.AtomicLong import kotlinx.coroutines.Dispatchers import kotlinx.coroutines.awaitCancellation +import kotlinx.coroutines.cancelAndJoin +import kotlinx.coroutines.channels.Channel import kotlinx.coroutines.delay import kotlinx.coroutines.flow.drop import kotlinx.coroutines.flow.first @@ -95,6 +97,15 @@ import kotlinx.coroutines.withContext */ private val LOADING_SPINNER = 48.dp +/** One callback from the blocking SSE reader, queued in wire order for the UI thread. */ +private sealed interface TranscriptDelivery { + data object Opened : TranscriptDelivery + + data object Reset : TranscriptDelivery + + data class Event(val entry: SeqEvent) : TranscriptDelivery +} + /** * How close, in screenfuls of estimated scroll, the reader may come to the end of loaded history * before the next page is fetched. @@ -360,6 +371,10 @@ fun SessionScreen( // state rather than of item indices, because a zero-height first item makes an index ambiguous. // This is what the jump-to-newest button watches, and the gate on recording. val atNewest by remember { derivedStateOf { !listState.canScrollBackward } } + // A decision made only by an actual scroll gesture. `canScrollBackward` also changes while a + // streaming row is laid out, and treating that displacement as intent freezes the rest of the + // reply off-screen while still advancing the SSE cursor. + var followingNewest by remember(address, epoch) { mutableStateOf(savedAnchor == null) } // Transcript events that arrived while somebody was reading further back, in the order they // arrived, waiting for them to return to the newest end. See [record]. var held by remember { mutableStateOf(listOf()) } @@ -492,7 +507,10 @@ fun SessionScreen( } // In order, always: one late event recorded ahead of the backlog would fold a // streamed delta into whatever row happened to be last by then. - if (atNewest && held.isEmpty()) record(entry) else held = held + entry + // Read on the UI thread (the stream below marshals every frame here). This direct + // check closes the small window before the scroll observer records the gesture. + if (listState.isScrollInProgress && !atNewest) followingNewest = false + if (followingNewest && held.isEmpty()) record(entry) else held = held + entry } } } @@ -797,6 +815,7 @@ fun SessionScreen( } // Whatever happened above: an empty transcript is a state the screen can draw, and a // permanently blank one is not. + followingNewest = atNewest restoring = false ready = true // The opening page is sized for time-to-first-frame, not for reading: it fills a viewport @@ -852,25 +871,42 @@ fun SessionScreen( probePassed = true } } - withContext(Dispatchers.IO) { - source.follow( - after = lastSeq.get(), - // Connected, measured rather than inferred: this is what takes a - // failure off the screen. Clearing on the first event instead meant - // an idle session kept displaying an error it had recovered from. - onOpen = { streamError = null }, - onReset = { - // Too far behind to continue from: what is on screen is a stale - // prefix of a conversation that has moved on, and the window - // arriving next is not adjacent to it. Dropping the rows makes - // this the same as opening the screen. The cache needs no - // telling: the window's first seq is not the one it expected, - // which closes its live run and starts another. - dropLoadedTranscript() - }, - ) { entry -> - apply(entry) + // The socket is blocking, but Compose state belongs to this UI coroutine. + // Queue every callback -- including reset -- through one channel so none + // can race layout or overtake another while crossing threads. + val delivery = Channel(Channel.UNLIMITED) + val follower = + launch(Dispatchers.IO) { + try { + source.follow( + after = lastSeq.get(), + onOpen = { delivery.trySend(TranscriptDelivery.Opened) }, + onReset = { delivery.trySend(TranscriptDelivery.Reset) }, + ) { entry -> + delivery.trySend(TranscriptDelivery.Event(entry)) + } + delivery.close() + } catch (error: Throwable) { + delivery.close(error) + } } + try { + for (next in delivery) { + when (next) { + TranscriptDelivery.Opened -> streamError = null + TranscriptDelivery.Reset -> { + // The fresh window has no position in common with the rows + // just dropped, so it becomes the view even if the reader + // had been further back in the stale prefix. + dropLoadedTranscript() + followingNewest = true + } + is TranscriptDelivery.Event -> apply(next.entry) + } + } + } finally { + source.close() + follower.cancelAndJoin() } } catch (e: kotlinx.coroutines.CancellationException) { // Leaving the screen or going below STARTED. Not a failure, and swallowing @@ -920,9 +956,12 @@ fun SessionScreen( // paced out: they are at the bottom, which is the one place the list is allowed to follow new // content. LaunchedEffect(listState) { - snapshotFlow { atNewest && held.isNotEmpty() } - .collect { due -> - if (!due) return@collect + snapshotFlow { Triple(listState.isScrollInProgress, atNewest, held.isNotEmpty()) } + .collect { (scrolling, newest, hasHeld) -> + if (scrolling && !newest) followingNewest = false + if (!newest) return@collect + followingNewest = true + if (!hasHeld) return@collect val backlog = held held = listOf() backlog.forEach { record(it) } diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt index 27a78c9..c2b163b 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt @@ -63,6 +63,8 @@ sealed class TranscriptItem { * inside that reply would step the list under them. */ val settled: Boolean = false, + /** A final value that supersedes provisional deltas behind a page boundary. */ + val replacesPrefix: Boolean = false, ) : TranscriptItem() data class ToolRun( @@ -308,6 +310,7 @@ private fun healSplitMessage( // The rule between them is put in here too, since the fold that would have made it never saw // these two side by side. if (head.settled) return earlier to (listOf(TranscriptItem.TurnBreak(tail.seq)) + later) + if (tail.replacesPrefix) return earlier.dropLast(1) to later return earlier.dropLast(1) to (listOf(tail.copy(text = head.text + tail.text)) + later.drop(1)) } @@ -411,6 +414,24 @@ fun foldEvent(items: List, entry: SeqEvent): List { + val last = items.lastOrNull() + if (last is TranscriptItem.AssistantMsg && !last.settled) { + items.dropLast(1) + last.copy(text = event.text, replacesPrefix = true) + } else { + val between = + if (last is TranscriptItem.AssistantMsg) + listOf(TranscriptItem.TurnBreak(entry.seq)) + else emptyList() + items + + between + + TranscriptItem.AssistantMsg( + entry.seq, + event.text, + replacesPrefix = true, + ) + } + } is SessionEvent.ToolStart -> items + TranscriptItem.ToolRun( diff --git a/app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptItemsTest.kt b/app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptItemsTest.kt index 41afa53..e5d1e8b 100644 --- a/app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptItemsTest.kt +++ b/app/androidApp/src/test/kotlin/com/example/aiapp/TranscriptItemsTest.kt @@ -47,6 +47,50 @@ class TranscriptItemsTest { assertEquals(listOf("Still running its tests."), texts(items)) } + @Test + fun completed_text_replaces_provisional_deltas_live() { + val items = + fold( + SessionEvent.AssistantText("I'll inspect the color-c concrete implementation"), + SessionEvent.AssistantTextFinal( + "I’ll inspect the color-correction TODO and the relevant design." + ), + ) + assertEquals( + listOf("I’ll inspect the color-correction TODO and the relevant design."), + texts(items), + ) + } + + @Test + fun completed_text_discards_provisional_deltas_across_a_page_boundary() { + val earlier = fold(SessionEvent.AssistantText("1. provisional section\n\n")) + val later = + fold( + SessionEvent.AssistantTextFinal("1. final first section\n\n2. final second section") + ) + + assertEquals( + listOf("1. final first section\n\n2. final second section"), + texts(joinPages(earlier, later)), + ) + } + + @Test + fun a_final_value_after_a_settled_reply_is_a_new_message_across_a_page_boundary() { + val earlier = + fold( + SessionEvent.AssistantText("Previous answer."), + SessionEvent.Status("idle"), + ) + val later = fold(SessionEvent.AssistantTextFinal("Next answer.")) + + assertEquals( + listOf("Previous answer.", "Next answer."), + texts(joinPages(earlier, later)), + ) + } + /** * The rule that replaced the wall of reports. A turn that starts with nothing recorded in front * of it -- a subagent finishing, the CLI picking a conversation back up -- leaves two replies diff --git a/server/src/session/codex/translate.rs b/server/src/session/codex/translate.rs index 7f89f3a..3367873 100644 --- a/server/src/session/codex/translate.rs +++ b/server/src/session/codex/translate.rs @@ -52,12 +52,11 @@ impl Translator { Some("item.started") | Some("item/started") => start_item(&body["item"]), Some("item.updated") => update_item(&line["item"]), // The old `codex exec --json` dialect reports only the completed message. App-server - // reports every message through durable delta notifications and its completed copy - // must always be skipped. That rule cannot live in an in-memory set: after a backend - // restart the deltas are behind the persisted log cursor while the completion is not, - // which used to append the whole message again after its already-recorded prefix. + // also reports deltas, but they are provisional: safety buffering can revise their + // text before completion. Keep its completed copy as an append-only correction rather + // than guessing that the two representations concatenate to the same answer. Some("item.completed") => complete_item(&body["item"], true), - Some("item/completed") => complete_item(&body["item"], false), + Some("item/completed") => final_item(&body["item"]), Some("item/agentMessage/delta") => { let Some(delta) = body.get("delta").and_then(Value::as_str) else { return Vec::new(); @@ -221,6 +220,21 @@ fn complete_item(item: &Value, include_agent_message: bool) -> Vec { } } +fn final_item(item: &Value) -> Vec { + match item.get("type").and_then(Value::as_str) { + Some("agentMessage") => item + .get("text") + .and_then(Value::as_str) + .map(|text| { + vec![Event::AssistantTextFinal { + text: text.to_string(), + }] + }) + .unwrap_or_default(), + _ => complete_item(item, false), + } +} + fn tool(item: &Value) -> Option<(String, String, Value)> { let id = item.get("id")?.as_str()?.to_string(); let kind = item.get("type")?.as_str()?; @@ -633,7 +647,7 @@ mod tests { } #[test] - fn translates_native_app_server_streaming_without_repeating_the_final_item() { + fn native_app_server_completion_corrects_provisional_streaming() { let mut translator = Translator::default(); assert_eq!( translator.translate(&line( @@ -651,12 +665,13 @@ mod tests { delta: "hello".to_string() }] ); - assert!( - translator - .translate(&line( - r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"hello"}}}"# - )) - .is_empty() + assert_eq!( + translator.translate(&line( + r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"hello, revised"}}}"# + )), + vec![Event::AssistantTextFinal { + text: "hello, revised".to_string() + }] ); assert!( translator @@ -702,16 +717,17 @@ mod tests { } #[test] - fn an_app_server_completion_never_repeats_streamed_text_after_adoption() { + fn an_app_server_completion_corrects_streamed_text_after_adoption() { // A newly adopted translator has not seen the deltas already recorded by the previous // backend. The dialect, rather than process-local memory, decides that this is a copy. let mut adopted = Translator::default(); - assert!( - adopted - .translate(&line( - r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"the complete message"}}}"# - )) - .is_empty() + assert_eq!( + adopted.translate(&line( + r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"the complete message"}}}"# + )), + vec![Event::AssistantTextFinal { + text: "the complete message".to_string() + }] ); } } diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index 774c5c3..951c034 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -156,6 +156,16 @@ pub enum Event { AssistantText { delta: String, }, + /// The authoritative text of the assistant message whose deltas immediately + /// precede this event. + /// + /// Some providers stream a provisional rendering and revise it before the + /// item completes. This stays append-only like every other transcript + /// correction: readers replace the open message rather than editing an old + /// line, and replay therefore reaches the same text as the live stream. + AssistantTextFinal { + text: String, + }, ToolStart { id: String, tool: String, diff --git a/server/src/session/llama.rs b/server/src/session/llama.rs index e70e041..f3feb91 100644 --- a/server/src/session/llama.rs +++ b/server/src/session/llama.rs @@ -497,6 +497,9 @@ fn conversation(path: &Path) -> Vec { }); } Event::AssistantText { delta } => pending.push_str(&delta), + Event::AssistantTextFinal { text } => { + pending = text; + } _ => {} } } diff --git a/server/src/session/transcript.rs b/server/src/session/transcript.rs index 1357ef1..1b26d5c 100644 --- a/server/src/session/transcript.rs +++ b/server/src/session/transcript.rs @@ -834,6 +834,9 @@ mod tests { attachments: Vec::new(), }, text("hello"), + Event::AssistantTextFinal { + text: "hello, revised".into(), + }, Event::ToolStart { id: "t1".into(), tool: "bash".into(),