iris
749b2db287
Typing "/" now suggests what this app understands -- `/compact` and `/rename <name>` -- with a line each about what they do, and anything else beginning with a slash is passed to whatever runs the session, because a dialect's own vocabulary grows without this list. None of them are messages, and that is the substance of the change. A line written into a running turn is read by the *model*, so a command sent mid-turn either does nothing or arrives as text somebody has to puzzle over. They now wait for the turn to end. The waiting is done once for every provider, in the pump that already watches every event for the boundary, rather than in each driver where a new provider could get it wrong by leaving it out. Waiting is a state, so it is on screen: the command sits at the reader's end of the conversation in blue, with a spinner and "waiting for this turn to end", and becomes an ordinary blue row when it goes. Blue because these are about the session rather than about the task -- the same blue a compaction already used, which is now one colour with one name rather than two. Renaming from the settings screen sends exactly this, so it waits and draws the same way. The name itself is not held: it is this server's own datum, so the list and the header change at once and only telling the session waits. Echo grew the same split, which is where the bug in it showed: its commands are its messages, so running one announced a `MessageTaken` as well, and the same line drew twice -- once blue, once purple. A command owes no announcement; the manager has already recorded that it was sent. Watched rather than reasoned about: `/compact` during a 25 second turn held with its bubble up, went out when the turn ended, and the compaction that followed reported what it recovered.
573 lines
23 KiB
Rust
573 lines
23 KiB
Rust
//! The phase-1 fake driver: no child process, just events. It exists to
|
|
//! prove the whole pipe -- spawn, transcript, SSE cursors, questions,
|
|
//! interrupts, compaction -- before any AI is involved, and stays useful afterwards as
|
|
//! a connectivity check that costs no tokens.
|
|
//!
|
|
//! Behavior: every message is echoed back as a few streamed text deltas.
|
|
//! A leading word asks for something more specific:
|
|
//!
|
|
//! - `/tool [input]` -- a full tool run, start through end.
|
|
//! - `/tools [n]` -- n calls back to back, for what a run of them looks
|
|
//! like when a screen groups them.
|
|
//! - `/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
|
|
//! that is awkward to get a real model to produce on demand.
|
|
//! - `/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.
|
|
//! - `/compact` -- a compaction, start to finish. Typed rather than
|
|
//! pressed, because the real dialects take it as a typed command too and
|
|
//! the phone no longer has a button for it.
|
|
//!
|
|
//! This is exactly the event vocabulary the real drivers produce, so a UI
|
|
//! that renders echo sessions correctly renders the real thing.
|
|
//!
|
|
//! `/slow` earns its place: a queued message, a Stop button, a spinner
|
|
//! where the answer will go are all states that only exist mid-turn, and
|
|
//! the obvious way to get one -- ask a real model to sleep -- does not
|
|
//! work. It declines, reasonably, and answers instantly instead, so the
|
|
//! state never arrives and the attempt still costs a turn on somebody's
|
|
//! account. A driver that can be *told* to take its time costs nothing and
|
|
//! is the same every run.
|
|
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
use std::time::Duration;
|
|
|
|
use super::driver::{Driver, Event, EventSink, ImageRef, QuestionOption, SessionStatus};
|
|
|
|
/// Delay between streamed deltas -- long enough that streaming is visibly
|
|
/// streaming in the UI, short enough that tests waiting on a full turn
|
|
/// stay fast.
|
|
const DELTA_DELAY: Duration = Duration::from_millis(50);
|
|
|
|
/// 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);
|
|
|
|
/// A question echo is waiting on, and the tool call it belongs to.
|
|
///
|
|
/// `call` is `None` for `/question`, which asks on its own the way a
|
|
/// permission does; `Some` for `/ask`, where several questions share one
|
|
/// call and the call ends when the last of them is answered.
|
|
struct PendingQuestion {
|
|
id: String,
|
|
call: Option<String>,
|
|
}
|
|
|
|
pub struct EchoDriver {
|
|
sink: EventSink,
|
|
/// Whether a turn is in flight, and what arrived during it.
|
|
///
|
|
/// A real CLI holds a message sent mid-turn and injects it at the next
|
|
/// tool boundary; echo used to answer it on the spot, which made it
|
|
/// the wrong shape for testing anything about queueing -- the status
|
|
/// dropped to idle immediately, so a phone had nothing to show as
|
|
/// pending. Holding it here is what makes echo able to stand in.
|
|
busy: Arc<AtomicBool>,
|
|
queued: Arc<Mutex<Vec<String>>>,
|
|
/// Ids of the questions awaiting an answer, in the order they were
|
|
/// asked. A list because `/ask` puts up to four on one tool call, the
|
|
/// way AskUserQuestion does, and the turn resumes when the last of
|
|
/// them is answered rather than the first.
|
|
pending_questions: Mutex<Vec<PendingQuestion>>,
|
|
}
|
|
|
|
impl EchoDriver {
|
|
/// An AskUserQuestion call, in the shape the CLI sends one.
|
|
///
|
|
/// Two questions on one call, because that is where the display is
|
|
/// hardest and where it was wrong: one question with four options
|
|
/// reads fine even when the options are laid out badly. Written out
|
|
/// in full rather than generated so it carries the parts that are
|
|
/// easy to leave out of a fixture -- a header, an option with a
|
|
/// description, an option with a preview block, and a multi-select.
|
|
fn ask_user_question(&self) {
|
|
// Written once, in the shape the events carry, and turned into
|
|
// the tool call's own input below -- the CLI sends both, and two
|
|
// hand-written copies of one question would drift.
|
|
let asked = [
|
|
(
|
|
"Theme",
|
|
"Which colour scheme should the transcript use?",
|
|
false,
|
|
vec![
|
|
QuestionOption {
|
|
label: "Catppuccin Mocha (Recommended)".to_string(),
|
|
description: Some(
|
|
"What the app uses now: a dark base with muted accents.".to_string(),
|
|
),
|
|
preview: None,
|
|
},
|
|
QuestionOption {
|
|
label: "Solarized Dark".to_string(),
|
|
description: Some(
|
|
"Lower contrast, warmer. Easier at night, harder in sun.".to_string(),
|
|
),
|
|
preview: None,
|
|
},
|
|
QuestionOption {
|
|
label: "High contrast".to_string(),
|
|
description: Some(
|
|
"Pure black behind white text, for reading outdoors.".to_string(),
|
|
),
|
|
preview: Some(
|
|
"background: #000000\nforeground: #ffffff\naccent: #ffd700"
|
|
.to_string(),
|
|
),
|
|
},
|
|
],
|
|
),
|
|
(
|
|
"Collapsed",
|
|
"Which of these should be shown collapsed by default?",
|
|
true,
|
|
vec![
|
|
QuestionOption {
|
|
label: "Tool calls".to_string(),
|
|
description: Some("A run of them becomes one card.".to_string()),
|
|
preview: None,
|
|
},
|
|
QuestionOption {
|
|
label: "Peer messages".to_string(),
|
|
description: Some("Messages from other agents.".to_string()),
|
|
preview: None,
|
|
},
|
|
QuestionOption {
|
|
label: "Compaction notes".to_string(),
|
|
description: Some("What a compaction recovered.".to_string()),
|
|
preview: None,
|
|
},
|
|
],
|
|
),
|
|
];
|
|
|
|
let call = format!("echo-ask-{}", super::random_hex());
|
|
self.emit(Event::Status {
|
|
state: SessionStatus::Running,
|
|
});
|
|
self.emit(Event::ToolStart {
|
|
id: call.clone(),
|
|
tool: "AskUserQuestion".to_string(),
|
|
input: serde_json::json!({"questions": asked
|
|
.iter()
|
|
.map(|(header, question, multi, options)| serde_json::json!({
|
|
"question": question,
|
|
"header": header,
|
|
"multiSelect": multi,
|
|
"options": options,
|
|
}))
|
|
.collect::<Vec<_>>()}),
|
|
});
|
|
for (index, (header, question, multi, options)) in asked.into_iter().enumerate() {
|
|
let id = format!("{call}#{index}");
|
|
self.pending_questions
|
|
.lock()
|
|
.unwrap()
|
|
.push(PendingQuestion {
|
|
id: id.clone(),
|
|
call: Some(call.clone()),
|
|
});
|
|
self.emit(Event::Question {
|
|
id,
|
|
prompt: question.to_string(),
|
|
header: Some(header.to_string()),
|
|
options,
|
|
multi_select: multi,
|
|
// The call that asked, so all of it draws as one thing --
|
|
// which is the whole point of the fixture.
|
|
about: Some(call.clone()),
|
|
});
|
|
}
|
|
self.emit(Event::Status {
|
|
state: SessionStatus::AwaitingInput,
|
|
});
|
|
}
|
|
|
|
/// One typed line, whether it arrived as a message or as a command.
|
|
///
|
|
/// `announce` is the difference and it is the whole of it: a message
|
|
/// is announced with `MessageTaken`, which is what puts it in the
|
|
/// transcript, and a command is not -- the manager has already
|
|
/// recorded that one was sent, and saying so twice drew the same
|
|
/// line in both colours.
|
|
fn handle(&self, text: String, _images: Vec<ImageRef>, announce: bool) {
|
|
let sink = self.sink.clone();
|
|
|
|
// Mid-turn messages are held rather than answered, the way a real
|
|
// CLI holds them until the next tool boundary. Without this the
|
|
// session went idle the instant one arrived, and every state that
|
|
// only exists while something is queued was untestable.
|
|
if self.busy.load(Ordering::SeqCst) {
|
|
self.queued.lock().unwrap().push(text);
|
|
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") {
|
|
if announce {
|
|
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;
|
|
}
|
|
|
|
// The same word the real CLI takes, so a phone drives both the same
|
|
// way. `Driver::compact` is what the manager's own route calls;
|
|
// this is the typed path onto it.
|
|
if text.trim() == "/compact" {
|
|
if announce {
|
|
self.emit(Event::MessageTaken { text });
|
|
}
|
|
self.compact();
|
|
return;
|
|
}
|
|
|
|
if text.trim() == "/ask" {
|
|
if announce {
|
|
self.emit(Event::MessageTaken { text });
|
|
}
|
|
self.ask_user_question();
|
|
return;
|
|
}
|
|
|
|
if let Some(rest) = text.strip_prefix("/question") {
|
|
let id = format!("q-{}", super::random_hex());
|
|
let prompt = if rest.trim().is_empty() {
|
|
"Echo asks: proceed?".to_string()
|
|
} else {
|
|
format!("Echo asks: {}", rest.trim())
|
|
};
|
|
self.pending_questions
|
|
.lock()
|
|
.unwrap()
|
|
.push(PendingQuestion {
|
|
id: id.clone(),
|
|
call: None,
|
|
});
|
|
self.emit(Event::Status {
|
|
state: SessionStatus::Running,
|
|
});
|
|
self.emit(Event::Question {
|
|
id,
|
|
prompt,
|
|
header: None,
|
|
options: vec![QuestionOption::plain("Yes"), QuestionOption::plain("No")],
|
|
multi_select: false,
|
|
about: None,
|
|
});
|
|
self.emit(Event::Status {
|
|
state: SessionStatus::AwaitingInput,
|
|
});
|
|
return;
|
|
}
|
|
|
|
// Checked before `/tool`, which is a prefix of it: matching the
|
|
// shorter one first would read "/tools 4" as a single tool whose
|
|
// input is "s 4".
|
|
let many_tools = text.strip_prefix("/tools").map(|rest| {
|
|
// At least two, because one call is not a run of them and this
|
|
// exists to produce a run.
|
|
rest.trim().parse::<usize>().unwrap_or(3).clamp(2, 12)
|
|
});
|
|
let run_tool = if many_tools.is_some() {
|
|
None
|
|
} else {
|
|
text.strip_prefix("/tool")
|
|
.map(|rest| rest.trim().to_string())
|
|
};
|
|
// Seconds to stay running before answering, default 30. Clamped
|
|
// rather than trusted: this is a test affordance, and a session
|
|
// pinned running for an hour by a typo is a worse outcome than a
|
|
// short wait.
|
|
let linger = text.strip_prefix("/slow").map(|rest| {
|
|
Duration::from_secs(rest.trim().parse::<u64>().unwrap_or(30).clamp(1, 600))
|
|
});
|
|
let fail = text
|
|
.strip_prefix("/error")
|
|
.map(|rest| rest.trim().to_string());
|
|
let busy = Arc::clone(&self.busy);
|
|
let queued = Arc::clone(&self.queued);
|
|
busy.store(true, Ordering::SeqCst);
|
|
tokio::spawn(async move {
|
|
let send = |event: Event| {
|
|
let _ = sink.send(event);
|
|
};
|
|
let finish = || finish_turn(&sink, &queued, &busy);
|
|
// Echo takes a message the instant it gets one, but it says so
|
|
// anyway: a driver that skips this leaves the phone holding a
|
|
// message it thinks is still queued, and the point of an echo
|
|
// provider is that it behaves like the real ones.
|
|
if announce {
|
|
send(Event::MessageTaken { text: text.clone() });
|
|
}
|
|
send(Event::Status {
|
|
state: SessionStatus::Running,
|
|
});
|
|
|
|
if let Some(linger) = linger {
|
|
// A delta a second: visibly alive rather than merely slow,
|
|
// which is what the states being looked at accompany.
|
|
let seconds = linger.as_secs();
|
|
for remaining in (1..=seconds).rev() {
|
|
send(Event::AssistantText {
|
|
delta: format!("still working, {remaining}s\n"),
|
|
});
|
|
tokio::time::sleep(Duration::from_secs(1)).await;
|
|
}
|
|
send(Event::AssistantText {
|
|
delta: "done.".to_string(),
|
|
});
|
|
finish();
|
|
return;
|
|
}
|
|
|
|
if let Some(message) = fail {
|
|
send(Event::Error {
|
|
message: if message.is_empty() {
|
|
"echo was asked to fail".to_string()
|
|
} else {
|
|
message
|
|
},
|
|
});
|
|
finish();
|
|
return;
|
|
}
|
|
|
|
if let Some(count) = many_tools {
|
|
for i in 1..=count {
|
|
let id = format!("t-{}", super::random_hex());
|
|
send(Event::ToolStart {
|
|
id: id.clone(),
|
|
tool: if i % 2 == 0 { "Read" } else { "Bash" }.to_string(),
|
|
input: serde_json::json!({
|
|
"command": format!("grep -rn 'call {i}' /tmp | head -3"),
|
|
"file_path": format!("/tmp/call-{i}.txt"),
|
|
"description": format!("The {i} of {count} calls in this run"),
|
|
"timeout": 5000,
|
|
}),
|
|
});
|
|
tokio::time::sleep(DELTA_DELAY).await;
|
|
send(Event::ToolEnd {
|
|
id,
|
|
output: format!("call {i} finished"),
|
|
});
|
|
}
|
|
finish();
|
|
return;
|
|
}
|
|
|
|
if let Some(input) = run_tool {
|
|
let id = format!("t-{}", super::random_hex());
|
|
send(Event::ToolStart {
|
|
id: id.clone(),
|
|
tool: "echo-tool".to_string(),
|
|
input: serde_json::json!({ "input": input }),
|
|
});
|
|
tokio::time::sleep(DELTA_DELAY).await;
|
|
send(Event::ToolUpdate {
|
|
id: id.clone(),
|
|
output: "working...".to_string(),
|
|
});
|
|
tokio::time::sleep(DELTA_DELAY).await;
|
|
send(Event::ToolEnd {
|
|
id,
|
|
output: format!("echoed: {input}"),
|
|
});
|
|
}
|
|
|
|
// Word-at-a-time so streaming is visibly streaming.
|
|
for word in format!("You said: {text}").split_inclusive(' ') {
|
|
send(Event::AssistantText {
|
|
delta: word.to_string(),
|
|
});
|
|
tokio::time::sleep(DELTA_DELAY).await;
|
|
}
|
|
send(Event::UsageDelta {
|
|
tokens: text.split_whitespace().count() as u64,
|
|
});
|
|
finish();
|
|
});
|
|
}
|
|
|
|
pub fn new(sink: EventSink) -> Self {
|
|
let driver = Self {
|
|
sink,
|
|
pending_questions: Mutex::new(Vec::new()),
|
|
busy: Arc::new(AtomicBool::new(false)),
|
|
queued: Arc::new(Mutex::new(Vec::new())),
|
|
};
|
|
driver.emit(Event::Status {
|
|
state: SessionStatus::Idle,
|
|
});
|
|
driver
|
|
}
|
|
|
|
/// Sends are infallible from the driver's point of view: a closed sink
|
|
/// means the session is being torn down, and there is nobody left to
|
|
/// report to.
|
|
fn emit(&self, event: Event) {
|
|
let _ = self.sink.send(event);
|
|
}
|
|
}
|
|
|
|
/// Ending a turn is also when anything held during it is taken up -- the
|
|
/// moment a real CLI would have injected it. One place, because a turn has
|
|
/// several ways to end (a reply, an interrupt, a compaction) and every one
|
|
/// of them owes the same answer.
|
|
fn finish_turn(sink: &EventSink, queued: &Mutex<Vec<String>>, busy: &AtomicBool) {
|
|
let held = std::mem::take(&mut *queued.lock().unwrap());
|
|
for text in held {
|
|
// Announced before it is answered, in that order: a phone showing
|
|
// the message as pending needs the signal that it has been read,
|
|
// and the answer is meaningless above a message still drawn as
|
|
// waiting.
|
|
let _ = sink.send(Event::MessageTaken { text: text.clone() });
|
|
let _ = sink.send(Event::AssistantText {
|
|
delta: format!("\n(taken from the queue) You said: {text}"),
|
|
});
|
|
}
|
|
busy.store(false, Ordering::SeqCst);
|
|
let _ = sink.send(Event::Status {
|
|
state: SessionStatus::Idle,
|
|
});
|
|
}
|
|
|
|
impl Driver for EchoDriver {
|
|
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
|
|
// from the transcript is the thing echo must not model. A command
|
|
// owes none -- the manager has already recorded that it was sent,
|
|
// and announcing it again drew the same line twice, once in each
|
|
// colour.
|
|
self.handle(text, images, true);
|
|
}
|
|
|
|
/// Echo's commands *are* its messages -- `/tool`, `/slow`, `/ask` --
|
|
/// so this is the same path with the same parsing, and the fixture
|
|
/// behaves like a real session driven the same way.
|
|
fn run_command(&self, text: &str) {
|
|
self.handle(text.to_string(), Vec::new(), false);
|
|
}
|
|
|
|
fn answer_question(&self, id: &str, answers: &[String]) {
|
|
let answer = answers.join(", ");
|
|
let (answered, waiting) = {
|
|
let mut pending = self.pending_questions.lock().unwrap();
|
|
let Some(at) = pending.iter().position(|question| question.id == id) else {
|
|
self.emit(Event::Error {
|
|
message: format!("no question {id} is awaiting an answer"),
|
|
});
|
|
return;
|
|
};
|
|
let answered = pending.remove(at);
|
|
// Whether anything on the same call is still unanswered: a
|
|
// tool that asked four questions ends once, not four times.
|
|
let waiting = answered
|
|
.call
|
|
.as_ref()
|
|
.is_some_and(|call| pending.iter().any(|q| q.call.as_ref() == Some(call)));
|
|
(answered, waiting)
|
|
};
|
|
if waiting {
|
|
return;
|
|
}
|
|
if let Some(call) = answered.call {
|
|
self.emit(Event::ToolEnd {
|
|
id: call,
|
|
output: format!("answered: {answer}"),
|
|
});
|
|
} else {
|
|
self.emit(Event::AssistantText {
|
|
delta: format!("You answered: {answer}"),
|
|
});
|
|
}
|
|
self.emit(Event::Status {
|
|
state: SessionStatus::Idle,
|
|
});
|
|
}
|
|
|
|
fn interrupt(&self) {
|
|
// Nothing real to stop; a pending question is abandoned so the
|
|
// session isn't stuck awaiting input forever.
|
|
self.pending_questions.lock().unwrap().clear();
|
|
self.emit(Event::Status {
|
|
state: SessionStatus::Idle,
|
|
});
|
|
}
|
|
|
|
// Nothing to forward: this process has no notion of what the
|
|
// conversation is called, and the rename it belongs to has already
|
|
// happened where the name lives. See `Driver::set_title`.
|
|
fn set_title(&self, _title: &str) {}
|
|
|
|
fn set_permission_mode(&self, mode: &str) {
|
|
self.emit(Event::Error {
|
|
message: format!("an echo session asks for nothing, so {mode} changes nothing"),
|
|
});
|
|
}
|
|
|
|
fn set_model(&self, model: &str) {
|
|
self.emit(Event::Error {
|
|
message: format!("echo sessions have no model to change to {model}"),
|
|
});
|
|
}
|
|
|
|
/// A compaction with nothing to compact.
|
|
///
|
|
/// The counts are invented, like everything else this driver says --
|
|
/// what is real is the shape and the order: busy, a pause long enough
|
|
/// to see, then the result. `Compacting` and `Compacted` are states a
|
|
/// screen has to draw, and the only other way to reach them is to fill
|
|
/// a real session's context and spend two minutes of somebody's
|
|
/// account getting it back.
|
|
fn compact(&self) {
|
|
let sink = self.sink.clone();
|
|
let queued = Arc::clone(&self.queued);
|
|
let busy = Arc::clone(&self.busy);
|
|
busy.store(true, Ordering::SeqCst);
|
|
tokio::spawn(async move {
|
|
let _ = sink.send(Event::Status {
|
|
state: SessionStatus::Compacting,
|
|
});
|
|
tokio::time::sleep(COMPACT_TIME).await;
|
|
let _ = sink.send(Event::Compacted {
|
|
pre_tokens: Some(128_402),
|
|
post_tokens: Some(9_617),
|
|
trigger: Some("manual".to_string()),
|
|
});
|
|
finish_turn(&sink, &queued, &busy);
|
|
});
|
|
}
|
|
|
|
/// Nothing to detach from and nothing to stop: the echo driver has no
|
|
/// process, so both halves of the way out are already done.
|
|
fn detach(&self) {}
|
|
|
|
fn stop(&self) {}
|
|
}
|