diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Compaction.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Compaction.kt index 7c24db4..56065fa 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Compaction.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Compaction.kt @@ -1,7 +1,9 @@ package com.example.aiapp +import androidx.compose.foundation.layout.Column import androidx.compose.foundation.layout.fillMaxWidth import androidx.compose.foundation.layout.padding +import androidx.compose.material3.LinearProgressIndicator import androidx.compose.material3.MaterialTheme import androidx.compose.material3.Text import androidx.compose.runtime.Composable @@ -51,3 +53,56 @@ fun compactionSummary(item: TranscriptItem.CompactedNote): String { } private fun tokens(count: Long): String = "%,d".format(count) + +/** + * Where the working indicator goes while a compaction is running. + * + * A bar across the whole row rather than the spinner an ordinary turn gets, because a compaction is + * not an ordinary turn: nothing arrives in the transcript while it runs, so the row it occupies is + * the only thing on screen that is moving, and at the width of a spinner that reads as a session + * that might have hung. + * + * The bar is indeterminate, and that is a statement rather than an omission. The CLI says a + * compaction has started and then says nothing at all until it has finished -- measured, not + * assumed -- so there is no fraction to fill, and a bar that crept along at the pace of the last + * compaction would be this screen inventing the part nobody sent it. What it can honestly say is + * that work is happening and for how long, which is [compactingLabel]. + */ +@Composable +fun CompactingRow(seconds: Long?, modifier: Modifier = Modifier) { + Column(modifier.fillMaxWidth()) { + Text( + compactingLabel(seconds), + style = MaterialTheme.typography.bodySmall, + // The colour is stated beside the fill rather than inherited: a semantic colour has + // to carry its own contrast, since the surface under it will not change to rescue it. + color = compactingColor, + ) + LinearProgressIndicator( + modifier = Modifier.fillMaxWidth().padding(top = 6.dp), + color = compactingColor, + trackColor = MaterialTheme.colorScheme.surfaceContainerHigh, + ) + } +} + +/** + * What the working indicator says while a compaction is running. + * + * Elapsed time and nothing else, because elapsed time is all there is: the CLI announces that a + * compaction has begun and then says nothing until it has finished, so any bar, percentage or + * estimate here would be this screen's guess wearing a measurement's clothes. Knowing it has been + * going forty seconds is what a reader actually wants -- it is the difference between waiting and + * going to look at why. + * + * [seconds] is null when this device did not see the compaction start, which is what opening a + * session that is already compacting looks like. That case says only "compacting": no number is the + * honest answer, and a number counted from the moment the screen opened would be wrong in the + * direction that matters, since a compaction somebody is asking about is a long one. + */ +fun compactingLabel(seconds: Long?): String = + when { + seconds == null -> "compacting" + seconds < 60 -> "compacting ${seconds}s" + else -> "compacting ${seconds / 60}m ${seconds % 60}s" + } 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 eaf547a..b094d34 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt @@ -37,6 +37,15 @@ sealed class SessionEvent { data class Answered(val id: String, val answer: String) : SessionEvent() + /** + * A message another agent sent this session. + * + * Not a [UserMessage]: nobody holding the phone said it, and drawing it in their voice would + * 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 Status(val state: String) : SessionEvent() data class UsageDelta(val tokens: Long) : SessionEvent() @@ -96,6 +105,8 @@ fun parseSeqEvent(json: String): SeqEvent { about = body.optString("about").ifEmpty { null }, ) "answered" -> SessionEvent.Answered(body.getString("id"), body.getString("answer")) + "peerMessage" -> + SessionEvent.PeerMessage(body.getString("from"), body.getString("text")) "status" -> SessionEvent.Status(body.getString("state")) "usageDelta" -> SessionEvent.UsageDelta(body.getLong("tokens")) "compacted" -> diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/PeerMessage.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/PeerMessage.kt new file mode 100644 index 0000000..ff4ad7a --- /dev/null +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/PeerMessage.kt @@ -0,0 +1,57 @@ +package com.example.aiapp + +import androidx.compose.foundation.clickable +import androidx.compose.foundation.layout.Column +import androidx.compose.foundation.layout.Row +import androidx.compose.foundation.layout.Spacer +import androidx.compose.foundation.layout.fillMaxWidth +import androidx.compose.foundation.layout.padding +import androidx.compose.foundation.layout.width +import androidx.compose.material3.Card +import androidx.compose.material3.MaterialTheme +import androidx.compose.material3.Text +import androidx.compose.runtime.Composable +import androidx.compose.ui.Alignment +import androidx.compose.ui.Modifier +import androidx.compose.ui.text.style.TextOverflow +import androidx.compose.ui.unit.dp + +/** + * A message another agent sent this session, closed until somebody asks. + * + * Closed by default, like a tool call and for the same reason: these are long, there can be several + * in a row, and what a reader scanning the transcript needs from one is that it happened and who + * sent it. The first line comes with the heading because a name alone does not say which message + * this was. + * + * Drawn as its own kind rather than as the reader's own bubble. They did not say this, and a + * transcript that puts it in their voice is making a claim about who asked for the work that + * follows -- which is exactly the question a peer message is usually the answer to. + */ +@Composable +fun PeerMessageRow( + item: TranscriptItem.PeerNote, + expanded: Boolean, + onToggle: () -> Unit, + modifier: Modifier = Modifier, +) { + Card(modifier.fillMaxWidth().clickable(onClick = onToggle)) { + Column(Modifier.padding(12.dp)) { + Row(verticalAlignment = Alignment.CenterVertically) { + Text("Message from ${item.from}", style = MaterialTheme.typography.titleSmall) + if (!expanded) { + Spacer(Modifier.width(8.dp)) + Text( + item.text.lineSequence().firstOrNull { it.isNotBlank() }.orEmpty(), + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + maxLines = 1, + // The head, not the tail: a message is identified by how it opens. + overflow = TextOverflow.Ellipsis, + ) + } + } + if (expanded) MarkdownText(item.text, Modifier.padding(top = 6.dp)) + } + } +} diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt index c03427b..5daebe4 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionListScreen.kt @@ -261,7 +261,7 @@ fun StatusText(status: String) { when (status) { "awaitingInput" -> "your turn" to awaitingColor "running" -> "running" to runningColor - "compacting" -> "compacting" to runningColor + "compacting" -> "compacting" to compactingColor "exited" -> "exited" to MaterialTheme.colorScheme.onSurfaceVariant // Said in words, because it differs in kind from the others rather than in degree: // the session is not idle and has not exited, nobody has been able to find out @@ -271,9 +271,12 @@ fun StatusText(status: String) { } Row(verticalAlignment = Alignment.CenterVertically) { if (status == "running" || status == "compacting") { + // The same colour as the word beside it: the two are one signal, and a spinner in + // the theme's accent says the state is something other than what the label says. CircularProgressIndicator( modifier = Modifier.width(14.dp).height(14.dp), strokeWidth = 2.dp, + color = color, ) Spacer(Modifier.width(6.dp)) } 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 a07efe5..1c11191 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -1,5 +1,6 @@ package com.example.aiapp +import android.os.SystemClock import androidx.activity.compose.rememberLauncherForActivityResult import androidx.activity.result.PickVisualMediaRequest import androidx.activity.result.contract.ActivityResultContracts @@ -125,6 +126,14 @@ sealed class TranscriptItem { /** An image by server-side ref, fetched from the session's files route. */ data class ImageItem(override val seq: Long, val ref: String) : TranscriptItem() + /** + * A message another agent sent this session. + * + * Its own row rather than a [UserMsg]: see [PeerMessageRow] for why the voice matters. + */ + data class PeerNote(override val seq: Long, val from: String, val text: String) : + TranscriptItem() + /** Placeholder row for events this build can't render (newer kinds). */ data class Note(override val seq: Long, val text: String) : TranscriptItem() @@ -222,6 +231,8 @@ fun foldEvent(items: List, entry: SeqEvent): List it } } + is SessionEvent.PeerMessage -> + items + TranscriptItem.PeerNote(entry.seq, event.from, event.text) is SessionEvent.Status -> items is SessionEvent.Error -> items + TranscriptItem.ErrorMsg(entry.seq, event.message) is SessionEvent.Image -> @@ -269,6 +280,11 @@ fun SessionScreen( var items by remember { mutableStateOf(listOf()) } var status by remember { mutableStateOf(summary.status) } var totalTokens by remember { mutableLongStateOf(0L) } + // When this screen saw the current compaction start, on this device's own clock, and how long + // ago that is. See `compactingLabel`: null is the honest answer whenever the start was not + // witnessed here, which is what opening a session that is already compacting looks like. + var compactingSince by remember { mutableStateOf(null) } + var compactingFor by remember { mutableStateOf(null) } var streamError by remember { mutableStateOf(null) } var actionError by remember { mutableStateOf(null) } var input by remember { mutableStateOf("") } @@ -276,6 +292,10 @@ fun SessionScreen( // Which runs of adjacent tool calls are open. Keyed by the first call's // id, so a group survives more calls arriving after it. var expandedGroups by remember { mutableStateOf(setOf()) } + // Which messages from other agents are open, by the seq that identifies their row. Closed + // by default, which is the rule for anything new in this transcript: a screen that opens + // everything it can is one nobody can scan. + var expandedNotes by remember { mutableStateOf(setOf()) } // Uploaded-but-not-yet-sent attachment ids; sent with the next message. var pendingAttachments by remember { mutableStateOf(listOf()) } // What this session is set to now, seeded from the row that opened it and @@ -326,7 +346,20 @@ fun SessionScreen( when (val event = entry.event) { is SessionEvent.UsageDelta -> totalTokens += event.tokens else -> { - if (event is SessionEvent.Status) status = event.state + if (event is SessionEvent.Status) { + // Started here, or nowhere. `ready` is what separates the live stream from + // the page of history the screen opens with, and a compaction found in that + // page began before anybody here was watching -- timing it from now would + // report the moment we arrived as the moment it started. + compactingSince = + when { + event.state != "compacting" -> null + status == "compacting" -> compactingSince + ready -> SystemClock.elapsedRealtime() + else -> null + } + status = event.state + } // The message coming back is the session saying it has // read it, so the bubble held below the indicator becomes // the row `foldEvent` is about to add. Matched by text -- @@ -344,6 +377,22 @@ fun SessionScreen( } } + // A compaction reports nothing about its own progress -- measured against the CLI, which + // says it has started, and then says nothing at all until it is done. So what this counts is + // the one thing anybody here can measure: how long it has been going. A bar filling up would + // be this screen inventing the part the CLI does not send. + LaunchedEffect(compactingSince) { + val since = compactingSince + if (since == null) { + compactingFor = null + return@LaunchedEffect + } + while (true) { + compactingFor = (SystemClock.elapsedRealtime() - since) / 1000 + delay(1000) + } + } + // The stream lifecycle: connect, follow, and on any drop reconnect // from the cursor -- so a flaky link (or a backend restart) costs // nothing but the gap's latency. @@ -423,18 +472,29 @@ fun SessionScreen( // stays started. DisposableEffect(summary.id) { onDispose { activeStream.get()?.close() } } - // Whether the view is pinned to the newest message. The list is laid - // out from the bottom (see the LazyColumn below), so "newest" is index - // 0 and being pinned is simply being at the start of it. - // - // Read from the scroll rather than remembered as a flag: with the list - // anchored this way there is no moment where new content pushes the - // anchor away, so there is nothing to protect a remembered value from. - val followTail by remember { + // Whether the newest message is on screen right now. The list is laid out from the bottom + // (see the LazyColumn below), so "newest" is index 0 and being there is being at the start of + // it. This is what the jump-to-newest button watches: it is about what the reader can see. + val atNewest by remember { derivedStateOf { listState.firstVisibleItemIndex == 0 && listState.firstVisibleItemScrollOffset == 0 } } + // Whether they *chose* to be there, which is a different question and the one that decides + // whether an arriving message brings the view with it. + // + // Remembered, and only ever written when a scroll settles -- so it records where the reader + // last left the list, and an insertion cannot change the answer. Reading the live position + // instead looks right and is subtly wrong: a keyed list moves its anchor to keep the reader's + // content still, so by the time the new item can be observed the view is already one item + // away from the newest and reports itself as scrolled back. The message then never followed, + // which was visible as a compaction whose progress bar sat just off the bottom of the screen + // while the button that started it said it was running. + var followTail by remember { mutableStateOf(true) } + LaunchedEffect(listState) { + snapshotFlow { listState.isScrollInProgress } + .collect { scrolling -> if (!scrolling) followTail = atNewest } + } // A new item at the newest end shifts every index by one, so the view // has to step back to 0 to stay put. One item, instantly -- not a // journey through the transcript. @@ -443,8 +503,15 @@ fun SessionScreen( // from one line to four and the keyboard opens under it, and neither // is a new message, so watching the item count alone leaves the newest // text drifting out of sight while somebody writes a reply to it. + // + // Counted in list items rather than in transcript rows, because the rows are not all of it: + // the working indicator and a queued message are items too, and they arrive at exactly the + // same end. Sibling to the paging trigger below, which is the same count read from the other + // end for the same reason. LaunchedEffect(listState) { - snapshotFlow { Pair(items.size, listState.layoutInfo.viewportSize.height) } + snapshotFlow { + Pair(listState.layoutInfo.totalItemsCount, listState.layoutInfo.viewportSize.height) + } .collect { (count, _) -> if (followTail && count > 0) listState.scrollToItem(0) } } // Reaching the far end of what is loaded -- the oldest item, which in @@ -658,17 +725,24 @@ fun SessionScreen( // where the newest message is. if (running) { item(key = "indicator") { - Row(verticalAlignment = Alignment.CenterVertically) { - CircularProgressIndicator( - modifier = Modifier.width(14.dp).height(14.dp), - strokeWidth = 2.dp, - ) - Spacer(Modifier.width(8.dp)) - Text( - if (status == "compacting") "compacting" else "working", - style = MaterialTheme.typography.bodySmall, - color = MaterialTheme.colorScheme.onSurfaceVariant, - ) + // Two shapes for two kinds of busy: an ordinary turn is a spinner beside + // a word, because the answer it is producing appears directly below it, + // and a compaction takes the whole row because nothing else will. + if (status == "compacting") { + CompactingRow(compactingFor) + } else { + Row(verticalAlignment = Alignment.CenterVertically) { + CircularProgressIndicator( + modifier = Modifier.width(14.dp).height(14.dp), + strokeWidth = 2.dp, + ) + Spacer(Modifier.width(8.dp)) + Text( + "working", + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.onSurfaceVariant, + ) + } } } } else if (status == "exited") { @@ -768,6 +842,17 @@ fun SessionScreen( color = MaterialTheme.colorScheme.onSurfaceVariant, ) is TranscriptItem.CompactedNote -> CompactedRow(item) + is TranscriptItem.PeerNote -> + PeerMessageRow( + item = item, + expanded = item.seq in expandedNotes, + onToggle = { + expandedNotes = + if (item.seq in expandedNotes) + expandedNotes - item.seq + else expandedNotes + item.seq + }, + ) } } } @@ -783,7 +868,7 @@ fun SessionScreen( // which is where this goes. The name is carried in the // description, since an arrow alone says nothing to a screen // reader and nothing to whoever finds this in six months. - if (!followTail) { + if (!atNewest) { Surface( onClick = { scope.launch { listState.animateScrollToItem(0) } }, shape = CircleShape, diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Theme.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Theme.kt index d7b1e31..2fec588 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Theme.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Theme.kt @@ -109,6 +109,18 @@ val runningColor: Color val failedColor: Color @Composable get() = MaterialTheme.colorScheme.error +/** + * Working on the conversation rather than in it: a compaction. + * + * Its own colour because it is its own kind of busy. Everything else a session does is progress + * through the task; this is the session rewriting what it remembers, it can take minutes, and + * nothing it produces appears in the transcript until it is over. A reader who has learned that + * blue means "not stuck, but not answering you either" has learned the only thing that + * distinguishes it from a session that has hung. + */ +val compactingColor: Color + @Composable get() = Mocha.Blue + /** Waiting on a person: a question, a permission, a turn that is theirs. */ val awaitingColor: Color @Composable get() = Mocha.Peach diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index 0ca7db9..037d87c 100644 --- a/server/src/session/claude.rs +++ b/server/src/session/claude.rs @@ -732,6 +732,26 @@ fn translate_line( } } } + // A turn nobody here started -- see `proves_a_turn`. Said before + // the event that proves it, for the same reason a steer is: the + // session was already working when it produced this. + let started = { + let mut queue = queue.lock().unwrap(); + let started = proves_a_turn(&event) && !queue.running && !queue.closed; + if started { + queue.running = true; + } + started + }; + if started + && sink + .send(Event::Status { + state: SessionStatus::Running, + }) + .is_err() + { + return false; + } if matches!( event, Event::Status { @@ -747,6 +767,39 @@ fn translate_line( true } +/// Whether this event could only have come from a turn in flight. +/// +/// The turn this side starts is announced where it is started, and that +/// covers the common case and nothing else. Everything below happens +/// without a phone asking for it: a compaction the CLI decided on by +/// itself, a session adopted while it was already mid-turn, a message +/// that reached the conversation by some route other than this server -- +/// another agent writing to it, or somebody at the terminal. In all of +/// them the CLI is plainly working and the only thing that would ever +/// have said so is a `Running` nobody sent, so the session sits there +/// reading as idle until the turn ends. +/// +/// So the driver says it from what it observes rather than from what it +/// was asked to do, and this is the same set as [`announces_a_steer`] +/// with the ends swapped: that one takes the `Idle` that closes a turn +/// and this one takes the states that open one. `Idle` is the pair to +/// this -- it is where `running` goes back to false, a few lines above +/// where it is set here. +fn proves_a_turn(event: &Event) -> bool { + matches!( + event, + Event::AssistantText { .. } + | Event::ToolStart { .. } + | Event::ToolUpdate { .. } + | Event::ToolEnd { .. } + | Event::Question { .. } + | Event::Compacted { .. } + | Event::Status { + state: SessionStatus::Compacting | SessionStatus::AwaitingInput + } + ) +} + /// Whether this event proves the CLI has consumed anything written to it /// since the last one did. /// @@ -985,6 +1038,73 @@ mod tests { assert!(queue.closed); } + #[test] + fn a_turn_this_side_did_not_start_still_reports_as_running() { + // The case: a session picked up while it was already working, or + // one another agent wrote to. Nothing called `send_user_message`, + // so the only thing that can say the session is busy is what it + // is observed doing. + let dir = tempfile::tempdir().expect("tempdir"); + let (sink, mut received) = mpsc::unbounded_channel(); + let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf()))); + let queue = Arc::new(Mutex::new(Queue::default())); + + let text = r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"working"}},"parent_tool_use_id":null}"#; + assert!(translate_line(text, dir.path(), &state, &sink, &queue)); + assert_eq!( + received.try_recv().ok(), + Some(Event::Status { + state: SessionStatus::Running + }), + "a turn in flight has to be reported before the output proving it" + ); + assert!(matches!( + received.try_recv().ok(), + Some(Event::AssistantText { .. }) + )); + + // Once only: the turn is known to be running now, and a status per + // delta would be a status per word. + assert!(translate_line(text, dir.path(), &state, &sink, &queue)); + assert!(matches!( + received.try_recv().ok(), + Some(Event::AssistantText { .. }) + )); + + // And the end of the turn puts it back, so the next one is + // reported the same way. + let done = r#"{"type":"result","subtype":"success","is_error":false,"usage":{}}"#; + assert!(translate_line(done, dir.path(), &state, &sink, &queue)); + assert_eq!( + received.try_recv().ok(), + Some(Event::Status { + state: SessionStatus::Idle + }) + ); + assert!(!queue.lock().unwrap().running); + } + + #[test] + fn output_from_a_process_that_has_gone_does_not_revive_the_turn() { + // `close` is what says the process is gone and reports the + // messages that died with it. Anything still in the pipe after + // that must not put the session back to work, because there is + // nothing left to do the work. + let dir = tempfile::tempdir().expect("tempdir"); + let (sink, mut received) = mpsc::unbounded_channel(); + let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf()))); + let queue = Arc::new(Mutex::new(Queue::default())); + queue.lock().unwrap().close(&sink, "the session ended"); + + let text = r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"late"}},"parent_tool_use_id":null}"#; + assert!(translate_line(text, dir.path(), &state, &sink, &queue)); + assert!(matches!( + received.try_recv().ok(), + Some(Event::AssistantText { .. }) + )); + assert!(!queue.lock().unwrap().running); + } + #[test] fn closing_an_empty_queue_says_nothing() { let (sink, mut received) = mpsc::unbounded_channel(); diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index 11607ad..1c36f9e 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -104,6 +104,19 @@ pub enum Event { #[serde(default, skip_serializing_if = "Option::is_none")] about: Option, }, + /// A message another agent sent this session. + /// + /// Its own kind rather than a `UserMessage`, because it is not + /// something the reader said and a transcript that renders it in their + /// voice is claiming they did. It also explains what would otherwise + /// be inexplicable: a session that starts working on something nobody + /// on this phone asked for. + PeerMessage { + /// The sending session's own name, which is what the reader + /// recognises it by -- the socket path it came from is not. + from: String, + text: String, + }, /// The manager's record of a question being answered, so a rendered /// question card resolves on every device, not just the one that /// answered it. diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index b82651d..a059ab5 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -13,6 +13,8 @@ //! - `/slow [seconds]` -- a turn that stays running (default 30), so states that only //! exist *while* something is happening can be looked at. //! - `/error [text]` -- a failure, which is otherwise awkward to cause. +//! - `/peer [text]` -- a message from another agent, which otherwise takes +//! two live sessions and one of them deciding to write. //! //! This is exactly the event vocabulary the real drivers produce, so a UI //! that renders echo sessions correctly renders the real thing. @@ -36,11 +38,14 @@ use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus}; /// stay fast. const DELTA_DELAY: Duration = Duration::from_millis(50); -/// How long a fake compaction takes. A real one runs for a minute or two, -/// which is too long to sit through when what is being checked is what the -/// screen does; this is long enough that the state is visible and short -/// enough to wait for. -const COMPACT_TIME: Duration = Duration::from_secs(3); +/// How long a fake compaction takes. +/// +/// A measured one, near enough: driving a real session through `/compact` +/// on 2026-08-29 took 13 seconds for a small conversation, and a large one +/// takes minutes. Three seconds -- what this was -- is too short to look +/// at the row that only exists while a compaction is running, and too +/// short to watch its elapsed count reach two digits. +const COMPACT_TIME: Duration = Duration::from_secs(13); pub struct EchoDriver { sink: EventSink, @@ -115,6 +120,28 @@ impl Driver for EchoDriver { return; } + // Answered on the spot rather than in the turn below, because a + // peer message is not a turn: it is something that arrives, and + // what is being exercised is the row it becomes. The message that + // 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. + if let Some(rest) = text.strip_prefix("/peer") { + self.emit(Event::MessageTaken { text: text.clone() }); + self.emit(Event::PeerMessage { + from: "dev-updater-f5".to_string(), + text: if rest.trim().is_empty() { + "Pull before you touch AGENTS.md -- I pushed three commits to it \ + in the last hour, and origin/main has moved since you last looked.\n\n\ + The tree is clean as of now, but it was not for most of that time." + .to_string() + } else { + rest.trim().to_string() + }, + }); + return; + } + if let Some(rest) = text.strip_prefix("/question") { let id = format!("q-{}", super::random_hex()); let prompt = if rest.trim().is_empty() { diff --git a/server/src/session/import.rs b/server/src/session/import.rs index 149db1e..b6ebc6f 100644 --- a/server/src/session/import.rs +++ b/server/src/session/import.rs @@ -410,13 +410,28 @@ pub async fn read_tail(transport: &Transport, path: &str) -> Result { /// which reads its own session file. pub fn events_from(text: &str, session_dir: &std::path::Path) -> Vec { let mut events = Vec::new(); + // What the newest record that had an opinion says the session is + // doing. Kept to the end rather than pushed as it is found, because + // the answer is the last one and everything before it is history. + let mut state = None; for line in text.lines() { let Ok(record) = serde_json::from_str::(line) else { continue; }; + if let Some(peer) = peer_message(&record) { + // Before `is_hidden`, which these records are: the CLI marks + // them meta because they are not the user's own words, and + // that is the reason to draw them differently rather than the + // reason to drop them. A session working on something a phone + // never asked for is otherwise unexplainable from the phone. + state = turn_state(&record).or(state); + events.push(peer); + continue; + } if is_hidden(&record) { continue; } + state = turn_state(&record).or(state); let Some(message) = record.get("message") else { continue; }; @@ -429,9 +444,72 @@ pub fn events_from(text: &str, session_dir: &std::path::Path) -> Vec { _ => {} } } + if let Some(state) = state { + events.push(Event::Status { state }); + } events } +/// A message from another agent, as the CLI records one. +/// +/// Measured from a real session file (2026-08-29): the record is a `user` +/// one marked `isMeta`, and its `origin` carries `kind: "peer"`, the +/// sending session's `name`, and the message itself as `body`. The +/// message content beside it is the same text wrapped in an explanatory +/// preamble and a `` tag, which is written for the +/// model that has to read it rather than for a person -- so the body is +/// what a reader is shown, and the name is who they are told sent it. +fn peer_message(record: &Value) -> Option { + let origin = record.get("origin")?; + if origin.get("kind").and_then(Value::as_str) != Some("peer") { + return None; + } + Some(Event::PeerMessage { + from: origin + .get("name") + .and_then(Value::as_str) + .unwrap_or("another session") + .to_string(), + text: origin.get("body").and_then(Value::as_str)?.to_string(), + }) +} + +/// Whether this record means the session is working, as far as it can be +/// told from the file. +/// +/// The one thing a session file does not contain is the CLI saying "this +/// turn is over": there is no `result` record, only the messages. What +/// there is instead is why the last assistant message stopped, and that +/// answers it -- `tool_use` means a call is being made and more is coming, +/// anything else means the model has finished talking. Anything on the +/// user's side of the conversation -- a person, a tool's result, another +/// agent -- means the session has something to answer and is answering it. +/// +/// `None` is the third answer and it matters: a record that says nothing +/// about the turn leaves the status alone rather than voting for idle. The +/// same goes for a record whose reason for stopping is missing, which is +/// what a future CLI adding a shape we do not know looks like. +/// +/// What this cannot see is a session that stopped existing mid-turn -- its +/// file's last record still says `tool_use`, so it reads as working +/// forever. Nothing in the file distinguishes that from a model thinking, +/// and inventing a timeout here would replace a stale reading with a +/// confident wrong one. +fn turn_state(record: &Value) -> Option { + use super::driver::SessionStatus; + match record.get("type").and_then(Value::as_str)? { + "user" => Some(SessionStatus::Running), + "assistant" => match record["message"] + .get("stop_reason") + .and_then(Value::as_str)? + { + "tool_use" => Some(SessionStatus::Running), + _ => Some(SessionStatus::Idle), + }, + _ => None, + } +} + fn push_user(events: &mut Vec, content: &Value, session_dir: &std::path::Path) { // A tool result arrives as a user record, because that is how the API // models it -- but it is the other half of a tool call, not something @@ -678,8 +756,83 @@ mod tests { let dir = tempfile::tempdir().expect("tempdir"); let line = r#"{"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"toolu_1","content":"plain output"}]}}"#; let events = events_from(line, dir.path()); - assert_eq!(events.len(), 1, "{events:?}"); + // The result, and the turn state it implies: a tool has answered, + // so the model is about to be asked again. + assert_eq!(events.len(), 2, "{events:?}"); + assert_eq!( + events[1], + Event::Status { + state: super::super::driver::SessionStatus::Running + } + ); // No stray directory for a session that never produced one. assert!(!dir.path().join("files").exists()); } + + #[test] + fn a_message_from_another_agent_is_kept_and_named() { + // The real shape, from a session file: the CLI marks these meta, + // and everything a reader needs is in `origin`. + let dir = tempfile::tempdir().expect("tempdir"); + let line = r#"{"type":"user","isMeta":true,"origin":{"kind":"peer","from":"uds:/run/user/1000/cc-socks/605.sock","verifiedPeerPid":605,"name":"dev-updater-f5","fromMode":"prompting","body":"Pull before you touch AGENTS.md."},"message":{"role":"user","content":"Another Claude session sent a message:\n\nPull before you touch AGENTS.md.\n"}}"#; + let events = events_from(line, dir.path()); + assert_eq!( + events[0], + Event::PeerMessage { + from: "dev-updater-f5".to_string(), + // The body, not the wrapper the model is given. + text: "Pull before you touch AGENTS.md.".to_string(), + }, + "{events:?}" + ); + // And it counts as the session having been given something. + assert_eq!( + events[1], + Event::Status { + state: super::super::driver::SessionStatus::Running + } + ); + } + + #[test] + fn the_last_record_says_whether_the_session_is_working() { + use super::super::driver::SessionStatus; + let dir = tempfile::tempdir().expect("tempdir"); + let asked = r#"{"type":"user","message":{"role":"user","content":"do the thing"}}"#; + let calling = r#"{"type":"assistant","message":{"role":"assistant","stop_reason":"tool_use","content":[{"type":"tool_use","id":"t1","name":"Bash","input":{}}]}}"#; + let done = r#"{"type":"assistant","message":{"role":"assistant","stop_reason":"end_turn","content":[{"type":"text","text":"done"}]}}"#; + + let state = |text: &str| { + events_from(text, dir.path()) + .into_iter() + .rev() + .find_map(|event| match event { + Event::Status { state } => Some(state), + _ => None, + }) + }; + assert_eq!(state(asked), Some(SessionStatus::Running)); + assert_eq!( + state(&[asked, calling].join("\n")), + Some(SessionStatus::Running) + ); + assert_eq!( + state(&[asked, calling, done].join("\n")), + Some(SessionStatus::Idle), + "a turn that has finished talking is over" + ); + + // A subagent's own messages are not the session's turn, and a + // record with no stop reason is not an answer -- neither may + // overrule what the conversation itself last said. + let sidechain = r#"{"type":"assistant","isSidechain":true,"message":{"role":"assistant","stop_reason":"end_turn","content":[{"type":"text","text":"sub"}]}}"#; + let unknown = r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"text","text":"?"}]}}"#; + assert_eq!( + state(&[asked, calling, sidechain, unknown].join("\n")), + Some(SessionStatus::Running) + ); + + // And nothing at all to go on says nothing, rather than idle. + assert_eq!(state(r#"{"type":"summary","summary":"x"}"#), None); + } } diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 6a5bce6..f13d5e5 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -1016,6 +1016,18 @@ async fn pump( Event::MessageTaken { text } => Event::UserMessage { text }, other => other, }; + // A status the session is already in is not news. Imported + // sessions make this the common case rather than a rarity: each + // sync reads the turn state off the file's newest record, and + // most of them find the same answer as the sync before -- which + // would otherwise be a transcript entry, a broadcast, and a + // recomposition on every phone, several times a minute, to say + // nothing at all. + if let Event::Status { state } = &event + && *shared.status.lock().unwrap() == *state + { + continue; + } match transcript.append(event, ts) { Ok(entry) => { if let Event::Status { state } = &entry.event {