diff --git a/AGENTS.md b/AGENTS.md index daa9847..1a890b3 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -286,8 +286,12 @@ first if a remote spawn ever mangles an argument. cross-session message to a real stream-json session: no `user` record, and nothing in the partial-message stream -- the whole of it is an `origin` object on the `result`, the same shape the session file records, which is - why `import::peer_message` reads both. So the note is drawn *after* the - reply it caused; that is the wire, not a bug. See PLAN.md. + why `import::peer_message` reads both. So it is *recorded* after the reply + it caused, and cannot be recorded anywhere else in an append-only log -- + which is why the event carries `turnStart`, the seq of the status that + opened its turn, and the phone draws the note at that seq instead of where + it arrived. Exercise it with the echo driver's `/peer-turn`; plain `/peer` + is the in-place shape an import replays. See PLAN.md. - **A queued message can be tapped to take it back**, which is `POST /sessions/{id}/unqueue` and a `messageDropped` event -- see PLAN.md's "Taking a queued message back". On a **Claude** session it always refuses, diff --git a/PLAN.md b/PLAN.md index ed0064a..bbb68ca 100644 --- a/PLAN.md +++ b/PLAN.md @@ -300,13 +300,30 @@ both, and there is one function for one wire format. Only peer-caused turns carry it: four ordinary results on a real session's stdout had no `origin` between them. -**The cost is the position.** The note lands after the reply it caused rather -than above it, because at no earlier point in the turn does the CLI say why -the turn started. The alternative is a second reader tailing the CLI's own -session file for the one record stdout does not carry — two sources of truth -for one conversation and a poll per live session — and it was rejected on -that. If the CLI ever announces the injection at the point it happens, this -moves to that record and the ordering comes right with it. +**The cost was the position, and it is paid on the wire rather than on +screen** (2026-09-01). The event cannot be *recorded* in place: at no earlier +point in the turn does the CLI say why the turn started, and the transcript +is append-only, so by the time anyone knows, everything the message caused +has already been written above it. Reading it out of the CLI's own session +file instead — a second reader tailing the one record stdout does not carry — +was rejected then and stays rejected: two sources of truth for one +conversation and a poll per live session. + +So the event carries **where it belongs** instead. `PeerMessage` has a +`turnStart`: the seq of the `Status` that opened the turn it started, stamped +by the pump, which is the only thing that knows a seq and the only thing that +sees every driver's turns. The phone gives the note that seq, so it sorts +into the transcript above the turn rather than being drawn out of order at +the end. That seq belongs to a status change, and a status draws no row, so +there is nothing for the note to collide with and the list stays sorted — +which is what the scroll anchor and paging depend on. + +`turnStart` is absent where there is nothing to correct: a message replayed +out of a session file by `import` is already in the right place, and one that +opened no turn has no turn to sit above. Both are drawn where they arrive. +The echo driver models both shapes — `/peer` for the in-place one, and +`/peer-turn` for the live one, which reveals the note only after a reply and +a run of tool calls. ### Taking a queued message back (decided 2026-08-31) 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 771ba89..859c98c 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt @@ -99,7 +99,19 @@ sealed class SessionEvent { * claim they had. It is also the explanation for a session that starts working on something * this device never asked for. */ - data class PeerMessage(val from: String, val text: String) : SessionEvent() + data class PeerMessage( + val from: String, + val text: String, + /** + * Where the turn this started begins, when the server could say. + * + * The live Claude Code path only learns a turn was somebody else's when the turn ends, so + * the event arrives below everything it caused; this is what puts it back above it. Null + * for a message read out of a session file, which is already in the right place, and for + * one that started no turn. See the server's `Event::PeerMessage`. + */ + val turnStart: Long? = null, + ) : SessionEvent() /** * A command the session was asked to run on itself and cannot run yet. @@ -237,7 +249,11 @@ fun parseSeqEvent(json: String): SeqEvent { }, ) "peerMessage" -> - SessionEvent.PeerMessage(body.getString("from"), body.getString("text")) + SessionEvent.PeerMessage( + body.getString("from"), + body.getString("text"), + if (body.has("turnStart")) body.getLong("turnStart") else null, + ) "commandQueued" -> SessionEvent.CommandQueued(body.getString("id"), body.getString("text")) "commandSent" -> SessionEvent.CommandSent(body.getString("id"), body.getString("text")) 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 ea36397..d624fa2 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt @@ -268,6 +268,56 @@ private fun adoptRun( tail.map { (it as TranscriptItem.ToolRun).copy(runId = joining) } } +/** + * A peer message goes above the turn it started, not where it happened to arrive. + * + * The live Claude Code path cannot record it in place: the CLI says nothing about a peer message + * until the turn's `result`, so the event lands below the whole reply it caused -- the answer + * printed above the question. The server stamps it with where that turn began + * ([SessionEvent.PeerMessage.turnStart]) and the note takes that seq, so it sorts into the list + * where it belongs rather than being drawn out of order at the end. + * + * Taking the turn's opening seq as its own is also what keeps the list sorted, which anchors and + * paging both depend on. That seq belongs to a status change, and a status draws no row, so there + * is nothing for it to collide with. + * + * Without a stamp -- a message replayed out of a session file, which is already in the right place + * -- it stays where it arrived. + */ +private fun placePeerNote( + items: List, + seq: Long, + event: SessionEvent.PeerMessage, +): List { + val at = event.turnStart ?: return items + TranscriptItem.PeerNote(seq, event.from, event.text) + val note = TranscriptItem.PeerNote(at, event.from, event.text) + val index = items.indexOfFirst { it.seq > at } + if (index < 0) return items + note + val behind = (items.getOrNull(index - 1) as? TranscriptItem.ToolRun)?.runId + return items.subList(0, index) + note + splitRun(items.subList(index, items.size), behind) +} + +/** + * The calls the note now sits in front of, renamed if they were sharing a run with the calls behind + * it. + * + * A run is named from what a call landed next to (see [runIdFor]), and nothing there knows about + * turns -- so a turn opening with a tool call, straight after one that ended with one, folds them + * into a single run. Left alone, [groupToolRuns] would flush at the note and hand both halves the + * same name: two rows with one key, which a keyed list cannot draw at all. + * + * The later half is the one renamed, which is the opposite of a page join ([adoptRun]) and right + * for the opposite reason. There the two halves were always one run and the newer was already on + * screen; here they were never one turn's work, and both halves change appearance at the same + * moment the note appears between them. + */ +private fun splitRun(tail: List, behind: String?): List { + val first = tail.firstOrNull() as? TranscriptItem.ToolRun ?: return tail + if (behind == null || first.runId != behind) return tail + val run = tail.takeWhile { it is TranscriptItem.ToolRun && it.runId == behind } + return run.map { (it as TranscriptItem.ToolRun).copy(runId = first.id) } + tail.drop(run.size) +} + fun foldEvent(items: List, entry: SeqEvent): List = when (val event = entry.event) { is SessionEvent.UserMessage -> @@ -361,8 +411,7 @@ fun foldEvent(items: List, entry: SeqEvent): List it } } - is SessionEvent.PeerMessage -> - items + TranscriptItem.PeerNote(entry.seq, event.from, event.text) + is SessionEvent.PeerMessage -> placePeerNote(items, entry.seq, event) is SessionEvent.CommandSent -> items + TranscriptItem.CommandRow(entry.seq, event.text) // Screen-level state, not transcript rows -- see SessionScreen. is SessionEvent.CommandQueued -> items diff --git a/server/src/session/claude/translate.rs b/server/src/session/claude/translate.rs index 17a7d51..692ed6c 100644 --- a/server/src/session/claude/translate.rs +++ b/server/src/session/claude/translate.rs @@ -1147,6 +1147,9 @@ mod tests { Event::PeerMessage { from: "ai-app-2-fb".to_string(), text: "Reply with just the word ACK.".to_string(), + // Stamped by the pump, which is the only place that + // knows what seq the turn started at. + turn_start: None, }, Event::UsageDelta { tokens: 7, diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index 59a1ceb..f80ca29 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -225,6 +225,24 @@ pub enum Event { /// recognises it by -- the socket path it came from is not. from: String, text: String, + /// The seq of the `Status::Running` that opened the turn this + /// message started, so a reader can draw it above that turn. + /// + /// It exists because the live Claude Code path cannot record the + /// message where it belongs. The CLI says nothing about a peer + /// message until the turn's `result` -- see + /// `claude::translate` -- so the event is appended after + /// everything it caused, and an append-only transcript cannot go + /// back and insert it. Carrying the position instead keeps one + /// order on the wire and one order on screen without a second + /// source for either. + /// + /// Filled in by the pump, which is the only place that knows a + /// seq, and only where a turn was open: `None` for a message read + /// out of a session file by `import`, which already has it in the + /// right place, and for one that started no turn. + #[serde(default, skip_serializing_if = "Option::is_none")] + turn_start: Option, }, /// The manager's record of a question being answered, so a rendered /// question card resolves on every device, not just the one that diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index 877518f..7266050 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -278,6 +278,42 @@ impl EchoDriver { // asked for it is still announced -- every driver owes exactly one // `MessageTaken` per message, and a command that quietly vanishes // from the transcript is the one thing echo must not model. + // The live Claude Code shape, which is the one the ordering has to + // survive: the CLI says nothing about a peer message until the + // turn's `result`, so the event arrives below the whole reply it + // caused and the phone has to put it back. Checked before `/peer`, + // which would otherwise take the rest of this word as the body. + if let Some(rest) = text.strip_prefix("/peer-turn") { + if announce { + self.emit(Event::MessageTaken { + id: None, + text: text.clone(), + images: images.clone(), + }); + } + self.emit(Event::Status { + state: SessionStatus::Running, + }); + self.some_calls("peer"); + self.emit(Event::AssistantText { + delta: "Pulled, and AGENTS.md is up to date here now.".to_string(), + }); + self.emit(Event::PeerMessage { + from: "dev-updater-f5".to_string(), + text: if rest.trim().is_empty() { + "Pull before you touch AGENTS.md.".to_string() + } else { + rest.trim().to_string() + }, + // Stamped by the manager, exactly as a real one is. + turn_start: None, + }); + self.emit(Event::Status { + state: SessionStatus::Idle, + }); + return; + } + if let Some(rest) = text.strip_prefix("/peer") { if announce { self.emit(Event::MessageTaken { @@ -296,6 +332,7 @@ impl EchoDriver { } else { rest.trim().to_string() }, + turn_start: None, }); return; } @@ -710,6 +747,7 @@ async fn write_beat(sink: &EventSink, session_dir: &Path, beat: usize) { send(Event::PeerMessage { from: format!("beat-{beat}-peer"), text: format!("Message {beat} from another session, for the row it becomes."), + turn_start: None, }); } } diff --git a/server/src/session/import.rs b/server/src/session/import.rs index 67f460d..87db289 100644 --- a/server/src/session/import.rs +++ b/server/src/session/import.rs @@ -550,6 +550,8 @@ pub(in crate::session) fn peer_message(record: &Value) -> Option { .unwrap_or("another session") .to_string(), text: origin.get("body").and_then(Value::as_str)?.to_string(), + // The session file has it in the right place already. + turn_start: None, }) } @@ -1151,6 +1153,7 @@ mod tests { from: "dev-updater-f5".to_string(), // The body, not the wrapper the model is given. text: "Pull before you touch AGENTS.md.".to_string(), + turn_start: None, }, "{events:?}" ); diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index c46ece6..1835d27 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -2200,14 +2200,29 @@ async fn pump( // the order the transcript has them -- and because the answer has to // survive being asked a moment later than the driver would have said it. let mut unread: usize = 0; + // Where the turn currently running began: the seq of the `Status` + // that opened it, which is recorded before any of the turn's own + // output. Held here because the pump is the only place that knows a + // seq at all, and the only one that sees every driver's turns. + let mut turn_start: Option = None; while let Some(event) = source.recv().await { let ts = now(); // Taking a message is how it enters the conversation, and the // conversation is what a phone renders -- so the event becomes the // message here rather than being carried alongside it. One rule // for where a user's message sits: where the session read it. + // + // A peer message is stamped with the same knowledge for the + // opposite reason: it arrives *after* everything it caused, and + // the position is the only way a reader can put it back where it + // happened -- see `Event::PeerMessage::turn_start`. let event = match event { Event::MessageTaken { id, text, images } => Event::UserMessage { id, text, images }, + Event::PeerMessage { from, text, .. } => Event::PeerMessage { + from, + text, + turn_start, + }, other => other, }; // Where the session row's figure comes from. Kept here rather than @@ -2264,6 +2279,20 @@ async fn pump( } *shared.last_activity.lock().unwrap() = ts; *shared.written.lock().unwrap() += 1; + // The turn's own first line, kept for whatever arrives at + // the end of it needing to say where it started. Only the + // *opening* status counts: a turn that pauses for a + // question or a compaction and resumes is still the turn + // that began where it began. + match &entry.event { + Event::Status { + state: SessionStatus::Running, + } if turn_start.is_none() => turn_start = Some(entry.seq), + Event::Status { + state: SessionStatus::Idle | SessionStatus::Exited, + } => turn_start = None, + _ => {} + } // The boundary a held command was waiting for, and the one // place that sees every driver's. Done after the status is // recorded, so the command that runs next sees an idle @@ -2423,6 +2452,62 @@ mod tests { assert!(matches!(events.try_recv(), Ok(Event::CommandSent { .. }))); } + /// A peer message is stamped with where its turn began, so a phone can + /// draw it above the reply it caused rather than below it. + /// + /// The live CLI reveals the message only on the turn's `result`, and an + /// append-only transcript cannot go back and insert it -- so the + /// position has to travel with the event. Without it the note is drawn + /// at the end of the turn, which reads as an answer printed above its + /// question. + #[tokio::test] + async fn a_peer_message_carries_the_seq_its_turn_started_at() { + let dir = tempfile::tempdir().expect("tempdir"); + let config_path = dir.path().join("config.ron"); + let data_dir = dir.path().join("sessions"); + seed_echo_only(&config_path); + let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models")) + .expect("manager"); + let info = manager.spawn_session(echo_spec()).expect("spawn"); + let session = manager.session(&info.id).expect("live"); + let mut rx = session.subscribe(); + + session.send_message("/peer-turn".to_string(), Vec::new()); + let seen = collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await; + + let opened = seen + .iter() + .find(|entry| { + matches!( + entry.event, + Event::Status { + state: SessionStatus::Running + } + ) + }) + .expect("the turn's opening status") + .seq; + let note = seen.last().expect("the peer message"); + let Event::PeerMessage { turn_start, .. } = ¬e.event else { + unreachable!() + }; + assert_eq!(*turn_start, Some(opened), "{seen:?}"); + // And it is worth stamping only because it is genuinely behind the + // turn it explains. + assert!(note.seq > opened, "{seen:?}"); + + // A message that opened no turn is left where it arrived: an + // import replays those in place already. + session.send_message("/peer".to_string(), Vec::new()); + let alone = + collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await; + let Some(Event::PeerMessage { turn_start, .. }) = alone.last().map(|entry| &entry.event) + else { + unreachable!() + }; + assert_eq!(*turn_start, None, "{alone:?}"); + } + /// A command waits for the *driver* to be between turns, not for the /// recorded status to say idle. ///