diff --git a/AGENTS.md b/AGENTS.md index e9c8a95..bbb65aa 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -245,6 +245,20 @@ first if a remote spawn ever mangles an argument. on the same transcript. What that costs is that the rows below slide up under the reader's finger, so a row that has just moved ignores taps for half a second (`SETTLE_MS`). +- **All transcript text is selectable, from one `SelectionContainer` around + the whole list** (`TranscriptList.kt`). Not per row: a transcript is one + body of text to a reader, so a selection has to be able to run from a + reply into the tool output under it -- and a container per row leaves + whatever was drawn without one silently unselectable, which nothing on + screen reports. Rows keep their tap handlers; selection is a long press. +- **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, + and that is correct rather than broken: the driver writes a steer into the + CLI the moment it arrives, so what the bubble is waiting for is the CLI + *reading* it, not this server sending it. The refusal is drawn on the + bubble. The echo driver really does hold its queue, so that is the rig for + the case where the drop succeeds. - **Deleting a session offers to take the machine's own transcript with it.** `DELETE /sessions/{id}?deleteForeign=true`, behind a switch in the confirmation, and only where the driver keeps a record of its own diff --git a/PLAN.md b/PLAN.md index a0c9ee1..4dd4edf 100644 --- a/PLAN.md +++ b/PLAN.md @@ -247,6 +247,35 @@ turn. Claude's dialect: a `user` message on stdin mid-stream; pi's: `steer`. - Images in: base64 image content blocks in the stream-json user message. - Working directory, host, and model are spawn-screen fields. +### Taking a queued message back (decided 2026-08-31) + +A message sent into a running turn is drawn as a bubble waiting below the +transcript, and tapping it asks the server to drop it before the session +reads it — `POST /sessions/{id}/unqueue {messageId}`, answered by +`Driver::unqueue` and recorded as `Event::MessageDropped` so that every +device watching loses the bubble and a reconnect does not replay it back. + +The answer has **three** states rather than a yes/no, and that is the whole +of the design: `Dropped`, `AlreadySent`, and `Unknown`. The reason is that +the Claude driver can only ever give the middle one. It writes a steer into +the CLI's stdin the instant it arrives — that is what makes a steer reach +the model at the next tool boundary instead of at the end of the turn, and +it was measured (see `Queue`'s doc comment) — so the line is gone before the +phone could ask for it back. What waits in `awaiting` is the *announcement*, +not the message. + +Holding the write until a boundary was considered and rejected on 2026-08-31: +it would make the drop real everywhere, but it costs a steer one model call, +which is the latency the immediate write was introduced to remove. So the +refusal is the honest answer and it is reported where the reader pressed — +on the bubble itself, not in the screen's error row, which is under the +header a screen away. What a tap buys on a Claude session is therefore +knowing that the session has already been told; on a driver that really does +hold a queue (echo today) the message goes. + +`Unknown` is not "we could not find out": a driver that is gone reported +everything it was holding when it closed, so there is nothing waiting. + ### Session processes outlive the backend (decided 2026-08-29) A session's process is **left running when the backend stops, and adopted @@ -659,6 +688,7 @@ GET /sessions list (id, provider, host, title, model, st POST /sessions spawn {provider, host, model, cwd, permission_mode, title} GET /sessions/:id/events?after=N SSE: transcript replay from N, then live POST /sessions/:id/message {text, attachment_ids} +POST /sessions/:id/unqueue {message_id} take back one not read yet POST /sessions/:id/answer {question_id, answer} (questions and permissions) POST /sessions/:id/interrupt stop the running turn; the process stays POST /sessions/:id/stop end the process; the session and transcript stay diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt index 8d21262..eff62ff 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Api.kt @@ -492,6 +492,23 @@ fun sendMessage( ) {} } +/** + * Takes back a message the session has not read yet, named by the id its `messageQueued` carried. + * + * Throws rather than returning an outcome, because both ways of failing are things the reader has + * to be told: 409 means the session was already given it, and 404 means nothing is waiting under + * that id. The bubble disappearing is the success case and it arrives on the event stream, not from + * here -- every device drops it, not only the one that tapped. + */ +fun unqueueMessage(settings: ServerSettings, sessionId: String, messageId: String) { + requestFromServer( + settings, + "/sessions/$sessionId/unqueue", + method = "POST", + jsonBody = JSONObject().put("messageId", messageId).toString(), + ) {} +} + /** Uploads one picked image; the returned id goes into [sendMessage]. */ fun uploadAttachment( settings: ServerSettings, 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 3f612c5..771ba89 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/Events.kt @@ -53,6 +53,16 @@ sealed class SessionEvent { data class MessageQueued(val id: String, val text: String, val images: List) : SessionEvent() + /** + * A queued message taken back before the session read it. + * + * Recorded by the server for the same reason [MessageQueued] is: a phone that reconnects + * replays both, and without this one it would put back a bubble for a message that is never + * coming -- with nothing left to resolve it, since the [UserMessage] that normally does is + * exactly what was cancelled. + */ + data class MessageDropped(val id: String) : SessionEvent() + data class AssistantText(val delta: String) : SessionEvent() data class ToolStart(val id: String, val tool: String, val input: String) : SessionEvent() @@ -183,6 +193,7 @@ fun parseSeqEvent(json: String): SeqEvent { body.getString("text"), body.stringList("images"), ) + "messageDropped" -> SessionEvent.MessageDropped(body.getString("id")) "assistantText" -> SessionEvent.AssistantText(body.getString("delta")) "toolStart" -> SessionEvent.ToolStart( 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 ce673f8..02f4f65 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/SessionScreen.kt @@ -8,6 +8,7 @@ import androidx.activity.compose.rememberLauncherForActivityResult import androidx.activity.result.PickVisualMediaRequest import androidx.activity.result.contract.ActivityResultContracts import androidx.compose.foundation.background +import androidx.compose.foundation.clickable import androidx.compose.foundation.gestures.awaitEachGesture import androidx.compose.foundation.gestures.awaitFirstDown import androidx.compose.foundation.layout.Box @@ -371,6 +372,11 @@ fun SessionScreen(settings: ServerSettings, summary: SessionSummary, onBack: () if (event is SessionEvent.UserMessage) { queued = queued.filterNot { it.id == event.id } } + // Waiting, then taken back. From the server rather than from the tap, so every device + // drops the bubble and a reconnect does not put back one that was cancelled. + if (event is SessionEvent.MessageDropped) { + queued = queued.filterNot { it.id == event.id } + } // Waiting, then gone: a command leaves this list when the session takes it, // and the row it becomes is added by `foldEvent` in the same pass. if (event is SessionEvent.CommandQueued) { @@ -883,6 +889,33 @@ fun SessionScreen(settings: ServerSettings, summary: SessionSummary, onBack: () } } + /** + * Asks the server to take back a message the session has not read yet. + * + * Nothing is removed here. The bubble goes on the `messageDropped` the server records, which is + * what makes the cancellation the session's own fact rather than this screen's opinion of it -- + * a second device watching the same session has to lose the bubble too, and this one has to + * still lose it after a reconnect. + * + * The refusal is kept on the message it was about rather than in [actionError]: the error row + * lives under the header, and a bubble at the foot of the transcript is the thing that was + * pressed. It is the ordinary answer here rather than the exceptional one -- a Claude session + * writes a steer into the CLI the moment it arrives, so what is on screen as "waiting" is + * waiting to be *read*, not waiting to be sent. + */ + fun takeBack(messageId: String) { + scope.launch { + val refusal = + try { + withContext(Dispatchers.IO) { unqueueMessage(settings, summary.id, messageId) } + null + } catch (e: ApiException) { + e.message ?: "this message could not be taken back" + } + queued = queued.map { if (it.id == messageId) it.copy(refusal = refusal) else it } + } + } + fun act(onDone: () -> Unit = {}, action: () -> Unit) { scope.launch { try { @@ -1154,6 +1187,13 @@ fun SessionScreen(settings: ServerSettings, summary: SessionSummary, onBack: () text = waiting.text, images = waiting.images, pending = true, + refusal = waiting.refusal, + // The bubble goes away on the `messageDropped` this + // produces, not here: the server is what knows whether + // the message was still its to take back, and the + // other devices watching this session have to be told + // by the same event. + onTakeBack = { takeBack(waiting.id) }, ) } } @@ -1672,6 +1712,11 @@ private fun ProcessAction.perform(settings: ServerSettings, sessionId: String) = * * [pending] is one the server has taken and the session has not read yet -- drawn quieter, because * "said" and "heard" are different claims and the transcript must not merge them. + * + * A pending bubble is tappable: [onTakeBack] asks the server to drop the message before the session + * reads it, and [refusal] is what came back when it would not. The refusal is drawn here rather + * than with the screen's other errors because this is where the reader pressed -- the error row is + * under the header, a screen away from the bubble they were looking at. */ @Composable private fun UserBubble( @@ -1680,6 +1725,8 @@ private fun UserBubble( text: String, images: List = emptyList(), pending: Boolean = false, + refusal: String? = null, + onTakeBack: (() -> Unit)? = null, ) { Box(Modifier.fillMaxWidth()) { Card( @@ -1693,7 +1740,19 @@ private fun UserBubble( if (pending) MaterialTheme.colorScheme.surfaceVariant else MaterialTheme.colorScheme.primaryContainer ), - modifier = Modifier.align(Alignment.CenterEnd).padding(start = 48.dp), + modifier = + Modifier.align(Alignment.CenterEnd) + .padding(start = 48.dp) + .then( + if (onTakeBack == null) Modifier + else + Modifier.clickable(onClick = onTakeBack).semantics { + // The bubble is its own control and its own label; without this + // the only thing to read is the message, which does not say what + // pressing it does. + contentDescription = "Waiting to be read; tap to take it back" + } + ), ) { Column(Modifier.padding(12.dp)) { // A message can be nothing but an attachment, and an empty line above a picture @@ -1713,13 +1772,32 @@ private fun UserBubble( if (index > 0 || text.isNotEmpty()) Spacer(Modifier.height(4.dp)) SessionImage(settings, sessionId, ref) } + refusal?.let { + Spacer(Modifier.height(6.dp)) + Text( + it, + style = MaterialTheme.typography.bodySmall, + color = MaterialTheme.colorScheme.error, + ) + } } } } } -/** A message the server has accepted and the session has not read yet. */ -private data class QueuedMessage(val id: String, val text: String, val images: List) +/** + * A message the server has accepted and the session has not read yet. + * + * [refusal] is why taking it back did not work, kept per message rather than on the screen: two + * bubbles can be waiting at once, and an error above them both would not say which one it was + * about. + */ +private data class QueuedMessage( + val id: String, + val text: String, + val images: List, + val refusal: String? = null, +) /** * Asked before switching model, because switching is not free and the cost is invisible. 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 54b5b2d..c358417 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptItems.kt @@ -363,6 +363,9 @@ fun foldEvent(items: List, entry: SeqEvent): List items + // The bubble goes away and nothing takes its place: the message was never read, so there + // is nothing it belongs above. + is SessionEvent.MessageDropped -> items is SessionEvent.Settings -> items is SessionEvent.Status -> items is SessionEvent.Error -> items + TranscriptItem.ErrorMsg(entry.seq, event.message) diff --git a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptList.kt b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptList.kt index bb6f460..9fabfe3 100644 --- a/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptList.kt +++ b/app/androidApp/src/main/kotlin/com/example/aiapp/TranscriptList.kt @@ -7,6 +7,7 @@ import androidx.compose.foundation.layout.padding import androidx.compose.foundation.layout.size import androidx.compose.foundation.lazy.LazyColumn import androidx.compose.foundation.lazy.LazyListState +import androidx.compose.foundation.text.selection.SelectionContainer import androidx.compose.material3.CircularProgressIndicator import androidx.compose.runtime.Composable import androidx.compose.ui.Alignment @@ -35,6 +36,14 @@ import androidx.compose.ui.unit.dp * What keeps a unit's arrival cheap enough to happen mid-fling: a unit is at most one block of a * reply, and its parse is already made by [warm] before the fold that introduces it -- so entering * composition costs laying out one paragraph, not parsing a message. + * + * The whole list sits in a [SelectionContainer], which is what makes every word in the transcript + * selectable by the platform's own press-and-hold. Here rather than at each place text is drawn: a + * transcript is one body of text to a reader, and a container per row would mean a selection could + * never cross from a reply into the tool output that follows it -- and would leave whatever was + * drawn without one silently unselectable, which is a state nothing on screen reports. Rows keep + * their tap handlers: selection is a long press, and the container passes an ordinary click through + * to the card under it. */ @Composable fun TranscriptList( @@ -45,51 +54,58 @@ fun TranscriptList( below: @Composable () -> Unit, unit: @Composable (TranscriptUnit) -> Unit, ) { - LazyColumn( - state = state, - reverseLayout = true, - contentPadding = TRANSCRIPT_PADDING, - modifier = - // Timed in two halves because the frame's draw phase is where Compose's measurement - // lands, and "draw is high while nothing is being recorded" does not say which half; - // see [drawAccounting]. Measure includes composing the items that scrolled in. - modifier - .layout { measurable, constraints -> - val started = System.nanoTime() - val placeable = measurable.measure(constraints) - DebugStats.record("measure: the whole transcript", System.nanoTime() - started) - layout(placeable.width, placeable.height) { - val placing = System.nanoTime() - placeable.place(0, 0) + SelectionContainer { + LazyColumn( + state = state, + reverseLayout = true, + contentPadding = TRANSCRIPT_PADDING, + modifier = + // Timed in two halves because the frame's draw phase is where Compose's measurement + // lands, and "draw is high while nothing is being recorded" does not say which + // half; + // see [drawAccounting]. Measure includes composing the items that scrolled in. + modifier + .layout { measurable, constraints -> + val started = System.nanoTime() + val placeable = measurable.measure(constraints) DebugStats.record( - "place: the whole transcript", - System.nanoTime() - placing, + "measure: the whole transcript", + System.nanoTime() - started, + ) + layout(placeable.width, placeable.height) { + val placing = System.nanoTime() + placeable.place(0, 0) + DebugStats.record( + "place: the whole transcript", + System.nanoTime() - placing, + ) + } + } + .drawWithContent { + val started = System.nanoTime() + drawContent() + DebugStats.record("draw: the whole transcript", System.nanoTime() - started) + }, + ) { + // The bottom of the screen: what is waiting to be read sits under the newest message. + item(key = "below", contentType = "below") { below() } + items(count = units.size, key = { units[it].key }, contentType = { units[it]::class }) { + val u = units[it] + DebugStats.count("unit composed") + Box(Modifier.fillMaxWidth().padding(top = u.gap)) { unit(u) } + } + // Standing in for everything not fetched yet. Only here while there is more -- its + // appearance at the top edge is also roughly when the next page is asked for, so what + // it + // reports is a fetch in flight rather than an end reached. + if (moreHistory) { + item(key = "history", contentType = "history") { + Box(Modifier.fillMaxWidth().padding(vertical = 24.dp)) { + CircularProgressIndicator( + Modifier.align(Alignment.Center).size(HISTORY_SPINNER) ) } } - .drawWithContent { - val started = System.nanoTime() - drawContent() - DebugStats.record("draw: the whole transcript", System.nanoTime() - started) - }, - ) { - // The bottom of the screen: what is waiting to be read sits under the newest message. - item(key = "below", contentType = "below") { below() } - items(count = units.size, key = { units[it].key }, contentType = { units[it]::class }) { - val u = units[it] - DebugStats.count("unit composed") - Box(Modifier.fillMaxWidth().padding(top = u.gap)) { unit(u) } - } - // Standing in for everything not fetched yet. Only here while there is more -- its - // appearance at the top edge is also roughly when the next page is asked for, so what it - // reports is a fetch in flight rather than an end reached. - if (moreHistory) { - item(key = "history", contentType = "history") { - Box(Modifier.fillMaxWidth().padding(vertical = 24.dp)) { - CircularProgressIndicator( - Modifier.align(Alignment.Center).size(HISTORY_SPINNER) - ) - } } } } diff --git a/server/src/routes.rs b/server/src/routes.rs index 5205e30..c305e9b 100644 --- a/server/src/routes.rs +++ b/server/src/routes.rs @@ -17,6 +17,8 @@ //! `reset` frame plus the newest window) //! POST /sessions/{id}/message {text, attachmentIds?} //! (starts the process first if it has exited) +//! POST /sessions/{id}/unqueue {messageId} -- take back one not read yet +//! (409 when the session already has it) //! POST /sessions/{id}/answer {questionId, answers} (questions and permissions) //! POST /sessions/{id}/interrupt stop the running turn; the process stays //! POST /sessions/{id}/stop end the process; the session and transcript stay @@ -68,7 +70,7 @@ use tokio::sync::{broadcast, mpsc}; use tokio_stream::StreamExt; use tokio_stream::wrappers::{BroadcastStream, ReceiverStream}; -use crate::session::driver::SessionCommand; +use crate::session::driver::{SessionCommand, Unqueued}; use crate::session::pending::Operation; use crate::session::transcript::{CATCH_UP_LIMIT, CatchUp, SeqEvent, catch_up}; use crate::session::{LiveSession, SessionInfo, SessionManager, SpawnSpec}; @@ -93,6 +95,7 @@ pub fn router(manager: Arc) -> Router { .route("/sessions/{id}/events", get(events)) .route("/sessions/{id}/transcript", get(transcript)) .route("/sessions/{id}/message", post(message)) + .route("/sessions/{id}/unqueue", post(unqueue)) .route("/sessions/{id}/answer", post(answer)) .route("/sessions/{id}/interrupt", post(interrupt)) .route("/sessions/{id}/stop", post(stop)) @@ -123,6 +126,10 @@ enum ApiError { UnknownRoute, #[error("{0}")] BadRequest(String), + /// The request was understood and the state it names has moved on -- + /// distinct from `BadRequest`, which is a caller that got it wrong. + #[error("{0}")] + Conflict(String), #[error(transparent)] Internal(#[from] anyhow::Error), } @@ -132,6 +139,7 @@ impl IntoResponse for ApiError { let status = match self { Self::NotFound(_) | Self::UnknownRoute => StatusCode::NOT_FOUND, Self::BadRequest(_) => StatusCode::BAD_REQUEST, + Self::Conflict(_) => StatusCode::CONFLICT, Self::Internal(err) => { // The only variant whose real cause isn't safe to hand // back verbatim, and the only one worth a log line. @@ -900,6 +908,40 @@ async fn message( Ok(StatusCode::NO_CONTENT) } +#[derive(Deserialize)] +#[serde(rename_all = "camelCase")] +#[serde(deny_unknown_fields)] +struct UnqueueRequest { + /// The id the `messageQueued` event carried, which is what the bubble + /// on screen is drawn from. + message_id: String, +} + +/// Takes back a message the session has not read yet. +/// +/// The two failures are separate answers rather than one refusal, because +/// they are different things to whoever tapped: `409` means the session has +/// already been told and the message is on its way into the conversation, +/// and `404` means nothing is waiting under that id -- a bubble on screen +/// that something else has already resolved. See [`Driver::unqueue`]; the +/// Claude driver can only ever give the first, since it writes a steer into +/// the CLI the moment it arrives. +async fn unqueue( + State(manager): State>, + UrlPath(id): UrlPath, + axum::Json(body): axum::Json, +) -> Result { + match lookup(&manager, &id)?.unqueue(&body.message_id) { + Unqueued::Dropped => Ok(StatusCode::NO_CONTENT), + Unqueued::AlreadySent => Err(ApiError::Conflict( + "the session has already been given this message".to_string(), + )), + Unqueued::Unknown => Err(ApiError::NotFound( + "this message is not waiting to be read".to_string(), + )), + } +} + #[derive(Deserialize)] #[serde(rename_all = "camelCase")] #[serde(deny_unknown_fields)] diff --git a/server/src/session/claude.rs b/server/src/session/claude.rs index 0805307..a036f2a 100644 --- a/server/src/session/claude.rs +++ b/server/src/session/claude.rs @@ -57,7 +57,7 @@ use serde_json::{Value, json}; use tokio::io::AsyncWriteExt; use tokio::sync::mpsc; -use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus}; +use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus, Unqueued}; use super::process; use super::transport::{Launch, Streams, Transport}; use crate::config::{ProviderConfig, SessionConfig}; @@ -595,6 +595,24 @@ impl Driver for ClaudeDriver { self.send_line(line); } + /// Never droppable, and that is a property of the design rather than + /// an omission. + /// + /// A message queued here has already been written to the CLI's stdin + /// -- see [`Queue`], where only the *announcement* waits -- because + /// that is what makes a steer reach the model at the next tool + /// boundary instead of at the end of the turn. A line in the fifo + /// cannot be recalled, so the only honest answers are "the session has + /// already been told" and "nothing is waiting under that id". + fn unqueue(&self, id: &str) -> Unqueued { + let queue = self.queue.lock().unwrap(); + if queue.awaiting.iter().any(|(waiting, ..)| waiting == id) { + Unqueued::AlreadySent + } else { + Unqueued::Unknown + } + } + fn answer_question(&self, id: &str, answers: &[String]) { let response = { let mut state = self.state.lock().unwrap(); diff --git a/server/src/session/driver.rs b/server/src/session/driver.rs index 1eeeb80..59a1ceb 100644 --- a/server/src/session/driver.rs +++ b/server/src/session/driver.rs @@ -114,6 +114,22 @@ pub enum Event { #[serde(default, skip_serializing_if = "Vec::is_empty")] images: Vec, }, + /// A message taken out of the queue before the session read it, by + /// somebody tapping the bubble that was waiting for it. + /// + /// Recorded for the same reason `MessageQueued` is: the queue is the + /// server's, so what is waiting has to be answerable from the + /// transcript alone. Without it a phone that reconnects replays the + /// `MessageQueued` and puts back a bubble for a message that will + /// never arrive -- and nothing later would ever resolve it, since the + /// `UserMessage` that normally does is exactly what is not coming. + /// + /// Only ever sent for a message that had not been handed over. One + /// that has is not droppable and says so instead; see + /// [`Unqueued::AlreadySent`]. + MessageDropped { + id: String, + }, /// A driver has taken one of the user's messages and started reading /// it. The manager turns this into the `UserMessage` above, so it /// never reaches a phone itself. @@ -444,6 +460,25 @@ pub enum SessionStatus { Unknown, } +/// What became of a request to take a queued message back. +/// +/// Three states rather than a bool because the two failures are not the +/// same fact. A driver that writes into its session the moment a message +/// arrives -- which is what `ClaudeDriver` does, so that a steer reaches +/// the model at the next tool boundary rather than at the end of the turn +/// -- can never take one back, and a phone that was told only "no" would +/// have to guess whether it had asked too late or asked about nothing. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum Unqueued { + /// Out of the queue; the session will never read it. + Dropped, + /// Already handed to the session, so there is nothing left to take + /// back. The message is on its way into the conversation. + AlreadySent, + /// Nothing is waiting under that id. + Unknown, +} + /// Where a driver reports events. Unbounded because producers are child /// processes a slow phone must never be able to stall; the transcript file /// is the backpressure-free buffer of record. @@ -463,6 +498,19 @@ pub trait Driver: Send + Sync { /// message in the transcript, so a driver that never sends it drops /// the message from the conversation entirely. fn send_user_message(&self, text: String, images: Vec); + /// Takes back a message that is still waiting, named by the id its + /// [`Event::MessageQueued`] carried. + /// + /// Answering is the whole of the contract: a driver that drops the + /// message owes an [`Event::MessageDropped`], and one that cannot must + /// say which of the two reasons it is, because they are different + /// things to a reader -- "the session has already been told" is worth + /// knowing, and "there is nothing under that id" means the bubble on + /// screen is stale. The default is the honest answer for a driver with + /// no queue at all: nothing of yours is waiting. + fn unqueue(&self, _id: &str) -> Unqueued { + Unqueued::Unknown + } /// Answers one question with everything that was chosen, in the order /// it was offered. One answer is a list of one; a driver whose dialect /// takes a single value joins them where it writes it. diff --git a/server/src/session/echo.rs b/server/src/session/echo.rs index 2f5beeb..6e57e8f 100644 --- a/server/src/session/echo.rs +++ b/server/src/session/echo.rs @@ -53,7 +53,7 @@ use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex}; use std::time::Duration; -use super::driver::{Driver, Event, EventSink, ImageRef, QuestionOption, SessionStatus}; +use super::driver::{Driver, Event, EventSink, ImageRef, QuestionOption, SessionStatus, Unqueued}; /// Delay between streamed deltas -- long enough that streaming is visibly /// streaming in the UI, short enough that tests waiting on a full turn @@ -796,6 +796,23 @@ impl Driver for EchoDriver { !self.busy.load(Ordering::SeqCst) } + /// Really droppable, which is what makes this the rig for the phone's + /// side of it: the held message is this driver's own and nothing has + /// been written anywhere, so a tap here exercises the whole path + /// through to the bubble disappearing on every device. The Claude + /// driver can only ever refuse -- see its own `unqueue` -- so it + /// cannot exercise the case where the drop succeeds. + fn unqueue(&self, id: &str) -> Unqueued { + let mut queued = self.queued.lock().unwrap(); + let Some(at) = queued.iter().position(|(waiting, ..)| waiting == id) else { + return Unqueued::Unknown; + }; + queued.remove(at); + drop(queued); + self.emit(Event::MessageDropped { id: id.to_string() }); + Unqueued::Dropped + } + fn send_user_message(&self, text: String, images: Vec) { // Announced, because this is a message: every driver owes exactly // one `MessageTaken` per message, and one that quietly vanishes diff --git a/server/src/session/mod.rs b/server/src/session/mod.rs index 4fbac3b..cda3e89 100644 --- a/server/src/session/mod.rs +++ b/server/src/session/mod.rs @@ -32,7 +32,9 @@ use crate::config::{ Config, DriverKind, ProviderConfig, SessionConfig, SetupConfig, SshConfig, TokenEntry, }; use claude::ClaudeDriver; -use driver::{Driver, Event, EventSink, ImageRef, SessionCommand, SessionStatus, context_after}; +use driver::{ + Driver, Event, EventSink, ImageRef, SessionCommand, SessionStatus, Unqueued, context_after, +}; use echo::EchoDriver; use llama::LlamaDriver; use transcript::{SeqEvent, Transcript}; @@ -440,6 +442,21 @@ impl LiveSession { self.ask("be interrupted", |driver| driver.interrupt()); } + /// Takes back a message the session has not read yet, named by the id + /// its `MessageQueued` carried. See [`Driver::unqueue`] for why the + /// answer has three states. + /// + /// A session with no process answers `Unknown` rather than being + /// reported as a failure, and that is the true answer: a driver on its + /// way out already said what it was holding (`Queue::close`), so there + /// is nothing waiting to take back. + pub fn unqueue(&self, message_id: &str) -> Unqueued { + match self.driver() { + Some(driver) => driver.unqueue(message_id), + None => Unqueued::Unknown, + } + } + /// Leaves this session's process running and stops attending to it, /// for a server that is going away and means to come back. See /// [`Driver::detach`]. @@ -2716,6 +2733,86 @@ mod tests { ); } + /// A queued message can be taken back until the driver has handed it + /// over, and the taking back is an event rather than a return value -- + /// which is what makes the bubble disappear on every device watching, + /// and stay gone when one of them reconnects and replays. + /// + /// Exercised on echo because echo really holds its queue. The Claude + /// driver writes a steer into the CLI the moment it arrives, so it can + /// only ever answer `AlreadySent`; the case where a drop *succeeds* + /// has no other driver to be tested against. + #[tokio::test] + async fn a_queued_message_can_be_taken_back_until_the_session_has_it() { + let dir = tempfile::tempdir().expect("tempdir"); + seed_echo_only(&dir.path().join("config.ron")); + let manager = SessionManager::new( + dir.path().join("config.ron"), + dir.path().join("sessions"), + dir.path().join("models"), + ) + .expect("manager"); + let info = manager.spawn_session(echo_spec()).expect("spawn"); + let session = manager.session(&info.id).expect("live session"); + let mut rx = session.subscribe(); + + // A turn long enough that the next message has to wait behind it. + session.send_message("/slow 1".to_string(), Vec::new()); + collect_until(&mut rx, |event| { + matches!( + event, + Event::Status { + state: SessionStatus::Running + } + ) + }) + .await; + + session.send_message("second thoughts".to_string(), Vec::new()); + let seen = collect_until(&mut rx, |event| { + matches!(event, Event::MessageQueued { .. }) + }) + .await; + let Some(Event::MessageQueued { id, .. }) = + seen.iter().map(|entry| entry.event.clone()).next_back() + else { + panic!("expected the message to be queued: {seen:?}"); + }; + + assert_eq!(session.unqueue(&id), Unqueued::Dropped); + let seen = collect_until(&mut rx, |event| { + matches!(event, Event::MessageDropped { .. }) + }) + .await; + assert!( + seen.iter().any(|entry| matches!( + &entry.event, + Event::MessageDropped { id: dropped } if *dropped == id + )), + "the drop has to be recorded, not merely returned: {seen:?}" + ); + + // Gone for good: the turn ends without the message ever entering + // the conversation, and asking again says there is nothing there + // rather than dropping it twice. + assert_eq!(session.unqueue(&id), Unqueued::Unknown); + let seen = collect_until(&mut rx, |event| { + matches!( + event, + Event::Status { + state: SessionStatus::Idle + } + ) + }) + .await; + assert!( + !seen + .iter() + .any(|entry| matches!(entry.event, Event::UserMessage { .. })), + "a message taken back must never be read: {seen:?}" + ); + } + #[tokio::test] async fn a_command_on_an_idle_session_goes_straight_out() { let dir = tempfile::tempdir().expect("tempdir");