Merge remote-tracking branch 'origin/main'
This commit is contained in:
iris committed
commit
7a1a462caf
19 files changed
+1388
-160
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};
|
||||
@@ -150,10 +150,22 @@ impl Queue {
|
||||
/// Reported rather than dropped. These are messages somebody typed
|
||||
/// that never reached the session and never reached the transcript, so
|
||||
/// this is the only place they can be mentioned at all.
|
||||
///
|
||||
/// Each one is also *resolved*, with the same `MessageDropped` that a
|
||||
/// phone tapping the bubble produces. Without it the bubble sat there
|
||||
/// for good: a message drawn as waiting to be read, by a session that
|
||||
/// no longer exists, with the only thing that ever clears it -- the
|
||||
/// `UserMessage` -- exactly what is not coming. The error says what
|
||||
/// happened and the drop is what ends it, which is the same division
|
||||
/// of labour as everywhere else here.
|
||||
fn close(&mut self, sink: &EventSink, why: &str) {
|
||||
self.closed = true;
|
||||
self.running = false;
|
||||
let lost: Vec<String> = self.awaiting.drain(..).map(|(_, text, _)| text).collect();
|
||||
let lost: Vec<(String, String)> = self
|
||||
.awaiting
|
||||
.drain(..)
|
||||
.map(|(id, text, _)| (id, text))
|
||||
.collect();
|
||||
if lost.is_empty() {
|
||||
return;
|
||||
}
|
||||
@@ -165,9 +177,15 @@ impl Queue {
|
||||
} else {
|
||||
format!("{} queued messages", lost.len())
|
||||
},
|
||||
lost.join(" / ")
|
||||
lost.iter()
|
||||
.map(|(_, text)| text.as_str())
|
||||
.collect::<Vec<_>>()
|
||||
.join(" / ")
|
||||
),
|
||||
});
|
||||
for (id, _) in lost {
|
||||
let _ = sink.send(Event::MessageDropped { id });
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -595,6 +613,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();
|
||||
|
||||
@@ -221,6 +221,33 @@ impl Translator {
|
||||
.and_then(Value::as_u64)
|
||||
.unwrap_or(0);
|
||||
let mut events = Vec::new();
|
||||
// A turn another agent started, which is only knowable here.
|
||||
//
|
||||
// Measured against CLI 2.1.237 (2026-08-31) by sending a
|
||||
// real cross-session message to a real stream-json session:
|
||||
// the CLI emits no `user` record for it, and nothing in the
|
||||
// partial-message stream mentions it either. The whole of
|
||||
// it arrives as an `origin` object on the turn's `result`,
|
||||
// in the same shape the session file records -- so this is
|
||||
// `import::peer_message` reading a different record.
|
||||
//
|
||||
// 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.
|
||||
// Taken deliberately over the alternative, which 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. What
|
||||
// it buys is the thing that was missing entirely: a session
|
||||
// that starts working on something nobody on this phone
|
||||
// asked for is otherwise unexplainable from the phone.
|
||||
//
|
||||
// Only peer-caused turns carry it: measured over a real
|
||||
// session's stdout, four ordinary results and no `origin`
|
||||
// between them.
|
||||
if let Some(peer) = crate::session::import::peer_message(message) {
|
||||
events.push(peer);
|
||||
}
|
||||
// Whichever way this result went, the interrupt it may have
|
||||
// been answering is now spent.
|
||||
let asked_to_stop = std::mem::take(&mut self.interrupting);
|
||||
@@ -1089,6 +1116,70 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// A turn another agent started says so, on the record that carries it.
|
||||
///
|
||||
/// The line is the real shape, taken from a real cross-session message
|
||||
/// sent to a real stream-json session on CLI 2.1.237 (2026-08-31) --
|
||||
/// including the `from` socket path, which is deliberately *not* what a
|
||||
/// reader is shown: the sending session's `name` is what they recognise
|
||||
/// it by. The `body` is the message as it was written; the content the
|
||||
/// model is given beside it wraps the same text in a preamble and a
|
||||
/// `<cross-session-message>` tag, which is written for the model rather
|
||||
/// than for a person.
|
||||
///
|
||||
/// The note comes before the usage and the idle, so it sits as close to
|
||||
/// the turn it explains as the wire allows -- which is after the reply,
|
||||
/// not above it. See the comment at the callsite for why that is the
|
||||
/// best available position rather than an oversight.
|
||||
#[test]
|
||||
fn a_turn_started_by_another_agent_records_who_and_what() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let mut translator = Translator::new(dir.path().to_path_buf());
|
||||
let events = translate_lines(
|
||||
&mut translator,
|
||||
&[
|
||||
r#"{"type":"result","subtype":"success","is_error":false,"session_id":"s","usage":{"input_tokens":2,"output_tokens":5},"origin":{"kind":"peer","from":"uds:/run/user/1000/cc-socks/137108.sock","verifiedPeerPid":137108,"msg_id":"1e729740","name":"ai-app-2-fb","fromMode":"prompting","body":"Reply with just the word ACK."}}"#,
|
||||
],
|
||||
);
|
||||
assert_eq!(
|
||||
events,
|
||||
vec![
|
||||
Event::PeerMessage {
|
||||
from: "ai-app-2-fb".to_string(),
|
||||
text: "Reply with just the word ACK.".to_string(),
|
||||
},
|
||||
Event::UsageDelta {
|
||||
tokens: 7,
|
||||
context: None
|
||||
},
|
||||
Event::Status {
|
||||
state: SessionStatus::Idle
|
||||
},
|
||||
]
|
||||
);
|
||||
}
|
||||
|
||||
/// And an ordinary turn does not, which is the half that decides
|
||||
/// whether the check above is a check or a rubber stamp. Measured over
|
||||
/// a real session's stdout: four results, no `origin` between them.
|
||||
#[test]
|
||||
fn an_ordinary_turn_carries_no_peer_note() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let mut translator = Translator::new(dir.path().to_path_buf());
|
||||
let events = translate_lines(
|
||||
&mut translator,
|
||||
&[
|
||||
r#"{"type":"result","subtype":"success","is_error":false,"session_id":"s","usage":{"input_tokens":2,"output_tokens":5}}"#,
|
||||
],
|
||||
);
|
||||
assert!(
|
||||
!events
|
||||
.iter()
|
||||
.any(|event| matches!(event, Event::PeerMessage { .. })),
|
||||
"a turn nobody else started must not be attributed to anyone: {events:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// The context is the last assistant message's, not the result's.
|
||||
///
|
||||
/// Real figures from a two-message haiku turn on 2.1.237, captured
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -11,7 +11,9 @@
|
||||
//! looks like when a screen groups them. `gap` is seconds between one
|
||||
//! call and the next, default none: it is what makes a run *grow* while
|
||||
//! somebody is looking at it, which is the only way to reach the state
|
||||
//! where a call opened on its own gains a neighbour.
|
||||
//! where a call opened on its own gains a neighbour. The first call
|
||||
//! carries a screenshot, so that state can also be reached with an image
|
||||
//! open full screen -- which is where it used to close itself.
|
||||
//! - `/question [text]` -- a question, exercising the answer path.
|
||||
//! - `/ask` -- an AskUserQuestion call: two questions on one tool call,
|
||||
//! with descriptions, a preview and a multi-select, which is the shape
|
||||
@@ -53,7 +55,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
|
||||
@@ -482,6 +484,27 @@ impl EchoDriver {
|
||||
"timeout": 5000,
|
||||
}),
|
||||
});
|
||||
// The first call carries a screenshot, and only the
|
||||
// first. That is what makes this rig cover the case a
|
||||
// growing run is actually about: an image opened full
|
||||
// screen from a call that is alone, and then a second
|
||||
// call arriving and turning that row into a group. The
|
||||
// dialog used to be inside the row, so the reader was
|
||||
// thrown back to the transcript by the session making
|
||||
// another tool call. Any of the calls would do; the
|
||||
// first is the one that is on its own for a whole
|
||||
// `gap`, which is the window somebody can open it in.
|
||||
if i == 1 {
|
||||
let part = serde_json::json!({
|
||||
"source": {"media_type": "image/png", "data": SAMPLE_PNG}
|
||||
});
|
||||
if let Some(name) = super::claude::translate::save_image(&dir, &part) {
|
||||
send(Event::Image {
|
||||
image: name,
|
||||
about: Some(id.clone()),
|
||||
});
|
||||
}
|
||||
}
|
||||
tokio::time::sleep(DELTA_DELAY).await;
|
||||
send(Event::ToolEnd {
|
||||
id,
|
||||
@@ -796,6 +819,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
|
||||
|
||||
@@ -520,7 +520,7 @@ pub fn events_from(text: &str, session_dir: &std::path::Path) -> Vec<Event> {
|
||||
events
|
||||
}
|
||||
|
||||
/// A message from another agent, as the CLI records one.
|
||||
/// A message from another agent, as the CLI reports 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
|
||||
@@ -529,7 +529,14 @@ pub fn events_from(text: &str, session_dir: &std::path::Path) -> Vec<Event> {
|
||||
/// preamble and a `<cross-session-message>` 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<Event> {
|
||||
///
|
||||
/// Shared with the live driver (`claude::translate`), which finds the same
|
||||
/// `origin` object on a different record -- so this reads the object and
|
||||
/// not the record around it. One function because it is one wire format:
|
||||
/// two copies would drift the first time the CLI renames a field, and the
|
||||
/// half that drifted would go on producing nothing at all, which is
|
||||
/// indistinguishable from nobody having sent anything.
|
||||
pub(in crate::session) fn peer_message(record: &Value) -> Option<Event> {
|
||||
let origin = record.get("origin")?;
|
||||
if origin.get("kind").and_then(Value::as_str) != Some("peer") {
|
||||
return None;
|
||||
|
||||
+280
-21
@@ -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`].
|
||||
@@ -483,15 +500,31 @@ impl LiveSession {
|
||||
Ok(name)
|
||||
}
|
||||
|
||||
/// `setup_name` is passed in rather than stored: only the manager
|
||||
/// holds the config, and the label can change under a running session.
|
||||
/// `setup_name` and `cwd` are passed in rather than read from the
|
||||
/// snapshot this session launched with: only the manager holds the
|
||||
/// config, and both of them can change under a running session. The
|
||||
/// label changes when a setup is renamed; the directory changes when
|
||||
/// somebody moves the session, and reading the snapshot reported the
|
||||
/// old one for as long as the process lived -- a screen showing a
|
||||
/// directory the next launch will not use, with nothing saying so.
|
||||
///
|
||||
/// Passed rather than mirrored into `Shared`, which is where `title`
|
||||
/// and `notify` live: a second copy is a second thing to keep level,
|
||||
/// and this way there is one answer, read where the row is built.
|
||||
///
|
||||
/// `kind` rather than the facts derived from it: two of this row's
|
||||
/// fields are answers about the provider's *kind*, and passing them
|
||||
/// separately meant every caller deriving each one and a third arriving
|
||||
/// as a third parameter. `None` where the provider has been edited away,
|
||||
/// which is a session that cannot run -- so both answers are the
|
||||
/// cautious one rather than a guess.
|
||||
fn info(&self, setup_name: &str, imported: bool, kind: Option<DriverKind>) -> SessionInfo {
|
||||
fn info(
|
||||
&self,
|
||||
setup_name: &str,
|
||||
cwd: Option<&Path>,
|
||||
imported: bool,
|
||||
kind: Option<DriverKind>,
|
||||
) -> SessionInfo {
|
||||
SessionInfo {
|
||||
id: self.meta.id.clone(),
|
||||
provider: self.meta.provider.clone(),
|
||||
@@ -505,7 +538,7 @@ impl LiveSession {
|
||||
max_image_edge: kind.and_then(DriverKind::max_image_edge),
|
||||
imported,
|
||||
keeps_own_transcript: kind.is_some_and(DriverKind::keeps_own_transcript),
|
||||
cwd: self.meta.cwd.clone(),
|
||||
cwd: cwd.map(Path::to_path_buf),
|
||||
status: *self.shared.status.lock().unwrap(),
|
||||
last_activity: *self.shared.last_activity.lock().unwrap(),
|
||||
created: self.meta.created,
|
||||
@@ -945,6 +978,7 @@ impl SessionManager {
|
||||
.map(|meta| match inner.live.get(&meta.id) {
|
||||
Some(session) => session.info(
|
||||
label_of(&inner.config, &meta.setup),
|
||||
meta.cwd.as_deref(),
|
||||
import::read_cursor(&self.data_dir.join(&meta.id)).is_some(),
|
||||
kind_of(&inner.config, &meta.setup, &meta.provider),
|
||||
),
|
||||
@@ -1106,6 +1140,7 @@ impl SessionManager {
|
||||
// listing asks of the directory a moment later.
|
||||
let info = session.info(
|
||||
&setup.name,
|
||||
session.meta.cwd.as_deref(),
|
||||
import::read_cursor(&self.data_dir.join(&id)).is_some(),
|
||||
Some(provider.kind),
|
||||
);
|
||||
@@ -1268,6 +1303,60 @@ impl SessionManager {
|
||||
/// for every provider that has a process at all -- so asking it here
|
||||
/// stops a session whose driver is in no state to be asked, and adds no
|
||||
/// method a new driver could implement wrongly.
|
||||
/// Moves a session to a different working directory.
|
||||
///
|
||||
/// The directory is settled at spawn -- the CLI is launched with it as
|
||||
/// its cwd and there is no control request that changes one -- so this
|
||||
/// records the new one and ends the process that is in the old one. It
|
||||
/// does **not** start a replacement: a session with no process starts
|
||||
/// on the next thing said to it, or on Start, which is this app's one
|
||||
/// rule for that everywhere else. Starting one here would have to wait
|
||||
/// for the recorded status to catch up with a process that is already
|
||||
/// gone, and "usually restarts" is a worse control than "always stops".
|
||||
///
|
||||
/// Nothing of Claude Code's own is moved, and that is a measurement
|
||||
/// rather than an omission: `claude --resume <id>` finds a session from
|
||||
/// any working directory (checked against 2.1.237 on 2026-08-31 -- an
|
||||
/// id that does not exist says "No conversation found with session ID"
|
||||
/// and a real one resumed from an unrelated directory did not), so the
|
||||
/// conversation continues in the new place with nothing relocated. The
|
||||
/// file stays under the project directory the CLI made for it, which is
|
||||
/// where the CLI itself looks. Reimplementing that directory's name to
|
||||
/// move it would mean reproducing a rule this app cannot see the whole
|
||||
/// of -- the CLI truncates at 200 characters and appends a hash of its
|
||||
/// own, and an override can replace the name entirely -- to relocate a
|
||||
/// file the CLI is still writing.
|
||||
///
|
||||
/// Whether the directory exists is the caller's question, because
|
||||
/// asking it is an ssh round trip on a remote setup; see the route.
|
||||
pub fn set_session_cwd(&self, id: &str, cwd: PathBuf) -> Result<()> {
|
||||
{
|
||||
let mut inner = self.inner.write().unwrap();
|
||||
if !inner.config.sessions.iter().any(|meta| meta.id == id) {
|
||||
bail!("no session {id}");
|
||||
}
|
||||
let mut candidate = inner.config.clone();
|
||||
for meta in candidate.sessions.iter_mut().filter(|meta| meta.id == id) {
|
||||
meta.cwd = Some(cwd.clone());
|
||||
}
|
||||
candidate.save(&self.config_path)?;
|
||||
inner.config = candidate;
|
||||
}
|
||||
// Saved first, so a process that cannot be stopped leaves a session
|
||||
// that will start in the right place rather than one recorded in a
|
||||
// directory nothing agrees with.
|
||||
let dir = self.data_dir.join(id);
|
||||
if let Some(record) = process::live(&dir) {
|
||||
tracing::info!(
|
||||
"moving session {id} to {} -- stopping pid {}",
|
||||
cwd.display(),
|
||||
record.pid
|
||||
);
|
||||
process::stop(&record, process::STOP_GRACE);
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub fn stop_session(&self, id: &str) -> Result<()> {
|
||||
if !self
|
||||
.inner
|
||||
@@ -2069,10 +2158,26 @@ fn is_news(event: &Event, shared: &Shared) -> bool {
|
||||
/// adopted at startup, or because a driver announced itself, is not news
|
||||
/// that anything ended, and sending it would put "finished" on the phone for
|
||||
/// every session in the config every time the backend restarts.
|
||||
fn notification_for(was: SessionStatus, now: SessionStatus) -> Option<NotificationKind> {
|
||||
///
|
||||
/// `unread` is how many messages the session has been handed and not yet
|
||||
/// started reading, and it suppresses *Finished* for the same reason: with
|
||||
/// one waiting, the turn ending is not the work ending. A message written
|
||||
/// into the tail of a turn is read as soon as that turn's `result` lands, so
|
||||
/// the session goes idle and immediately runs again -- and the phone that
|
||||
/// sent it was told its work had finished, seconds before anything of it had
|
||||
/// been done. It cannot suppress *AwaitingInput*: a question is worth saying
|
||||
/// whatever else is queued behind it, and the queue is precisely what will
|
||||
/// not move until it is answered.
|
||||
fn notification_for(
|
||||
was: SessionStatus,
|
||||
now: SessionStatus,
|
||||
unread: usize,
|
||||
) -> Option<NotificationKind> {
|
||||
match (was, now) {
|
||||
(_, SessionStatus::AwaitingInput) => Some(NotificationKind::AwaitingInput),
|
||||
(SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle) => {
|
||||
(SessionStatus::Running | SessionStatus::Compacting, SessionStatus::Idle)
|
||||
if unread == 0 =>
|
||||
{
|
||||
Some(NotificationKind::Finished)
|
||||
}
|
||||
_ => None,
|
||||
@@ -2088,6 +2193,13 @@ async fn pump(
|
||||
commands: Arc<Commands>,
|
||||
notifications: broadcast::Sender<Notification>,
|
||||
) {
|
||||
// Messages the session has been given and not started reading, which is
|
||||
// what makes a turn ending not the same thing as the work ending; see
|
||||
// `notification_for`. Counted from the recorded events rather than asked
|
||||
// of the driver, because this is the one place that sees every event in
|
||||
// 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;
|
||||
while let Some(event) = source.recv().await {
|
||||
let ts = now();
|
||||
// Taking a message is how it enters the conversation, and the
|
||||
@@ -2137,8 +2249,8 @@ async fn pump(
|
||||
// Read before it is overwritten: what makes a status
|
||||
// worth announcing is the transition, not the value.
|
||||
let was = std::mem::replace(&mut *shared.status.lock().unwrap(), *state);
|
||||
if let Some(kind) =
|
||||
notification_for(was, *state).filter(|_| *shared.notify.lock().unwrap())
|
||||
if let Some(kind) = notification_for(was, *state, unread)
|
||||
.filter(|_| *shared.notify.lock().unwrap())
|
||||
{
|
||||
// No subscribers is the ordinary case -- nobody has
|
||||
// the app open -- and it is not an error.
|
||||
@@ -2163,6 +2275,13 @@ async fn pump(
|
||||
Event::Status {
|
||||
state: SessionStatus::Exited,
|
||||
} => commands.abandon("this session's process has exited"),
|
||||
// The two ends of a message's wait. A `UserMessage` with
|
||||
// no id never waited -- it is one sent between turns, and
|
||||
// counting it would take the total below zero.
|
||||
Event::MessageQueued { .. } => unread += 1,
|
||||
Event::UserMessage { id: Some(_), .. } | Event::MessageDropped { .. } => {
|
||||
unread = unread.saturating_sub(1)
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
// No subscribers is fine; the transcript already has it.
|
||||
@@ -2383,28 +2502,43 @@ mod tests {
|
||||
// Waiting on a person is worth saying however it was reached: it
|
||||
// will sit unanswered until somebody is told.
|
||||
assert_eq!(
|
||||
notification_for(Running, SessionStatus::AwaitingInput),
|
||||
notification_for(Running, SessionStatus::AwaitingInput, 0),
|
||||
Some(AwaitingInput)
|
||||
);
|
||||
assert_eq!(
|
||||
notification_for(Idle, SessionStatus::AwaitingInput),
|
||||
notification_for(Idle, SessionStatus::AwaitingInput, 0),
|
||||
Some(AwaitingInput)
|
||||
);
|
||||
|
||||
// A turn this server watched run, ending.
|
||||
assert_eq!(notification_for(Running, Idle), Some(Finished));
|
||||
assert_eq!(notification_for(Compacting, Idle), Some(Finished));
|
||||
assert_eq!(notification_for(Running, Idle, 0), Some(Finished));
|
||||
assert_eq!(notification_for(Compacting, Idle, 0), Some(Finished));
|
||||
|
||||
// Idle arrived at from anywhere else is not an ending.
|
||||
assert_eq!(notification_for(Idle, Idle), None);
|
||||
assert_eq!(notification_for(Unknown, Idle), None);
|
||||
assert_eq!(notification_for(Exited, Idle), None);
|
||||
assert_eq!(notification_for(SessionStatus::AwaitingInput, Idle), None);
|
||||
assert_eq!(notification_for(Idle, Idle, 0), None);
|
||||
assert_eq!(notification_for(Unknown, Idle, 0), None);
|
||||
assert_eq!(notification_for(Exited, Idle, 0), None);
|
||||
assert_eq!(
|
||||
notification_for(SessionStatus::AwaitingInput, Idle, 0),
|
||||
None
|
||||
);
|
||||
|
||||
// Everything else a session does is progress nobody asked to hear.
|
||||
assert_eq!(notification_for(Idle, Running), None);
|
||||
assert_eq!(notification_for(Running, Compacting), None);
|
||||
assert_eq!(notification_for(Running, Exited), None);
|
||||
assert_eq!(notification_for(Idle, Running, 0), None);
|
||||
assert_eq!(notification_for(Running, Compacting, 0), None);
|
||||
assert_eq!(notification_for(Running, Exited, 0), None);
|
||||
|
||||
// A turn ending with a message the session has not started reading
|
||||
// is not the work ending: it goes straight back to running, and
|
||||
// "finished" would arrive seconds before any of that work was done.
|
||||
assert_eq!(notification_for(Running, Idle, 1), None);
|
||||
assert_eq!(notification_for(Compacting, Idle, 2), None);
|
||||
// A question is still worth saying with a queue behind it -- the
|
||||
// queue is exactly what will not move until it is answered.
|
||||
assert_eq!(
|
||||
notification_for(Running, SessionStatus::AwaitingInput, 1),
|
||||
Some(AwaitingInput)
|
||||
);
|
||||
}
|
||||
|
||||
/// The switch reaches the running pump, not just the config file.
|
||||
@@ -2435,7 +2569,7 @@ mod tests {
|
||||
assert_eq!(first.session_id, info.id);
|
||||
// The title travels with it, because the phone may have no screen
|
||||
// open to look one up on.
|
||||
assert_eq!(first.title, session.info("m", false, None).title);
|
||||
assert_eq!(first.title, session.info("m", None, false, None).title);
|
||||
|
||||
manager.set_session_notify(&info.id, false).expect("off");
|
||||
// Subscribed before the message, or the turn can finish in the gap
|
||||
@@ -2460,6 +2594,51 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
/// Counting the wait, rather than only deciding what to do about it.
|
||||
///
|
||||
/// `notification_for` is tested above on the number; this is the number
|
||||
/// itself, which is kept in `pump` from the recorded events and has no
|
||||
/// other way to be looked at. Echo takes its queued message *before*
|
||||
/// going idle -- the same order a real CLI has when the steer lands
|
||||
/// inside the turn -- so the count is back to zero by the end and the
|
||||
/// finish is still announced. That is the case a suppression written
|
||||
/// slightly wrong silences, and it is the common one.
|
||||
#[tokio::test]
|
||||
async fn a_turn_that_read_its_queued_message_still_announces_its_finish() {
|
||||
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 events = session.subscribe();
|
||||
let mut notifications = manager.subscribe_notifications();
|
||||
|
||||
session.send_message("/slow 1".to_string(), Vec::new());
|
||||
collect_until(&mut events, |event| {
|
||||
matches!(
|
||||
event,
|
||||
Event::Status {
|
||||
state: SessionStatus::Running
|
||||
}
|
||||
)
|
||||
})
|
||||
.await;
|
||||
session.send_message("and this behind it".to_string(), Vec::new());
|
||||
collect_until(&mut events, |event| {
|
||||
matches!(event, Event::MessageQueued { .. })
|
||||
})
|
||||
.await;
|
||||
|
||||
let announced = tokio::time::timeout(Duration::from_secs(5), notifications.recv())
|
||||
.await
|
||||
.expect("a notification within five seconds")
|
||||
.expect("channel open");
|
||||
assert_eq!(announced.kind, NotificationKind::Finished);
|
||||
}
|
||||
|
||||
/// A session this app *spawned* is one it is driving, and used to look
|
||||
/// like somebody else's.
|
||||
///
|
||||
@@ -2716,6 +2895,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