Select any of the transcript, and take a queued message back
Two things a reader could not do to what is on screen.
**Selection.** Nothing in the transcript was selectable at all, so a
command, a path or an error message could be read and not copied. One
`SelectionContainer` around the whole list rather than one per row: a
transcript is one body of text to a reader, and a selection has to be able
to run from a reply into the tool output under it. Per row it also could
not, and whatever was drawn without a container would have been silently
unselectable -- a state nothing on screen reports. Rows keep their tap
handlers; checked on the emulator that expanding a tool call, scrolling and
flinging are all unaffected, since a selection is a long press.
**Taking a message back.** A message sent into a running turn sits as a
bubble waiting to be read, and there was no way to change your mind: it is
tappable now, and the server answers `POST /sessions/{id}/unqueue`.
The answer has three states, and the middle one is the point. Claude's
driver writes a steer into the CLI's stdin the instant it arrives -- that
is what makes it reach the model at the next tool boundary rather than at
the end of the turn, and it was measured -- so the line is already gone and
`AlreadySent` is the only honest answer it can give. Holding the write
until a boundary would make the drop real and cost a steer one model call,
which is the latency the immediate write exists to remove; rejected on that
trade, with the reasoning in PLAN.md. The refusal is drawn on the bubble
that was pressed rather than in the error row under the header, a screen
away from it.
Where a driver really does hold its queue -- echo today -- the message goes
for good, and it goes as an `Event::MessageDropped` rather than as a return
value: every device watching the session loses the bubble, and a phone that
reconnects and replays the `messageQueued` does not put back one that was
cancelled with nothing left to resolve it.
This commit is contained in:
1 parent
778b2e3b04
commit
82401cd887
12 files changed
+439
-48
No files matched your search
@@ -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();
|
||||
|
||||
@@ -114,6 +114,22 @@ pub enum Event {
|
||||
#[serde(default, skip_serializing_if = "Vec::is_empty")]
|
||||
images: Vec<ImageRef>,
|
||||
},
|
||||
/// 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<ImageRef>);
|
||||
/// 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.
|
||||
|
||||
@@ -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<ImageRef>) {
|
||||
// Announced, because this is a message: every driver owes exactly
|
||||
// one `MessageTaken` per message, and one that quietly vanishes
|
||||
|
||||
@@ -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");
|
||||
|
||||
Reference in new issue
Block a user