iris
82401cd887
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.
1610 lines
70 KiB
Rust
1610 lines
70 KiB
Rust
//! The Claude Code driver: `claude -p` speaking stream-json on stdio,
|
|
//! translated into the common event model.
|
|
//!
|
|
//! This half owns the process -- starting it, *adopting one this server
|
|
//! left running*, writing lines to it, and ending it. Where it runs is
|
|
//! `session::transport`'s business, not this file's: this one emits a
|
|
//! `Launch` and never learns whether it became a local child or an ssh
|
|
//! invocation.
|
|
//!
|
|
//! The process is meant to outlive the server, so that restarting the
|
|
//! backend does not end a turn: its stdio lives in the session directory
|
|
//! (a fifo it holds open itself, plus logs read from a byte offset) and
|
|
//! `session::process` records what it is. Everything comes through
|
|
//! [`ClaudeDriver::launch`], which adopts if it can and starts if it
|
|
//! cannot -- `--resume` is reachable only on the second path, because two
|
|
//! CLIs on one session file duplicate the conversation into it.
|
|
//! Turning a line into [`Event`]s is [`translate`], which changes when the
|
|
//! CLI's wire format does rather than when any of the above does.
|
|
//!
|
|
//! The probing record below stays here, since it is the provenance for
|
|
//! both halves: the flags are this file's, the message catalogue is what
|
|
//! `translate` implements.
|
|
//!
|
|
//! Wire format pinned against CLI 2.1.237 by probing (2026-08-24; scripts
|
|
//! summarized here since they live outside the repo):
|
|
//!
|
|
//! - Outbound: `system/init` (carries `session_id`, the `--resume` token),
|
|
//! `stream_event` (raw API deltas; `text_delta` is the streaming text),
|
|
//! consolidated `assistant` messages (their `tool_use` blocks have the
|
|
//! complete input), `user` messages with `tool_result` blocks, a `result`
|
|
//! per turn (usage + cost), `control_request` for anything needing a
|
|
//! human, `control_response` answering ours.
|
|
//! - Permission prompts require the hidden `--permission-prompt-tool stdio`
|
|
//! flag; they arrive as `control_request{subtype:can_use_tool}` and are
|
|
//! answered with `{behavior:"allow",updatedInput}` or
|
|
//! `{behavior:"deny",message}`. `AskUserQuestion` uses the same shape,
|
|
//! with the chosen labels added to `updatedInput` as
|
|
//! `answers:{<question text>:<label>}`.
|
|
//! - Inbound `user` messages sent mid-turn are queued and injected at the
|
|
//! next tool boundary (verified live: the model acknowledged a steer
|
|
//! between two Bash calls) -- the behavior this app exists for.
|
|
//! - `control_request{subtype:set_model}` answers success;
|
|
//! `{subtype:interrupt}` stops the turn.
|
|
//! - `control_request{subtype:set_permission_mode}` answers success and
|
|
//! echoes the mode back (`{"response":{"mode":"acceptEdits"}}`), so the
|
|
//! mode is changeable mid-session rather than only at spawn. Probed the
|
|
//! same way as the rest, against 2.1.237 on 2026-08-29.
|
|
|
|
use std::collections::VecDeque;
|
|
use std::os::unix::fs::OpenOptionsExt;
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::atomic::{AtomicBool, Ordering};
|
|
use std::sync::{Arc, Mutex};
|
|
|
|
use anyhow::{Context, Result};
|
|
use serde_json::{Value, json};
|
|
use tokio::io::AsyncWriteExt;
|
|
use tokio::sync::mpsc;
|
|
|
|
use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus, Unqueued};
|
|
use super::process;
|
|
use super::transport::{Launch, Streams, Transport};
|
|
use crate::config::{ProviderConfig, SessionConfig};
|
|
use translate::{AnswerOutcome, Setting, Translator, starts_a_model_call};
|
|
|
|
/// How much of a failing process's stderr the exit report carries.
|
|
///
|
|
/// Enough for a shell's complaint plus the context it prints around it --
|
|
/// fish's `cd` failure is seven lines including a caret pointing at the
|
|
/// offending line -- and bounded because this is held per session for the
|
|
/// life of the process and a chatty program would otherwise grow without
|
|
/// limit.
|
|
const STDERR_LINES_KEPT: usize = 50;
|
|
|
|
/// The kept stderr as one block, with blank lines trimmed off both ends.
|
|
///
|
|
/// The trailing trim is the point: a shell's error ends with a blank line,
|
|
/// so the last line of stderr is routinely empty and anything that reports
|
|
/// "the last line" reports nothing at all.
|
|
fn tail_of(kept: &VecDeque<String>) -> String {
|
|
let lines: Vec<&str> = kept.iter().map(String::as_str).collect();
|
|
let start = lines
|
|
.iter()
|
|
.position(|line| !line.trim().is_empty())
|
|
.unwrap_or(lines.len());
|
|
let end = lines
|
|
.iter()
|
|
.rposition(|line| !line.trim().is_empty())
|
|
.map(|last| last + 1)
|
|
.unwrap_or(start);
|
|
lines[start..end].join("\n")
|
|
}
|
|
|
|
/// Where the driver remembers its CLI session id between backend runs --
|
|
/// the whole crash-recovery story: respawning with `--resume <id>` picks
|
|
/// the conversation back up from Claude's own session files. Kept in the
|
|
/// session directory rather than config.ron so the shared schema stays
|
|
/// free of per-driver state.
|
|
pub(super) mod translate;
|
|
|
|
const RESUME_FILE: &str = "claude-session.json";
|
|
|
|
/// Messages handed to the CLI that it has not visibly acted on yet.
|
|
///
|
|
/// The CLI *does* take a message written mid-turn: it goes into the next
|
|
/// model call, which is the next tool boundary, and the whole point of
|
|
/// this app is steering a turn that is already running. An earlier version
|
|
/// of this file claimed the opposite and held every mid-turn message until
|
|
/// the turn ended -- so a steer sent after the second tool call sat unread
|
|
/// until all the work it was meant to redirect had finished. Measured
|
|
/// rather than argued: a line written between two Bash calls was answered
|
|
/// inside the same turn, with one `result` for the whole thing.
|
|
///
|
|
/// What the CLI does not do is say on stdout that it has read one. So the
|
|
/// line goes out immediately and the *announcement* waits here instead,
|
|
/// until the CLI opens the next model call -- see
|
|
/// [`translate::starts_a_model_call`]. That keeps a phone's held bubble
|
|
/// where it belongs -- below the working indicator until the session has
|
|
/// actually taken it -- without delaying the message itself to get it.
|
|
///
|
|
/// The proof has to be the model call and not the output, which is what
|
|
/// an earlier version took it to be. Assistant text and a tool call both
|
|
/// keep arriving from a message that was *already in flight* when the
|
|
/// steer was written, and that message saw none of it: a steer sent
|
|
/// while an answer was streaming was recorded in the middle of it, above
|
|
/// tool calls the model had already committed to. On screen the answer
|
|
/// split into two bubbles around a message it had not read, and the tool
|
|
/// results that followed read as things the steer had asked for.
|
|
#[derive(Default)]
|
|
struct Queue {
|
|
/// A turn is in flight, so a message sent now is a steer into it.
|
|
running: bool,
|
|
/// Written, not yet announced, oldest first, each with the id of the
|
|
/// `MessageQueued` that told the phone it was waiting -- so the
|
|
/// announcement can name which bubble it resolves.
|
|
awaiting: VecDeque<(String, String, Vec<ImageRef>)>,
|
|
/// The process is gone, so nothing can be taken up any more.
|
|
///
|
|
/// Needed because every other way out of a turn is an `Idle` this
|
|
/// driver sees, and an exit is the one that is not. Without it a
|
|
/// process that died mid-turn left `running` true for good, and since
|
|
/// a message is only recorded when it is *announced*, each later one
|
|
/// vanished with nothing on screen to say it had not been delivered.
|
|
closed: bool,
|
|
}
|
|
|
|
impl Queue {
|
|
/// Gives up on everything held, because the process is gone.
|
|
///
|
|
/// 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.
|
|
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();
|
|
if lost.is_empty() {
|
|
return;
|
|
}
|
|
let _ = sink.send(Event::Error {
|
|
message: format!(
|
|
"{why} before it read {}: {}",
|
|
if lost.len() == 1 {
|
|
"this message".to_string()
|
|
} else {
|
|
format!("{} queued messages", lost.len())
|
|
},
|
|
lost.join(" / ")
|
|
),
|
|
});
|
|
}
|
|
}
|
|
|
|
/// The session directory's copies of the process's standard streams.
|
|
///
|
|
/// Named once rather than built at each use, because the spawn path and
|
|
/// the attach path must agree about which file is which; if they drift,
|
|
/// a reattached session reads a file nothing is writing and simply looks
|
|
/// idle forever.
|
|
const STDIN_FIFO: &str = "stdin.fifo";
|
|
const STDOUT_LOG: &str = "stdout.log";
|
|
const STDERR_LOG: &str = "stderr.log";
|
|
|
|
/// How often a reader with nothing to read looks again.
|
|
///
|
|
/// A poll rather than a watch: the alternative is an inotify dependency
|
|
/// for one file per session, and at this interval the streaming text is
|
|
/// already arriving faster than a phone renders it.
|
|
const POLL: std::time::Duration = std::time::Duration::from_millis(50);
|
|
|
|
pub struct ClaudeDriver {
|
|
sink: EventSink,
|
|
queue: Arc<Mutex<Queue>>,
|
|
/// Lines for the process's stdin.
|
|
///
|
|
/// Not closeable, unlike the pipe this used to be: stdin is a fifo the
|
|
/// process holds open itself, so closing this end says nothing to it.
|
|
/// Ending the process is [`Driver::stop`]'s job and it uses a signal.
|
|
to_child: mpsc::UnboundedSender<String>,
|
|
state: Arc<Mutex<Translator>>,
|
|
session_dir: PathBuf,
|
|
/// Cleared to stop the reader without touching the process -- which is
|
|
/// exactly what detaching is.
|
|
reading: Arc<AtomicBool>,
|
|
}
|
|
|
|
impl ClaudeDriver {
|
|
/// Takes charge of this session's process: the one already running if
|
|
/// there is one, otherwise a new one.
|
|
///
|
|
/// One entry point rather than two, because the choice is not the
|
|
/// caller's to make and getting it wrong is the expensive bug. A
|
|
/// second `--resume` against a session file that is already open
|
|
/// duplicates the whole conversation into it and bills the reattached
|
|
/// copy for re-reading it -- measured at 65 MB and 154 screenshots on
|
|
/// 2026-08-29, when an import of a *live* session did exactly this.
|
|
/// So `--resume` is reachable only through the spawn half below, under
|
|
/// a check that nothing is running.
|
|
pub fn launch(
|
|
meta: &SessionConfig,
|
|
provider: &ProviderConfig,
|
|
transport: &Transport,
|
|
session_dir: &Path,
|
|
sink: EventSink,
|
|
) -> Result<Self> {
|
|
let state = Arc::new(Mutex::new(Translator::new(session_dir.to_path_buf())));
|
|
let queue = Arc::new(Mutex::new(Queue::default()));
|
|
let reading = Arc::new(AtomicBool::new(true));
|
|
|
|
// Adopting is only possible for a process this server left behind
|
|
// on this machine: an ssh session's child is at the far end of a
|
|
// connection that died with the server, so there is nothing there
|
|
// to find. Nothing was ever recorded for one, so this answers "no"
|
|
// without needing to know that, which is why the remote case is
|
|
// not a branch here.
|
|
// Whether this launch *started* a process or picked up one that was
|
|
// already there. The two owe the session different things -- see the
|
|
// `Status` below.
|
|
let started_here;
|
|
let record = match process::recorded(session_dir) {
|
|
// Still running, and ours. Pick it up where it was left --
|
|
// the one path that must not pass `--resume`.
|
|
Some((record, process::Liveness::Alive)) => {
|
|
tracing::info!(
|
|
"session {} reattaching to the {} it left running (pid {})",
|
|
meta.id,
|
|
provider.name,
|
|
record.pid
|
|
);
|
|
started_here = false;
|
|
record
|
|
}
|
|
// Recorded, and the machine will not say whether it is still
|
|
// there. Starting one anyway is the mistake this module is
|
|
// for, so nothing is started; `follow` keeps asking and
|
|
// reports the state as unknown until it gets an answer.
|
|
Some((record, process::Liveness::Unknown)) => {
|
|
tracing::warn!(
|
|
"session {} recorded pid {} but this machine won't say whether it is running; \
|
|
not starting a second one",
|
|
meta.id,
|
|
record.pid
|
|
);
|
|
started_here = false;
|
|
record
|
|
}
|
|
Some((_, process::Liveness::Dead)) | None => {
|
|
started_here = true;
|
|
Self::start(meta, provider, transport, session_dir)?
|
|
}
|
|
};
|
|
|
|
// A process this driver has just started has been asked for nothing,
|
|
// which is what idle means. Said here because nothing else will say
|
|
// it: the CLI writes not one line until it is given work, so a
|
|
// session whose transcript last recorded `Exited` -- one whose
|
|
// process died while this server was down, or one somebody stopped
|
|
// from the phone -- would keep that word. `Exited` refuses every
|
|
// command sent to the session, and it offers a phone the chance to
|
|
// start a second CLI against a conversation that already has one.
|
|
//
|
|
// From the driver rather than from the manager, and before `follow`
|
|
// is spawned, so it cannot overtake the exit `follow` reports for a
|
|
// process that dies immediately: both come from here, in this order.
|
|
// Adopting says nothing, because a process that was already running
|
|
// may be mid-turn, and the transcript's last word is the better
|
|
// answer until its output says otherwise.
|
|
//
|
|
// The llama driver has always done this (see `LlamaDriver::attached`,
|
|
// which reports `Running` while the model loads and `Idle` when it
|
|
// answers); this side was the one silent about it.
|
|
if started_here {
|
|
let _ = sink.send(Event::Status {
|
|
state: SessionStatus::Idle,
|
|
});
|
|
}
|
|
// Where reading of its output had reached. A process just started
|
|
// has said nothing, so its record says zero and this is the same
|
|
// question with the same answer.
|
|
let resuming_from = match record.detail {
|
|
process::Detail::Stdio { stdout_read } => stdout_read,
|
|
// A record of the wrong shape belongs to a different driver;
|
|
// read its output from the start rather than trusting an
|
|
// offset into a file that means something else.
|
|
_ => 0,
|
|
};
|
|
|
|
// The writer end of the fifo. Opened write-only here: the process
|
|
// holds its own read-write handle, so this side coming and going
|
|
// across a restart is invisible to it.
|
|
let stdin = std::fs::OpenOptions::new()
|
|
.write(true)
|
|
.open(session_dir.join(STDIN_FIFO))
|
|
.with_context(|| format!("opening {STDIN_FIFO} for session {}", meta.id))?;
|
|
let (to_child, mut from_driver) = mpsc::unbounded_channel::<String>();
|
|
tokio::spawn(async move {
|
|
let mut stdin = tokio::fs::File::from_std(stdin);
|
|
while let Some(line) = from_driver.recv().await {
|
|
if stdin.write_all(line.as_bytes()).await.is_err()
|
|
|| stdin.write_all(b"\n").await.is_err()
|
|
|| stdin.flush().await.is_err()
|
|
{
|
|
break;
|
|
}
|
|
}
|
|
});
|
|
|
|
tokio::spawn(follow(
|
|
session_dir.to_path_buf(),
|
|
record,
|
|
resuming_from,
|
|
Arc::clone(&state),
|
|
sink.clone(),
|
|
Arc::clone(&queue),
|
|
Arc::clone(&reading),
|
|
format!("{} {}", provider.name, transport.describe()),
|
|
));
|
|
|
|
Ok(Self {
|
|
sink,
|
|
queue,
|
|
to_child,
|
|
state,
|
|
session_dir: session_dir.to_path_buf(),
|
|
reading,
|
|
})
|
|
}
|
|
|
|
/// Starts a new CLI for this session, with its streams in the session
|
|
/// directory so the next run of this server can find them.
|
|
///
|
|
/// The only path that passes `--resume`, and it is reached only when
|
|
/// nothing is running -- see [`ClaudeDriver::launch`].
|
|
fn start(
|
|
meta: &SessionConfig,
|
|
provider: &ProviderConfig,
|
|
transport: &Transport,
|
|
session_dir: &Path,
|
|
) -> Result<process::Record> {
|
|
let mut args: Vec<String> = ["-p", "--verbose"].iter().map(|a| a.to_string()).collect();
|
|
let mut push = |flag: &str, value: &str| {
|
|
args.push(flag.to_string());
|
|
args.push(value.to_string());
|
|
};
|
|
push("--input-format", "stream-json");
|
|
push("--output-format", "stream-json");
|
|
// Hidden but load-bearing: without it the CLI resolves permissions
|
|
// itself and nothing ever reaches the phone.
|
|
push("--permission-prompt-tool", "stdio");
|
|
if let Some(model) = &meta.model {
|
|
push("--model", model);
|
|
}
|
|
if let Some(mode) = &meta.permission_mode {
|
|
push("--permission-mode", mode);
|
|
}
|
|
// Named at birth, so this session is the same session in the CLI's
|
|
// own picker and in what other agents see when they list it.
|
|
//
|
|
// Only when we are the ones creating it. A resume is a session
|
|
// that already existed -- an import, or this server starting again
|
|
// -- and it already has whatever name it was given, quite possibly
|
|
// by the person who was typing in it. Renaming that from a title
|
|
// we derived from its first message would be taking something the
|
|
// app was only ever shown. `Driver::set_title` is how it changes
|
|
// after this point, and that one is asked for.
|
|
match read_resume_token(session_dir) {
|
|
Some(resume) => push("--resume", &resume),
|
|
None => push("--name", &meta.title),
|
|
}
|
|
args.push("--include-partial-messages".to_string());
|
|
// Makes `bypassPermissions` *reachable*, without selecting it: the
|
|
// session still starts in whatever mode was asked for above, and
|
|
// only moves if somebody moves it.
|
|
//
|
|
// Here because the CLI is asymmetric about that mode, which is not
|
|
// obvious and cost a confused bug report. It will *launch* in
|
|
// `bypassPermissions` on the strength of `--permission-mode` alone
|
|
// -- so spawning straight into it from the phone has always worked
|
|
// -- but it refuses to *switch* into it later:
|
|
//
|
|
// Cannot set permission mode to bypassPermissions because the
|
|
// session was not launched with --dangerously-skip-permissions
|
|
//
|
|
// So the phone's own mode picker offered a mode that could not be
|
|
// picked, on every session it had not been given at birth. Since
|
|
// the mode is already reachable at spawn, this grants nothing that
|
|
// was being withheld; it makes the two routes to it agree.
|
|
//
|
|
// Measured against 2.1.237, both ways round: without this flag the
|
|
// control request comes back `subtype: error` with the message
|
|
// above, and with it `subtype: success, mode: bypassPermissions`.
|
|
// Note it is the `--allow-` form -- `--dangerously-skip-permissions`
|
|
// is the one that turns it on for everything, and that would take
|
|
// the choice away from whoever is holding the phone.
|
|
args.push("--allow-dangerously-skip-permissions".to_string());
|
|
|
|
// Fresh logs, because the offsets that index them start at zero
|
|
// and everything the previous process said is already in the
|
|
// transcript.
|
|
let stdin = make_fifo(&session_dir.join(STDIN_FIFO))?;
|
|
let stdout = create_log(&session_dir.join(STDOUT_LOG))?;
|
|
let stderr = create_log(&session_dir.join(STDERR_LOG))?;
|
|
|
|
let program = provider.command.as_deref().unwrap_or("claude");
|
|
let launch = Launch::new(program, args, meta.cwd.as_deref());
|
|
let child = transport.spawn(
|
|
&launch,
|
|
Streams::Detached {
|
|
stdin: stdin.into(),
|
|
stdout: stdout.into(),
|
|
stderr: stderr.into(),
|
|
},
|
|
)?;
|
|
let pid = child
|
|
.id()
|
|
.context("the process exited before it could be recorded")?;
|
|
tracing::info!(
|
|
"session {} running {program} {} as pid {pid}",
|
|
meta.id,
|
|
transport.describe()
|
|
);
|
|
|
|
// Reaped rather than waited on. This server is the parent, so
|
|
// something has to collect the exit status or the process becomes
|
|
// a zombie -- but it is `follow` that decides what the session is
|
|
// doing, because after a restart there is no `Child` to wait on
|
|
// and the answer has to come from the same place either way.
|
|
tokio::spawn(async move {
|
|
let mut child = child;
|
|
let _ = child.wait().await;
|
|
});
|
|
|
|
let record = process::Record::of(pid, process::Detail::Stdio { stdout_read: 0 })
|
|
.context("the process was gone before its start time could be read")?;
|
|
process::write(session_dir, &record);
|
|
Ok(record)
|
|
}
|
|
|
|
fn send_line(&self, line: String) {
|
|
let _ = self.to_child.send(line);
|
|
}
|
|
|
|
/// Writes one of the CLI's own commands into the session.
|
|
///
|
|
/// Slash commands ride the normal user-message channel -- there is no
|
|
/// control request for them; `set_session_name` is not a subtype the
|
|
/// CLI knows, measured by asking. The turn they start is marked here
|
|
/// because they produce a `result` like any other, so a message sent
|
|
/// meanwhile belongs in the queue's "written, announce when read"
|
|
/// path rather than being reported as read the moment it is typed.
|
|
///
|
|
/// Nothing is emitted about the command itself: the manager has
|
|
/// already said it was sent, and the CLI announces what it does --
|
|
/// saying so here would be this side's guess standing in for its
|
|
/// measurement.
|
|
fn local_command(&self, text: String) {
|
|
let mut queue = self.queue.lock().unwrap();
|
|
// The same check `send_user_message` makes, for the same reason: a
|
|
// line written into a fifo nothing is reading goes nowhere and looks
|
|
// exactly like one that arrived. `Commands::submit` refuses a
|
|
// session already known to have exited, so what this catches is the
|
|
// process going away between that check and this write.
|
|
if queue.closed {
|
|
drop(queue);
|
|
let _ = self.sink.send(Event::Error {
|
|
message: format!("this session's process has exited, so it can't run {text}"),
|
|
});
|
|
return;
|
|
}
|
|
queue.running = true;
|
|
drop(queue);
|
|
// The session is working from this moment, and until now nothing
|
|
// said so: a command's reply carries no assistant text, so
|
|
// `proves_a_turn` never saw it and the recorded status stayed idle
|
|
// for the whole round trip -- which meant the *next* idle was not a
|
|
// change, so nothing was ever released behind it.
|
|
let _ = self.sink.send(Event::Status {
|
|
state: SessionStatus::Running,
|
|
});
|
|
self.send_line(
|
|
json!({"type": "user", "message": {"role": "user", "content": [
|
|
{"type": "text", "text": text}
|
|
]}})
|
|
.to_string(),
|
|
);
|
|
}
|
|
|
|
/// Sends a control request, remembering what it asked for.
|
|
///
|
|
/// `confirms` is the setting this request will have made if the CLI
|
|
/// answers success -- see [`Translator::expect_setting`], and
|
|
/// `Driver::set_model` for why a request is not a confirmation.
|
|
/// `None` for the ones that change no setting, like an interrupt.
|
|
///
|
|
/// The id is random rather than the clock it used to be: two requests
|
|
/// in the same second shared an id, which was harmless while nothing
|
|
/// looked one up and is not any more.
|
|
fn send_control(&self, request: Value, confirms: Option<Setting>) {
|
|
let id = format!("req-{}", super::random_hex());
|
|
if let Some(setting) = confirms {
|
|
// Before the line goes out: the reader thread is already
|
|
// running, and a fast answer to a slow lock arrives first.
|
|
self.state
|
|
.lock()
|
|
.unwrap()
|
|
.expect_setting(id.clone(), setting);
|
|
}
|
|
self.send_line(
|
|
json!({"type": "control_request", "request_id": id, "request": request}).to_string(),
|
|
);
|
|
}
|
|
}
|
|
|
|
impl Driver for ClaudeDriver {
|
|
fn send_user_message(&self, text: String, images: Vec<ImageRef>) {
|
|
let mut content = Vec::new();
|
|
for id in &images {
|
|
match attachment_block(&self.session_dir, id) {
|
|
Ok(block) => content.push(block),
|
|
Err(err) => {
|
|
let _ = self.sink.send(Event::Error {
|
|
message: format!("attachment {id} couldn't be sent: {err:#}"),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
if !text.is_empty() {
|
|
content.push(json!({"type": "text", "text": text}));
|
|
}
|
|
let line =
|
|
json!({"type": "user", "message": {"role": "user", "content": content}}).to_string();
|
|
let mut queue = self.queue.lock().unwrap();
|
|
// Saying so beats writing into a fifo that nothing is reading,
|
|
// which is what this used to do -- the message went nowhere and
|
|
// looked exactly like one that had been delivered.
|
|
if queue.closed {
|
|
drop(queue);
|
|
let _ = self.sink.send(Event::Error {
|
|
message: "this session's process has exited, so it can't be sent anything"
|
|
.to_string(),
|
|
});
|
|
return;
|
|
}
|
|
if queue.running {
|
|
// Into the running turn, now. Announced when the CLI shows it
|
|
// has been round the model again -- see `Queue`.
|
|
//
|
|
// The *waiting* is recorded here, though, which is the one
|
|
// thing that must not be left to the phone to remember: it put
|
|
// the bubble on screen from its own state, so leaving the
|
|
// session or restarting the app drew nothing pending while a
|
|
// message was still in the queue.
|
|
let id = super::random_hex();
|
|
queue
|
|
.awaiting
|
|
.push_back((id.clone(), text.clone(), images.clone()));
|
|
drop(queue);
|
|
let _ = self.sink.send(Event::MessageQueued { id, text, images });
|
|
self.send_line(line);
|
|
return;
|
|
}
|
|
queue.running = true;
|
|
drop(queue);
|
|
// Nothing is in flight, so there is nothing to wait for: this
|
|
// message *is* the turn about to start, and it never had a
|
|
// `MessageQueued` to resolve.
|
|
let _ = self.sink.send(Event::MessageTaken {
|
|
id: None,
|
|
text,
|
|
images,
|
|
});
|
|
let _ = self.sink.send(Event::Status {
|
|
state: SessionStatus::Running,
|
|
});
|
|
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();
|
|
state.answer(id, answers)
|
|
};
|
|
match response {
|
|
AnswerOutcome::Respond(control_response) => {
|
|
let _ = self.sink.send(Event::Status {
|
|
state: SessionStatus::Running,
|
|
});
|
|
self.send_line(control_response.to_string());
|
|
}
|
|
// A multi-question AskUserQuestion still waiting on the rest.
|
|
AnswerOutcome::Pending => {}
|
|
AnswerOutcome::Unknown => {
|
|
let _ = self.sink.send(Event::Error {
|
|
message: format!("no question {id} is awaiting an answer"),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Anything queued behind the interrupted turn still goes: it was
|
|
/// typed deliberately, and dropping it would lose a message that never
|
|
/// reached the transcript, with nothing on screen to say so.
|
|
fn interrupt(&self) {
|
|
// Recorded before the request goes out, so the result it produces is
|
|
// read as the stop somebody asked for rather than as a failure --
|
|
// see `Translator::interrupting`.
|
|
self.state.lock().unwrap().expect_interrupt();
|
|
self.send_control(json!({"subtype": "interrupt"}), None);
|
|
}
|
|
|
|
fn set_permission_mode(&self, mode: &str) {
|
|
self.send_control(
|
|
json!({"subtype": "set_permission_mode", "mode": mode}),
|
|
Some(Setting::PermissionMode(mode.to_string())),
|
|
);
|
|
}
|
|
|
|
fn set_model(&self, model: &str) {
|
|
self.send_control(
|
|
json!({"subtype": "set_model", "model": model}),
|
|
Some(Setting::Model(model.to_string())),
|
|
);
|
|
}
|
|
|
|
fn run_command(&self, text: &str) {
|
|
// Whatever the CLI's own vocabulary holds -- `/context`, `/usage`,
|
|
// a command added after this was written. It rides the same
|
|
// channel as `/compact` and `/rename` and starts a turn the same
|
|
// way, so the same bookkeeping applies; what it means is the
|
|
// CLI's business, not this file's.
|
|
self.local_command(text.to_string());
|
|
}
|
|
|
|
fn set_title(&self, title: &str) {
|
|
// The CLI's own mechanism, and a local command rather than a
|
|
// control request -- `set_session_name` is not a subtype it
|
|
// knows, measured by asking. It answers this the way it answers
|
|
// `/compact`: a fresh `init`, then a `result` for a turn with no
|
|
// model call in it, so the same "a turn is in flight" bookkeeping
|
|
// applies. A name with a newline in it would be two lines and the
|
|
// second would be a message, so it is refused rather than sent.
|
|
if title.contains('\n') {
|
|
let _ = self.sink.send(Event::Error {
|
|
message: "a session name cannot contain a line break".to_string(),
|
|
});
|
|
return;
|
|
}
|
|
self.local_command(format!("/rename {title}"));
|
|
}
|
|
|
|
fn compact(&self) {
|
|
self.local_command("/compact".to_string());
|
|
}
|
|
|
|
fn clear(&self) {
|
|
// Nothing is emitted here on purpose: the transcript should
|
|
// record a clear that happened, not one that was asked for. The
|
|
// CLI announces it with a `conversation_reset` line, which
|
|
// `translate.rs` turns into `Event::Cleared`, and follows it with
|
|
// a fresh `init` whose new `session_id` the reader persists as
|
|
// the resume token -- so the next launch resumes the cleared
|
|
// conversation with nothing here to keep in step.
|
|
self.local_command("/clear".to_string());
|
|
}
|
|
|
|
fn between_turns(&self) -> bool {
|
|
let queue = self.queue.lock().unwrap();
|
|
!queue.running && !queue.closed
|
|
}
|
|
|
|
fn detach(&self) {
|
|
// Stop reading and leave everything else exactly as it is. The
|
|
// process keeps its fifo (which it holds open itself), keeps
|
|
// writing its log, and keeps its record -- which is how the next
|
|
// run of this server finds it. See `Driver::detach`.
|
|
self.reading.store(false, Ordering::SeqCst);
|
|
}
|
|
|
|
fn stop(&self) {
|
|
self.reading.store(false, Ordering::SeqCst);
|
|
if let Some(record) = process::live(&self.session_dir) {
|
|
process::stop(&record, process::STOP_GRACE);
|
|
}
|
|
process::clear(&self.session_dir);
|
|
}
|
|
}
|
|
|
|
/// Follows the process's stdout log, turning it into events, and is also
|
|
/// what decides whether the session is still running.
|
|
///
|
|
/// One loop rather than a reader plus a monitor. After a restart there is
|
|
/// no `Child` to wait on -- the process was reparented away from this
|
|
/// server -- so liveness has to be a question asked of the record either
|
|
/// way, and asking it in two places is how the two answers come to
|
|
/// disagree.
|
|
///
|
|
/// Reading is resumable because the position is written down with the
|
|
/// process (see [`process::Record`]): everything before it is already in
|
|
/// the transcript, so a server coming back picks up exactly where the last
|
|
/// one stopped and the conversation has no hole in it.
|
|
#[allow(clippy::too_many_arguments)]
|
|
async fn follow(
|
|
session_dir: PathBuf,
|
|
mut record: process::Record,
|
|
mut offset: u64,
|
|
state: Arc<Mutex<Translator>>,
|
|
sink: EventSink,
|
|
queue: Arc<Mutex<Queue>>,
|
|
reading: Arc<AtomicBool>,
|
|
label: String,
|
|
) {
|
|
let stdout_path = session_dir.join(STDOUT_LOG);
|
|
let stderr_path = session_dir.join(STDERR_LOG);
|
|
// Whatever is already in the stderr log has been logged by whichever
|
|
// run of this server was watching when it was written, so a reattach
|
|
// starts at the end of it rather than repeating it. The tail is still
|
|
// read from the file if the process dies, which is when it matters.
|
|
let mut stderr_at = process::size_of(&stderr_path);
|
|
let mut said_unknown = false;
|
|
|
|
while reading.load(Ordering::SeqCst) {
|
|
let (bytes, _) = match process::read_from(&stdout_path, offset) {
|
|
Ok(found) => found,
|
|
Err(err) => {
|
|
// Reported, not only logged. This is the end of the
|
|
// session's output as far as anyone watching is
|
|
// concerned, and a phone told nothing shows a session
|
|
// that is merely quiet -- indistinguishable from one
|
|
// thinking. The status is `Unknown` rather than `Exited`
|
|
// because the process may well still be running; what
|
|
// has failed is this server's ability to hear it.
|
|
tracing::error!("couldn't read {}: {err:#}", stdout_path.display());
|
|
let _ = sink.send(Event::Error {
|
|
message: format!(
|
|
"lost track of {label}: its output can't be read ({err:#}). The process \
|
|
may still be running; restarting the backend will try to pick it up \
|
|
again."
|
|
),
|
|
});
|
|
queue
|
|
.lock()
|
|
.unwrap()
|
|
.close(&sink, "this server lost track of the session");
|
|
let _ = sink.send(Event::Status {
|
|
state: SessionStatus::Unknown,
|
|
});
|
|
return;
|
|
}
|
|
};
|
|
// Only whole lines, and the offset stops at the last newline -- so a
|
|
// line the process is halfway through writing is simply read again
|
|
// next pass. Deliberately *not* held in memory between passes: the
|
|
// offset would then have to point behind the bytes being held, and
|
|
// the next read would return them a second time to be prepended to
|
|
// the copy already there. It is also what makes the position
|
|
// crash-safe, since it never claims a partial line was handled.
|
|
//
|
|
// Counted in bytes rather than on a decoded string: a read can cut
|
|
// a multi-byte character in half, and the replacement character
|
|
// that decoding puts there is a different length from what it
|
|
// replaced -- which would slide the offset out of step with the
|
|
// file for the rest of the session.
|
|
let complete = complete_lines(&bytes);
|
|
|
|
for line in String::from_utf8_lossy(&bytes[..complete]).lines() {
|
|
if line.trim().is_empty() {
|
|
continue;
|
|
}
|
|
if !translate_line(line, &session_dir, &state, &sink, &queue) {
|
|
return; // session torn down
|
|
}
|
|
}
|
|
if complete > 0 {
|
|
offset += complete as u64;
|
|
record.detail = process::Detail::Stdio {
|
|
stdout_read: offset,
|
|
};
|
|
process::write(&session_dir, &record);
|
|
}
|
|
|
|
// Diagnostics only, and the tail of it is what an exit report
|
|
// carries -- so it is read from the file rather than kept in
|
|
// memory, which also means a reattached session can still explain
|
|
// a failure it did not witness.
|
|
if let Ok((bytes, at)) = process::read_from(&stderr_path, stderr_at)
|
|
&& at != stderr_at
|
|
{
|
|
stderr_at = at;
|
|
for line in String::from_utf8_lossy(&bytes).lines() {
|
|
if !line.trim().is_empty() {
|
|
tracing::warn!("{label} stderr: {line}");
|
|
}
|
|
}
|
|
}
|
|
|
|
match record.liveness() {
|
|
process::Liveness::Alive => said_unknown = false,
|
|
// Drain whatever it wrote on the way out before saying so.
|
|
//
|
|
// Progress, not "there were bytes": a process that died
|
|
// mid-line leaves a partial one that is re-read every pass and
|
|
// never completes, so waiting on a non-empty read would wait
|
|
// for ever and the exit would never be reported.
|
|
process::Liveness::Dead if complete > 0 => {}
|
|
process::Liveness::Dead => {
|
|
queue.lock().unwrap().close(&sink, "the session ended");
|
|
let detail = stderr_tail(&stderr_path);
|
|
if !detail.is_empty() {
|
|
let _ = sink.send(Event::Error {
|
|
message: format!("{label} exited:\n{detail}"),
|
|
});
|
|
}
|
|
let _ = sink.send(Event::Status {
|
|
state: SessionStatus::Exited,
|
|
});
|
|
process::clear(&session_dir);
|
|
return;
|
|
}
|
|
// The record is there and the machine will not say whether the
|
|
// process behind it is. Reported rather than guessed: calling
|
|
// it exited would invite starting a second one against the
|
|
// same conversation, which is the expensive mistake here.
|
|
// Kept polling, so it resolves itself if the answer comes back.
|
|
process::Liveness::Unknown => {
|
|
if !said_unknown {
|
|
said_unknown = true;
|
|
let _ = sink.send(Event::Status {
|
|
state: SessionStatus::Unknown,
|
|
});
|
|
}
|
|
}
|
|
}
|
|
tokio::time::sleep(POLL).await;
|
|
}
|
|
}
|
|
|
|
/// How many leading bytes of `bytes` form complete lines.
|
|
///
|
|
/// The offset only ever advances by this, which is what lets a read land
|
|
/// anywhere -- mid-line, mid-character -- without the reader losing its
|
|
/// place. See the call site for why the remainder is not kept.
|
|
///
|
|
/// A line ends at `\n` and at nothing else, deliberately. This stream is
|
|
/// JSONL: a record is a line, and something terminated by a bare `\r` is
|
|
/// not a record, so treating one as a line would hand `serde_json` a
|
|
/// fragment. The accepted consequence is that such a line is held here
|
|
/// forever rather than being reported -- and it is worth knowing what
|
|
/// that would look like, because it looks like nothing: the session goes
|
|
/// quiet with the process healthy, no error anywhere, and the cause is a
|
|
/// line splitter, which is not where anybody would look. A progress
|
|
/// indicator is the usual reason a program writes one (`\r` is how it
|
|
/// redraws in place), and the CLI has never written one here.
|
|
fn complete_lines(bytes: &[u8]) -> usize {
|
|
bytes
|
|
.iter()
|
|
.rposition(|byte| *byte == b'\n')
|
|
.map(|at| at + 1)
|
|
.unwrap_or(0)
|
|
}
|
|
|
|
/// One line of the CLI's output as events on the sink. `false` when the
|
|
/// session has been torn down and there is nothing left to send to.
|
|
fn translate_line(
|
|
line: &str,
|
|
session_dir: &Path,
|
|
state: &Arc<Mutex<Translator>>,
|
|
sink: &EventSink,
|
|
queue: &Arc<Mutex<Queue>>,
|
|
) -> bool {
|
|
let Ok(message) = serde_json::from_str::<Value>(line) else {
|
|
// By characters, not bytes: the CLI emits plenty of non-ASCII, and
|
|
// a byte slice that lands mid-character panics -- inside `follow`,
|
|
// which is the task reading this session's output, so the session
|
|
// would go permanently deaf with nothing on screen to say so. The
|
|
// other three truncations in this codebase (`setups.rs`,
|
|
// `translate.rs`, `import.rs`) already do it this way.
|
|
let shown: String = line.chars().take(200).collect();
|
|
tracing::warn!("unparseable claude output line: {shown}");
|
|
return true;
|
|
};
|
|
let opens_a_model_call = starts_a_model_call(&message);
|
|
let (events, new_session_id, before) = {
|
|
let mut state = state.lock().unwrap();
|
|
let before = state.session_id.clone();
|
|
let events = state.translate(&message);
|
|
let after = state.session_id.clone();
|
|
(events, if before != after { after } else { None }, before)
|
|
};
|
|
if let Some(session_id) = new_session_id {
|
|
write_resume_token(session_dir, &session_id);
|
|
}
|
|
// A turn the CLI began by itself, said one line earlier than anything
|
|
// else could say it.
|
|
//
|
|
// The CLI picks the conversation back up with nothing written to it --
|
|
// measured: a backgrounded `sleep` finished nine seconds after the
|
|
// turn's result and it started again unprompted. It announces that with
|
|
// an `init`, and the first assistant text follows about a second and a
|
|
// half later; until this, that second and a half read as idle, which is
|
|
// long enough to send a command into and have it read as text.
|
|
//
|
|
// `before.is_some()` is what separates this from the `init` at startup,
|
|
// which announces a session that is *waiting*. Our own `/clear` also
|
|
// produces one, and is excluded by `running` already being true --
|
|
// `local_command` set it before the line went out.
|
|
if opens_a_turn_by_itself(&message, before.is_some()) {
|
|
let started = {
|
|
let mut queue = queue.lock().unwrap();
|
|
let started = !queue.running && !queue.closed;
|
|
if started {
|
|
queue.running = true;
|
|
}
|
|
started
|
|
};
|
|
if started
|
|
&& sink
|
|
.send(Event::Status {
|
|
state: SessionStatus::Running,
|
|
})
|
|
.is_err()
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
// The steer is announced where the CLI opens the model call that read
|
|
// it, and the announcement goes out *before* that call's output, so
|
|
// the message sits above what it produced and below what it did not.
|
|
//
|
|
// This line carries no events of its own, which is what makes it the
|
|
// right place: everything the previous call produced -- its text, its
|
|
// tool calls, their results -- is already recorded above.
|
|
if opens_a_model_call && !announce_steers(queue, sink) {
|
|
return false;
|
|
}
|
|
for event in events {
|
|
// 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 {
|
|
state: SessionStatus::Idle
|
|
}
|
|
) {
|
|
// The case that must not be missed: a message written after
|
|
// the final model call of a turn has no later `message_start`
|
|
// to prove anything, so without this it would never be
|
|
// announced at all. The end of the turn is where it belongs
|
|
// anyway -- nothing above it came after the message.
|
|
if !announce_steers(queue, sink) {
|
|
return false;
|
|
}
|
|
queue.lock().unwrap().running = false;
|
|
}
|
|
if sink.send(event).is_err() {
|
|
return false; // session torn down
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
/// Whether this line is the CLI announcing work it started on its own.
|
|
///
|
|
/// `system/init` is how it says a conversation is beginning, and it sends
|
|
/// one in three cases: at startup, after a `/clear`, and when it picks the
|
|
/// conversation back up by itself. Only the third is a turn nobody here
|
|
/// asked for. `already_started` -- whether the translator had a session id
|
|
/// before this line -- rules out the first, and the caller's `running`
|
|
/// check rules out the second.
|
|
fn opens_a_turn_by_itself(message: &Value, already_started: bool) -> bool {
|
|
already_started
|
|
&& message.get("type").and_then(Value::as_str) == Some("system")
|
|
&& message.get("subtype").and_then(Value::as_str) == Some("init")
|
|
}
|
|
|
|
/// 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. Deliberately a wider set than what announces a steer
|
|
/// (see [`announce_steers`]): any sign of work proves a turn is running,
|
|
/// while only a `message_start` proves a line written a moment ago has
|
|
/// been read. `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
|
|
}
|
|
)
|
|
}
|
|
|
|
/// Records every message written since the last announcement, in the
|
|
/// order it was written. False means the session has been torn down.
|
|
///
|
|
/// Called from the two places that prove the CLI has consumed them: the
|
|
/// start of a new model call, and the end of the turn. Both are in
|
|
/// [`translate_line`], and the pair is the whole of the rule -- a steer
|
|
/// announced anywhere else lands above output that predates it.
|
|
fn announce_steers(queue: &Arc<Mutex<Queue>>, sink: &EventSink) -> bool {
|
|
let taken: Vec<(String, String, Vec<ImageRef>)> = {
|
|
let mut queue = queue.lock().unwrap();
|
|
queue.awaiting.drain(..).collect()
|
|
};
|
|
for (id, text, images) in taken {
|
|
if sink
|
|
.send(Event::MessageTaken {
|
|
id: Some(id),
|
|
text,
|
|
images,
|
|
})
|
|
.is_err()
|
|
{
|
|
return false;
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
/// The end of the stderr log, for an exit report a person reads.
|
|
///
|
|
/// Bounded because this is held in a message; trimmed of blank lines at
|
|
/// both ends because a shell's error ends with one, so anything reporting
|
|
/// "the last line" reports nothing at all. A failing `cd` cost an evening
|
|
/// to exactly that.
|
|
fn stderr_tail(path: &Path) -> String {
|
|
let Ok(text) = std::fs::read_to_string(path) else {
|
|
return String::new();
|
|
};
|
|
let kept: VecDeque<String> = text
|
|
.lines()
|
|
.rev()
|
|
.take(STDERR_LINES_KEPT)
|
|
.map(str::to_string)
|
|
.collect::<Vec<_>>()
|
|
.into_iter()
|
|
.rev()
|
|
.collect();
|
|
tail_of(&kept)
|
|
}
|
|
|
|
/// Creates the stdin fifo if it is not already there, and opens it
|
|
/// read-write for the process to inherit.
|
|
///
|
|
/// Read-write is the whole trick, and it is not an accident of
|
|
/// convenience: a fifo opened read-only delivers EOF as soon as the last
|
|
/// writer closes, so the process would exit the moment this server did --
|
|
/// which is exactly what leaving it running has to prevent. Holding it
|
|
/// open for writing as well means the process is its own last writer and
|
|
/// never sees the end of its input.
|
|
fn make_fifo(path: &Path) -> Result<std::fs::File> {
|
|
if !path.exists() {
|
|
let c_path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes())
|
|
.with_context(|| format!("{} is not a usable path", path.display()))?;
|
|
// SAFETY: a nul-terminated path this call only reads, and a mode
|
|
// with no bits the kernel can object to. Owner-only, like
|
|
// everything else in a session directory: this carries what the
|
|
// person typed.
|
|
let made = unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) };
|
|
if made != 0 {
|
|
return Err(std::io::Error::last_os_error())
|
|
.with_context(|| format!("creating the fifo {}", path.display()));
|
|
}
|
|
}
|
|
std::fs::OpenOptions::new()
|
|
.read(true)
|
|
.write(true)
|
|
.open(path)
|
|
.with_context(|| format!("opening the fifo {}", path.display()))
|
|
}
|
|
|
|
/// A fresh, empty, owner-only log for one of the process's output streams.
|
|
fn create_log(path: &Path) -> Result<std::fs::File> {
|
|
std::fs::OpenOptions::new()
|
|
.create(true)
|
|
.write(true)
|
|
.truncate(true)
|
|
.mode(0o600)
|
|
.open(path)
|
|
.with_context(|| format!("creating {}", path.display()))
|
|
}
|
|
|
|
pub(super) fn read_resume_token(session_dir: &Path) -> Option<String> {
|
|
let text = std::fs::read_to_string(session_dir.join(RESUME_FILE)).ok()?;
|
|
serde_json::from_str::<Value>(&text)
|
|
.ok()?
|
|
.get("sessionId")?
|
|
.as_str()
|
|
.map(String::from)
|
|
}
|
|
|
|
pub(super) fn write_resume_token(session_dir: &Path, session_id: &str) {
|
|
let path = session_dir.join(RESUME_FILE);
|
|
if let Err(err) = std::fs::write(&path, json!({"sessionId": session_id}).to_string()) {
|
|
tracing::error!("couldn't persist resume token to {}: {err}", path.display());
|
|
}
|
|
}
|
|
|
|
/// Reads an uploaded attachment into an API image content block.
|
|
fn attachment_block(session_dir: &Path, id: &str) -> Result<Value> {
|
|
// Ids are server-generated hex (see routes::upload_attachment); the
|
|
// check keeps a crafted "id" from naming an arbitrary file.
|
|
if !id
|
|
.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || c == '.' || c == '-')
|
|
{
|
|
anyhow::bail!("invalid attachment id");
|
|
}
|
|
let path = session_dir.join("attachments").join(id);
|
|
let bytes = std::fs::read(&path).with_context(|| format!("read {}", path.display()))?;
|
|
use base64::Engine;
|
|
Ok(json!({
|
|
"type": "image",
|
|
"source": {
|
|
"type": "base64",
|
|
// Ids carry the extension the upload was stored under, so an
|
|
// unrecognized one means a name this server didn't write.
|
|
"media_type": crate::media::media_type_for(id).unwrap_or("image/jpeg"),
|
|
"data": base64::engine::general_purpose::STANDARD.encode(bytes),
|
|
}
|
|
}))
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
/// Drives real CLI output lines through the reader and collects what
|
|
/// came out, which is the only way to check the wiring between "the
|
|
/// CLI said this" and "the transcript records that".
|
|
fn events_from_lines(lines: &[&str]) -> Vec<Event> {
|
|
let dir = tempfile::tempdir().expect("temp dir");
|
|
let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
|
|
let queue = Arc::new(Mutex::new(Queue::default()));
|
|
let (sink, mut out) = mpsc::unbounded_channel::<Event>();
|
|
for line in lines {
|
|
assert!(translate_line(line, dir.path(), &state, &sink, &queue));
|
|
}
|
|
drop(sink);
|
|
let mut events = Vec::new();
|
|
while let Ok(event) = out.try_recv() {
|
|
events.push(event);
|
|
}
|
|
events
|
|
}
|
|
|
|
/// Feeds lines through the reader, running `interject` between two of
|
|
/// them, and returns what came out.
|
|
///
|
|
/// The hook is what makes a steer testable at all: what matters is
|
|
/// not which events a line produces but *where* a message written
|
|
/// part-way through the stream ends up among them.
|
|
fn events_with_interjection(
|
|
lines: &[&str],
|
|
after: usize,
|
|
interject: impl FnOnce(&Arc<Mutex<Queue>>),
|
|
) -> Vec<Event> {
|
|
let dir = tempfile::tempdir().expect("temp dir");
|
|
let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
|
|
let queue = Arc::new(Mutex::new(Queue::default()));
|
|
let (sink, mut out) = mpsc::unbounded_channel::<Event>();
|
|
let mut interject = Some(interject);
|
|
for (i, line) in lines.iter().enumerate() {
|
|
assert!(translate_line(line, dir.path(), &state, &sink, &queue));
|
|
if i == after {
|
|
interject.take().expect("one interjection")(&queue);
|
|
}
|
|
}
|
|
drop(sink);
|
|
let mut events = Vec::new();
|
|
while let Ok(event) = out.try_recv() {
|
|
events.push(event);
|
|
}
|
|
events
|
|
}
|
|
|
|
/// One assistant message, streamed: two text deltas, then the
|
|
/// `tool_use` it ends with, then that call's result.
|
|
///
|
|
/// Written out rather than shortened because the point of both tests
|
|
/// below is the *order*, and the shape of a real turn is what makes
|
|
/// the order mean anything. Recorded from 2.1.237.
|
|
const STREAMED_CALL: &[&str] = &[
|
|
r#"{"type":"stream_event","event":{"type":"message_start"},"session_id":"s","parent_tool_use_id":null}"#,
|
|
r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Let me "}},"session_id":"s","parent_tool_use_id":null}"#,
|
|
r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"check that."}},"session_id":"s","parent_tool_use_id":null}"#,
|
|
r#"{"type":"assistant","message":{"role":"assistant","content":[{"type":"tool_use","id":"toolu_01","name":"Bash","input":{"command":"echo one"}}]},"parent_tool_use_id":null}"#,
|
|
r#"{"type":"user","message":{"role":"user","content":[{"tool_use_id":"toolu_01","type":"tool_result","content":"one","is_error":false}]},"parent_tool_use_id":null}"#,
|
|
];
|
|
|
|
/// A steer typed while an answer is streaming is recorded below that
|
|
/// answer's tool call and its result, not among them.
|
|
///
|
|
/// The message reaches the CLI immediately; what waits is saying so.
|
|
/// Everything the CLI emits after it was typed still belongs to a
|
|
/// model call that had not read it -- the rest of the text, the
|
|
/// `tool_use` the model had already committed to, the result that
|
|
/// came back. `message_start` is the first line that proves the next
|
|
/// call has it, so that is where the announcement goes.
|
|
#[test]
|
|
fn a_steer_is_recorded_below_the_call_that_had_not_read_it() {
|
|
let mut lines = STREAMED_CALL.to_vec();
|
|
lines.push(
|
|
r#"{"type":"stream_event","event":{"type":"message_start"},"session_id":"s","parent_tool_use_id":null}"#,
|
|
);
|
|
lines.push(
|
|
r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"Doing that instead."}},"session_id":"s","parent_tool_use_id":null}"#,
|
|
);
|
|
// Typed after the first delta, with the answer still arriving.
|
|
let events = events_with_interjection(&lines, 1, |queue| {
|
|
queue.lock().unwrap().awaiting.push_back((
|
|
"q1".into(),
|
|
"do the other one instead".into(),
|
|
Vec::new(),
|
|
))
|
|
});
|
|
|
|
let at = |find: fn(&Event) -> bool| {
|
|
events
|
|
.iter()
|
|
.position(find)
|
|
.unwrap_or_else(|| panic!("nothing matched in {events:?}"))
|
|
};
|
|
let taken = at(|e| matches!(e, Event::MessageTaken { .. }));
|
|
// Named, not just announced: the phone has a waiting bubble on screen for this message
|
|
// and clears the one with this id. Matching on the text instead would clear the wrong
|
|
// bubble whenever the same thing was sent twice.
|
|
assert!(
|
|
matches!(
|
|
&events[taken],
|
|
Event::MessageTaken { id: Some(id), .. } if id == "q1"
|
|
),
|
|
"an announcement must name the queue entry it resolves: {events:?}"
|
|
);
|
|
assert!(
|
|
taken > at(|e| matches!(e, Event::ToolStart { .. })),
|
|
"a steer must not sit above a call the model had already made: {events:?}"
|
|
);
|
|
assert!(
|
|
taken > at(|e| matches!(e, Event::ToolEnd { .. })),
|
|
"a steer must not sit above the result of that call: {events:?}"
|
|
);
|
|
assert_eq!(
|
|
events
|
|
.iter()
|
|
.filter(|e| matches!(e, Event::AssistantText { .. }))
|
|
.count(),
|
|
3,
|
|
"the streamed answer must stay whole: {events:?}"
|
|
);
|
|
}
|
|
|
|
/// A steer written after the turn's last model call is still recorded.
|
|
///
|
|
/// Nothing further is coming, so no `message_start` will ever prove
|
|
/// it was read -- and a message that is only recorded when announced
|
|
/// would otherwise vanish, leaving a phone drawing it as still
|
|
/// waiting forever. The end of the turn is also where it belongs:
|
|
/// nothing above it happened after it was typed.
|
|
#[test]
|
|
fn a_steer_with_no_model_call_left_is_recorded_at_the_end_of_the_turn() {
|
|
let mut lines = STREAMED_CALL.to_vec();
|
|
lines.push(
|
|
r#"{"type":"result","subtype":"success","usage":{"input_tokens":1,"output_tokens":1}}"#,
|
|
);
|
|
// Typed after the tool result, with only the turn's end to come.
|
|
let events = events_with_interjection(&lines, 4, |queue| {
|
|
queue
|
|
.lock()
|
|
.unwrap()
|
|
.awaiting
|
|
.push_back(("q2".into(), "never mind".into(), Vec::new()))
|
|
});
|
|
|
|
let taken = events
|
|
.iter()
|
|
.position(|e| matches!(e, Event::MessageTaken { .. }))
|
|
.unwrap_or_else(|| panic!("a steer must never be dropped: {events:?}"));
|
|
let idle = events
|
|
.iter()
|
|
.position(|e| {
|
|
matches!(
|
|
e,
|
|
Event::Status {
|
|
state: SessionStatus::Idle
|
|
}
|
|
)
|
|
})
|
|
.unwrap_or_else(|| panic!("expected the turn to end: {events:?}"));
|
|
assert!(
|
|
taken < idle,
|
|
"the steer belongs inside the turn it was typed into: {events:?}"
|
|
);
|
|
}
|
|
|
|
/// The divider comes from the CLI announcing the reset, not from an
|
|
/// `init` arriving.
|
|
///
|
|
/// Measured against 2.1.237: `/clear` emits `conversation_reset`,
|
|
/// then a fresh `init` carrying a new session id. Watching the id be
|
|
/// replaced would work, but it reads the event through one of its
|
|
/// side effects; the announcement says so directly and arrives first,
|
|
/// so the divider lands above the new conversation rather than below
|
|
/// its opening line.
|
|
#[test]
|
|
fn a_conversation_reset_is_what_records_a_clear() {
|
|
let events = events_from_lines(&[
|
|
r#"{"type":"system","subtype":"init","session_id":"first","tools":[],"model":"claude-haiku-4-5-20251001"}"#,
|
|
r#"{"type":"conversation_reset","session_id":"first"}"#,
|
|
r#"{"type":"system","subtype":"init","session_id":"second","tools":[],"model":"claude-haiku-4-5-20251001"}"#,
|
|
]);
|
|
assert_eq!(
|
|
events.iter().filter(|e| **e == Event::Cleared).count(),
|
|
1,
|
|
"got {events:?}"
|
|
);
|
|
}
|
|
|
|
/// An `init` on its own never records a clear, whatever id it carries.
|
|
///
|
|
/// Three ways one arrives and none of them is a cleared conversation:
|
|
/// the first init of a session, the one a compaction re-announces
|
|
/// carrying the *same* id, and the one that follows a resume. Reading
|
|
/// any of them as a clear would open sessions with a divider
|
|
/// announcing something that never happened, or draw one on top of a
|
|
/// compaction's own mark and tell the reader the conversation had been
|
|
/// dropped when it had been summarised.
|
|
#[test]
|
|
fn an_init_alone_is_never_a_clear() {
|
|
for ids in [["first", "first"], ["first", "second"]] {
|
|
let events = events_from_lines(&[
|
|
&format!(
|
|
r#"{{"type":"system","subtype":"init","session_id":"{}","tools":[],"model":"claude-haiku-4-5-20251001"}}"#,
|
|
ids[0]
|
|
),
|
|
&format!(
|
|
r#"{{"type":"system","subtype":"init","session_id":"{}","tools":[],"model":"claude-haiku-4-5-20251001"}}"#,
|
|
ids[1]
|
|
),
|
|
]);
|
|
assert!(!events.contains(&Event::Cleared), "{ids:?} gave {events:?}");
|
|
}
|
|
}
|
|
|
|
/// The failure this exists for: a shell's complaint ends with a blank
|
|
/// line, so reporting "the last line of stderr" reported nothing, and
|
|
/// the phone showed a bare exit status while the reason sat in the
|
|
/// server's log.
|
|
#[test]
|
|
fn the_report_keeps_the_message_and_not_the_blank_line_after_it() {
|
|
let fish_cd_failure = [
|
|
"cd: The directory '~/repos/ai-app' does not exist",
|
|
"",
|
|
"embedded:functions/cd.fish (line 26): ",
|
|
" builtin cd $argv",
|
|
" ^",
|
|
"in function 'cd' with arguments '~/repos/ai-app'",
|
|
"",
|
|
];
|
|
let kept: VecDeque<String> = fish_cd_failure.iter().map(|l| l.to_string()).collect();
|
|
|
|
let report = tail_of(&kept);
|
|
assert!(
|
|
report.starts_with("cd: The directory"),
|
|
"the complaint leads: {report}",
|
|
);
|
|
assert!(
|
|
report.ends_with("'~/repos/ai-app'"),
|
|
"the trailing blank is trimmed: {report:?}",
|
|
);
|
|
// The blank *between* lines is part of the message and stays.
|
|
assert!(report.contains("does not exist\n\nembedded:"), "{report:?}");
|
|
}
|
|
|
|
/// Nothing to say is said as nothing, so the caller can tell the two
|
|
/// apart and print just the exit status.
|
|
#[test]
|
|
fn stderr_that_is_only_blank_lines_reports_as_empty() {
|
|
let kept: VecDeque<String> = ["", " ", ""].iter().map(|l| l.to_string()).collect();
|
|
assert_eq!(tail_of(&kept), "");
|
|
assert_eq!(tail_of(&VecDeque::new()), "");
|
|
}
|
|
|
|
#[test]
|
|
fn a_stream_read_in_arbitrary_chunks_yields_each_line_once() {
|
|
// The property the reading position has to have: however a write
|
|
// is split -- mid-line, or mid-character -- every line comes out
|
|
// exactly once and in order. Chunked at every prime-ish size so
|
|
// the cuts land in different places, including inside the
|
|
// multi-byte character.
|
|
let stream = "{\"a\":1}\n{\"b\":\"caf\u{e9}\"}\n{\"c\":3}\n";
|
|
for chunk in [1usize, 2, 3, 5, 7, 11, 1000] {
|
|
let mut offset = 0usize;
|
|
let mut lines: Vec<String> = Vec::new();
|
|
let bytes = stream.as_bytes();
|
|
let mut available = 0usize;
|
|
while available < bytes.len() {
|
|
available = (available + chunk).min(bytes.len());
|
|
// What a read from the recorded offset returns: the file
|
|
// as far as it has been written, from where we left off.
|
|
let unread = &bytes[offset..available];
|
|
let complete = complete_lines(unread);
|
|
for line in String::from_utf8_lossy(&unread[..complete]).lines() {
|
|
lines.push(line.to_string());
|
|
}
|
|
offset += complete;
|
|
}
|
|
assert_eq!(offset, bytes.len(), "chunk {chunk} left bytes unread");
|
|
assert_eq!(
|
|
lines,
|
|
vec!["{\"a\":1}", "{\"b\":\"caf\u{e9}\"}", "{\"c\":3}"],
|
|
"chunk {chunk}"
|
|
);
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn an_incomplete_line_advances_nothing() {
|
|
// Nothing to do yet, and crucially the position does not move --
|
|
// so a crash here re-reads the line rather than skipping it.
|
|
assert_eq!(complete_lines(b"{\"partial\": tru"), 0);
|
|
assert_eq!(complete_lines(b""), 0);
|
|
// And a complete line followed by a partial one advances only past
|
|
// the complete one.
|
|
assert_eq!(complete_lines(b"done\nhalf"), 5);
|
|
}
|
|
|
|
#[test]
|
|
fn closing_the_queue_reports_what_was_never_read() {
|
|
let (sink, mut received) = mpsc::unbounded_channel();
|
|
let mut queue = Queue {
|
|
running: true,
|
|
..Queue::default()
|
|
};
|
|
queue
|
|
.awaiting
|
|
.push_back(("q1".into(), "first".into(), Vec::new()));
|
|
queue
|
|
.awaiting
|
|
.push_back(("q2".into(), "second".into(), Vec::new()));
|
|
queue.close(&sink, "the session ended");
|
|
|
|
// Named rather than counted, because these never reached the
|
|
// transcript: this message is the only record they existed.
|
|
let Some(Event::Error { message }) = received.try_recv().ok() else {
|
|
panic!("closing a queue holding messages must report them");
|
|
};
|
|
assert!(message.contains("2 queued messages"), "{message}");
|
|
assert!(message.starts_with("the session ended"), "{message}");
|
|
assert!(
|
|
message.contains("first") && message.contains("second"),
|
|
"{message}"
|
|
);
|
|
|
|
// And the flag is cleared, so a later message is refused with a
|
|
// reason rather than queued behind a turn that will never end.
|
|
assert!(!queue.running);
|
|
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();
|
|
let mut queue = Queue::default();
|
|
queue.close(&sink, "the session ended");
|
|
// A session that exits with nothing held has lost nothing, and an
|
|
// error saying so would be noise on every ordinary exit.
|
|
assert!(received.try_recv().is_err());
|
|
assert!(queue.closed);
|
|
}
|
|
}
|