Say when a session is working, and what it was told
Three things a phone could not see, all of them the same shape: the session was doing something and nothing on screen said so. A turn nobody here started never reported itself. `Running` was sent where a message was *sent*, so a session picked up mid-turn, one compacting on its own, or one another agent wrote to sat there reading as idle until it finished. The driver now says it from what it observes -- output that could only come from a turn in flight -- which is the same set of events that already announced a steer, with the ends swapped. An imported session had it worse: nothing but replayed lines ever reaches it, and a status was not among them, so it was permanently whatever it was when it was adopted. Its file does not record a turn ending, but it does record why each assistant message stopped, and `tool_use` versus anything else answers it. A record that says nothing leaves the status alone rather than voting for idle. Messages from other agents were dropped outright: the CLI marks them meta, and this replayed everything except meta. They are now a row of their own, closed by default like a tool call, named for the session that sent it -- not the reader's own bubble, because they did not say it, and a session working on something this phone never asked for is exactly what one of these explains. Measured against a real session file rather than guessed: the peer record carries the sender's name and the message body in `origin`, beside a copy wrapped for the model to read.
This commit is contained in:
iris committed
1 parent
f18639e4b1
commit
404066fa7d
11 files changed
+577
-29
No files matched your search
@@ -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();
|
||||
|
||||
@@ -104,6 +104,19 @@ pub enum Event {
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
about: Option<String>,
|
||||
},
|
||||
/// 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.
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -410,13 +410,28 @@ pub async fn read_tail(transport: &Transport, path: &str) -> Result<String> {
|
||||
/// which reads its own session file.
|
||||
pub fn events_from(text: &str, session_dir: &std::path::Path) -> Vec<Event> {
|
||||
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::<Value>(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<Event> {
|
||||
_ => {}
|
||||
}
|
||||
}
|
||||
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 `<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> {
|
||||
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<super::driver::SessionStatus> {
|
||||
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<Event>, 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<cross-session-message from-name=\"dev-updater-f5\">\nPull before you touch AGENTS.md.\n</cross-session-message>"}}"#;
|
||||
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);
|
||||
}
|
||||
}
|
||||
@@ -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 {
|
||||
|
||||
Reference in new issue
Block a user