Fix Codex transcript convergence

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
+App-server's `item/agentMessage/delta` notifications are provisional: safety
-`item/agentMessage/delta` notifications; the full text on `item/completed` is
+buffering can revise their words before `item/completed` supplies the durable
-always the consolidated copy and is ignored. This is decided from the dialect,
+text. The driver records that completion as `AssistantTextFinal`; the phone
-not an in-memory set of ids: after a backend restart, the previous deltas can
+replaces the open message both live and on replay. A distinct append-only event
-be behind the persisted stdout cursor while the completion is still ahead,
+also makes adoption safe: if a backend restart falls between the deltas and the
-and forgetting which ids streamed used to append the complete message after
+completion, the correction is still meaningful without process-local memory
-its already-recorded prefix.
+of which item ids streamed.
 
 ### The llama driver
 

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"),

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<SeqEvent>()) }
@@ -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) {
+                        // The socket is blocking, but Compose state belongs to this UI coroutine.
-                            source.follow(
+                        // Queue every callback -- including reset -- through one channel so none
-                                after = lastSeq.get(),
+                        // can race layout or overtake another while crossing threads.
-                                // Connected, measured rather than inferred: this is what takes a
+                        val delivery = Channel<TranscriptDelivery>(Channel.UNLIMITED)
-                                // failure off the screen. Clearing on the first event instead meant
+                        val follower =
-                                // an idle session kept displaying an error it had recovered from.
+                            launch(Dispatchers.IO) {
-                                onOpen = { streamError = null },
+                                try {
-                                onReset = {
+                                    source.follow(
-                                    // Too far behind to continue from: what is on screen is a stale
+                                        after = lastSeq.get(),
-                                    // prefix of a conversation that has moved on, and the window
+                                        onOpen = { delivery.trySend(TranscriptDelivery.Opened) },
-                                    // arriving next is not adjacent to it. Dropping the rows makes
+                                        onReset = { delivery.trySend(TranscriptDelivery.Reset) },
-                                    // this the same as opening the screen. The cache needs no
+                                    ) { entry ->
-                                    // telling: the window's first seq is not the one it expected,
+                                        delivery.trySend(TranscriptDelivery.Event(entry))
-                                    // which closes its live run and starts another.
+                                    }
-                                    dropLoadedTranscript()
+                                    delivery.close()
-                                },
+                                } catch (error: Throwable) {
-                            ) { entry ->
+                                    delivery.close(error)
-                                apply(entry)
+                                }
                             }
+                        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() }
+        snapshotFlow { Triple(listState.isScrollInProgress, atNewest, held.isNotEmpty()) }
-            .collect { due ->
+            .collect { (scrolling, newest, hasHeld) ->
-                if (!due) return@collect
+                if (scrolling && !newest) followingNewest = false
+                if (!newest) return@collect
+                followingNewest = true
+                if (!hasHeld) return@collect
                 val backlog = held
                 held = listOf()
                 backlog.forEach { record(it) }

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<TranscriptItem>, entry: SeqEvent): List<TranscriptItem
                 items + between + TranscriptItem.AssistantMsg(entry.seq, event.delta)
             }
         }
+        is SessionEvent.AssistantTextFinal -> {
+            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(

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

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
+            // also reports deltas, but they are provisional: safety buffering can revise their
-            // must always be skipped. That rule cannot live in an in-memory set: after a backend
+            // text before completion. Keep its completed copy as an append-only correction rather
-            // restart the deltas are behind the persisted log cursor while the completion is not,
+            // than guessing that the two representations concatenate to the same answer.
-            // which used to append the whole message again after its already-recorded prefix.
             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<Event> {
     }
 }
 
+fn final_item(item: &Value) -> Vec<Event> {
+    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!(
+        assert_eq!(
-            translator
+            translator.translate(&line(
-                .translate(&line(
+                r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"hello, revised"}}}"#
-                    r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"hello"}}}"#
+            )),
-                ))
+            vec![Event::AssistantTextFinal {
-                .is_empty()
+                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!(
+        assert_eq!(
-            adopted
+            adopted.translate(&line(
-                .translate(&line(
+                r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"the complete message"}}}"#
-                    r#"{"method":"item/completed","params":{"item":{"id":"message-1","type":"agentMessage","text":"the complete message"}}}"#
+            )),
-                ))
+            vec![Event::AssistantTextFinal {
-                .is_empty()
+                text: "the complete message".to_string()
+            }]
         );
     }
 }

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,

server/src/session/llama.rs

@@ -497,6 +497,9 @@ fn conversation(path: &Path) -> Vec<Message> {
                 });
             }
             Event::AssistantText { delta } => pending.push_str(&delta),
+            Event::AssistantTextFinal { text } => {
+                pending = text;
+            }
             _ => {}
         }
     }

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(),