iris-ai
bd9596d782
A multimodal model is loaded with the `mmproj` found beside its weights -- which is how a repository publishes the pair -- and an attached image rides in the request as an `image_url` data URI, so it reaches a model on another machine without the file going there. Nothing is done for a model without a projector: no captioning, no OCR, no second model. Whether a session takes pictures is measured rather than assumed: `/props`'s `modalities.vision` from the server that loaded the model, in three states, because a model still coming off disk has genuinely not said. Unknown is offered rather than refused -- a control withheld because nobody could ask goes missing from sessions that would have taken it. The answer reaches the phone twice per model as `Event::Images`, so the photo button is withdrawn the moment a model with vision is left rather than at whatever later point the session row is fetched again. A message carrying an image a model cannot read is stopped rather than stripped: `llama-server` refuses the whole request over one image part, and a message sent without its picture would be answered as though the picture had never been mentioned. The phone will not attach one, and the driver refuses it again at the three moments the answer can first exist -- at the door, when a message queued behind a loading model is read, and at the tool boundary a steer enters by. An earlier turn's image folds into a line of words for a model without vision, so switching a conversation onto one does not end it. A projector is filtered out of the models a provider *offers*, since a session started on one is a server that cannot load it; it stays in the machine's own model list, where a file on a disk is managed. Verified against ggml-org/SmolVLM-256M-Instruct-GGUF, local and over ssh: "In this picture there is a red circle." Switching that session to Qwen3-0.6B reports `refused`, refuses the next picture with the reason, and still answers an ordinary message.
1756 lines
76 KiB
Rust
1756 lines
76 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::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::{
|
|
AttachmentRef, BackgroundTask, Driver, Event, EventSink, SessionStatus, Unqueued,
|
|
};
|
|
use super::process;
|
|
use super::subagent::Subagents;
|
|
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, and bounded
|
|
/// because this is held per session for the life of the process.
|
|
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
|
|
/// anything reporting "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, since respawning with `--resume <id>` picks the
|
|
/// conversation back up. 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 steering a running turn is the
|
|
/// point of this app. An earlier version 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.
|
|
///
|
|
/// 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, until the CLI opens
|
|
/// the next model call -- see [`translate::starts_a_model_call`].
|
|
///
|
|
/// The proof has to be the model call and not the output. 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.
|
|
#[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<AttachmentRef>)>,
|
|
/// 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 *announced*, each later one vanished silently.
|
|
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.
|
|
///
|
|
/// Each is also *resolved*, with the same `MessageDropped` that tapping the
|
|
/// bubble produces -- otherwise the bubble sat there for good, waiting on a
|
|
/// `UserMessage` that is exactly what is not coming.
|
|
fn close(&mut self, sink: &EventSink, why: &str) {
|
|
self.closed = true;
|
|
self.running = false;
|
|
let lost: Vec<(String, String)> = self
|
|
.awaiting
|
|
.drain(..)
|
|
.map(|(id, text, _)| (id, 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.iter()
|
|
.map(|(_, text)| text.as_str())
|
|
.collect::<Vec<_>>()
|
|
.join(" / ")
|
|
),
|
|
});
|
|
for (id, _) in lost {
|
|
let _ = sink.send(Event::MessageDropped { id });
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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 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 already arrives 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.
|
|
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
|
|
/// 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. 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,
|
|
subagents: Arc<Subagents>,
|
|
initial_status: SessionStatus,
|
|
) -> Result<Self> {
|
|
let state = Arc::new(Mutex::new(Translator::new(
|
|
session_dir.to_path_buf(),
|
|
subagents,
|
|
)));
|
|
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 died with the connection.
|
|
// Nothing was ever recorded for one, so this answers "no" without
|
|
// needing to know that.
|
|
//
|
|
// `started_here` is whether this launch *started* a process or picked
|
|
// one up; the two owe the session different things.
|
|
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.
|
|
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)?
|
|
}
|
|
};
|
|
if !started_here && matches!(initial_status, SessionStatus::Idle | SessionStatus::Waiting) {
|
|
state.lock().unwrap().mark_settled();
|
|
}
|
|
|
|
// A process this driver has just started has been asked for nothing,
|
|
// which is what idle means. Said here because nothing else will: the
|
|
// CLI writes not one line until it is given work, so a session whose
|
|
// transcript last recorded `Exited` would keep that word -- and
|
|
// `Exited` refuses every command and invites starting a second CLI
|
|
// against a conversation that already has one.
|
|
//
|
|
// From the driver rather than the manager, and before `follow` is
|
|
// spawned, so it cannot overtake the exit `follow` reports for a
|
|
// process that dies immediately. Adopting says nothing, because a
|
|
// process already running may be mid-turn and the transcript's last
|
|
// word is the better answer until its output says otherwise.
|
|
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.
|
|
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()),
|
|
));
|
|
|
|
let driver = Self {
|
|
sink,
|
|
queue,
|
|
to_child,
|
|
state,
|
|
session_dir: session_dir.to_path_buf(),
|
|
reading,
|
|
};
|
|
if !started_here {
|
|
// Claude 2.1.261 sends a full background-task snapshot after a
|
|
// repeated initialize. That repairs a completion edge missed by a
|
|
// backend that was down while the CLI kept running; older CLIs
|
|
// accept the request and simply send no snapshot.
|
|
driver.send_control(json!({"subtype": "initialize"}), None);
|
|
}
|
|
Ok(driver)
|
|
}
|
|
|
|
/// 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.
|
|
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);
|
|
}
|
|
// Launch-only: see `SessionConfig::effort`. Omitted entirely when
|
|
// unset, so the CLI's own default is what an unchosen session gets
|
|
// rather than a level this app decided to call the default.
|
|
if let Some(effort) = &meta.effort {
|
|
push("--effort", effort);
|
|
}
|
|
// Named at birth, so this session is the same session in the CLI's own
|
|
// picker and in what other agents see.
|
|
//
|
|
// Only when we are 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 typing
|
|
// in it. `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.
|
|
//
|
|
// The CLI is asymmetric about that mode. It will *launch* in
|
|
// `bypassPermissions` on `--permission-mode` alone, but refuses to
|
|
// *switch* into it later ("the session was not launched with
|
|
// --dangerously-skip-permissions"), so the phone's mode picker offered
|
|
// a mode that could not be picked on every session not given it at
|
|
// birth. Since the mode is already reachable at spawn, this grants
|
|
// nothing that was being withheld.
|
|
//
|
|
// Measured against 2.1.237 both ways round. Note it is the `--allow-`
|
|
// form; `--dangerously-skip-permissions` turns it on for everything,
|
|
// which would take the choice away from whoever holds 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 = process::make_fifo(&session_dir.join(STDIN_FIFO))?;
|
|
let stdout = process::create_log(&session_dir.join(STDOUT_LOG))?;
|
|
let stderr = process::create_log(&session_dir.join(STDERR_LOG))?;
|
|
|
|
let program = provider.program();
|
|
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
|
|
// `follow` decides what the session is doing, because after a restart
|
|
// there is no `Child` to wait on.
|
|
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, 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.
|
|
///
|
|
/// Nothing is emitted about the command itself: the manager has already
|
|
/// said it was sent, and the CLI announces what it does.
|
|
fn local_command(&self, text: String) {
|
|
let mut queue = self.queue.lock().unwrap();
|
|
// The same check `send_user_message` makes: a line written into a fifo
|
|
// nothing is reading goes nowhere and looks exactly like one that
|
|
// arrived. What this catches is the process going away between
|
|
// `Commands::submit`'s 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.
|
|
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`]. `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.
|
|
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 background_tasks(&self) -> Option<Vec<BackgroundTask>> {
|
|
self.state.lock().unwrap().background_tasks()
|
|
}
|
|
|
|
fn send_user_message(&self, text: String, attachments: Vec<AttachmentRef>) {
|
|
let mut content = Vec::new();
|
|
// An image goes into the message itself; the model looks at it. Any
|
|
// other file stays where the upload put it and the message says where,
|
|
// because the CLI can read a file by path and a model cannot be handed
|
|
// a trace any other way.
|
|
let mut files = Vec::new();
|
|
for id in &attachments {
|
|
let sent = if crate::media::media_type_for(id).is_some() {
|
|
attachment_block(&self.session_dir, id).map(|block| content.push(block))
|
|
} else {
|
|
attachment_path(&self.session_dir, id).map(|path| files.push(path))
|
|
};
|
|
if let Err(err) = sent {
|
|
let _ = self.sink.send(Event::Error {
|
|
message: format!("attachment {id} couldn't be sent: {err:#}"),
|
|
});
|
|
}
|
|
}
|
|
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;
|
|
}
|
|
// Built under the lock because whether this is a steer decides what the
|
|
// CLI is told, and the answer must be the same one the branch below acts
|
|
// on -- see `super::driver::message_body`.
|
|
let body = super::driver::message_body(&text, &files, queue.running);
|
|
if !body.is_empty() {
|
|
content.push(json!({"type": "text", "text": body}));
|
|
}
|
|
let line =
|
|
json!({"type": "user", "message": {"role": "user", "content": content}}).to_string();
|
|
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, which is the one thing that must
|
|
// not be left to the phone to remember: it drew the bubble from its
|
|
// own state, so leaving the session showed nothing pending while a
|
|
// message was still queued.
|
|
let id = super::random_hex();
|
|
queue
|
|
.awaiting
|
|
.push_back((id.clone(), text.clone(), attachments.clone()));
|
|
drop(queue);
|
|
let _ = self.sink.send(Event::MessageQueued {
|
|
id,
|
|
text,
|
|
attachments,
|
|
});
|
|
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`.
|
|
let _ = self.sink.send(Event::MessageTaken {
|
|
id: None,
|
|
text,
|
|
attachments,
|
|
});
|
|
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.
|
|
/// A line in the fifo cannot be recalled.
|
|
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.
|
|
fn interrupt(&self) {
|
|
// Recorded before the request goes out, so the result it produces reads
|
|
// as the stop somebody asked for rather than as a failure.
|
|
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. It rides the same channel as
|
|
// `/compact` and starts a turn the same way, so the same bookkeeping
|
|
// applies; what it means is the CLI's business.
|
|
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 name with a
|
|
// newline 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 `conversation_reset`, 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.
|
|
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, keeps writing its log, and keeps its record -- which
|
|
// is how the next run of this server finds it.
|
|
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 -- 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:
|
|
/// everything before it is already in the transcript, so a server coming back
|
|
/// picks up exactly where the last one stopped.
|
|
#[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 was logged by whichever run of this
|
|
// server was watching, so a reattach starts at the end of 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. The status
|
|
// is `Unknown` rather than `Exited` because the process may well
|
|
// still be running; what has failed is hearing 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 read again next pass.
|
|
// Deliberately *not* held in memory between passes: the offset would
|
|
// then have to point behind the bytes being held. It is also what makes
|
|
// the position crash-safe.
|
|
//
|
|
// Counted in bytes rather than on a decoded string: a read can cut a
|
|
// multi-byte character in half, and the replacement character 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 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.
|
|
process::Liveness::Dead if complete > 0 => {}
|
|
process::Liveness::Dead => {
|
|
queue.lock().unwrap().close(&sink, "the session ended");
|
|
if !process::stopping(&session_dir) {
|
|
let detail = stderr_tail(&stderr_path);
|
|
// A token the CLI will not accept leaves the chat
|
|
// *unrecoverable* rather than merely failed: every later
|
|
// start passes the same `--resume` and dies the same way.
|
|
// So it is forgotten, and `Cleared` marks where the model's
|
|
// context stopped -- as for a Codex thread with no rollout.
|
|
let refused =
|
|
missing_conversation(&detail) && read_resume_token(&session_dir).is_some();
|
|
// Both the message and the divider follow the removal
|
|
// rather than the diagnosis: one that failed leaves the
|
|
// session as stuck as it was.
|
|
let recovered = refused && forget_resume_token(&session_dir);
|
|
if !detail.is_empty() {
|
|
// The CLI's sentence reads like the chat is lost when
|
|
// one more message is all it needs, and this is the
|
|
// only place anyone sees it. Its words stay on top,
|
|
// being what a person would search for.
|
|
let message = if recovered {
|
|
format!(
|
|
"{label} exited:\n{detail}\n\nThat conversation is gone from the \
|
|
CLI, so this session has stopped trying to resume it. Send a \
|
|
message to carry on in a new one -- everything above is kept, \
|
|
but the model starts without it."
|
|
)
|
|
} else {
|
|
format!("{label} exited:\n{detail}")
|
|
};
|
|
let _ = sink.send(Event::Error { message });
|
|
}
|
|
if recovered {
|
|
let _ = sink.send(Event::Cleared);
|
|
}
|
|
}
|
|
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. Kept polling, so it resolves itself.
|
|
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.
|
|
///
|
|
/// A line ends at `\n` and at nothing else, deliberately: this stream is JSONL,
|
|
/// so something terminated by a bare `\r` is not a record and treating one as a
|
|
/// line would hand `serde_json` a fragment. The accepted consequence is that
|
|
/// such a line is held here forever, and it is worth knowing what that looks
|
|
/// like, because it looks like nothing: the session goes quiet with the process
|
|
/// healthy and no error anywhere. 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`, so the
|
|
// session would go permanently deaf with nothing on screen to say so.
|
|
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 parent_running = queue.lock().unwrap().running;
|
|
let (events, new_session_id, before) = {
|
|
let mut state = state.lock().unwrap();
|
|
if parent_running {
|
|
state.mark_running();
|
|
}
|
|
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 read as idle, which is long enough to send a command into
|
|
// and have it read as text.
|
|
//
|
|
// `before.is_some()` separates this from the `init` at startup. 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 *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.
|
|
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.
|
|
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;
|
|
}
|
|
// Either status the turn can end in -- see `SessionStatus::Waiting`.
|
|
// A turn that ended with a subagent still running is over for this
|
|
// queue's purposes: the CLI will read the next message, and holding
|
|
// one back until the subagent reported would sit on it indefinitely.
|
|
if matches!(
|
|
event,
|
|
Event::Status {
|
|
state: SessionStatus::Idle | SessionStatus::Waiting
|
|
}
|
|
) {
|
|
// 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.
|
|
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
|
|
/// 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`
|
|
/// 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: a compaction the CLI decided on itself, a session adopted mid-turn,
|
|
/// a message that reached the conversation by another route. In all of them the
|
|
/// CLI is plainly working and the only thing that would have said so is a
|
|
/// `Running` nobody sent, so the session reads as idle until the turn ends.
|
|
///
|
|
/// Deliberately a wider set than what announces a steer: any sign of work
|
|
/// proves a turn is running, while only a `message_start` proves a line written
|
|
/// a moment ago has been read.
|
|
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. 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<AttachmentRef>)> = {
|
|
let mut queue = queue.lock().unwrap();
|
|
queue.awaiting.drain(..).collect()
|
|
};
|
|
for (id, text, attachments) in taken {
|
|
if sink
|
|
.send(Event::MessageTaken {
|
|
id: Some(id),
|
|
text,
|
|
attachments,
|
|
})
|
|
.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 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)
|
|
}
|
|
|
|
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());
|
|
}
|
|
}
|
|
|
|
/// Drop a resume token the CLI has refused, so the next start makes a session
|
|
/// rather than repeating the failure. Answers whether it is really gone: the
|
|
/// caller promises somebody the session has stopped resuming, and a failed
|
|
/// removal would make that a promise this server cannot keep.
|
|
fn forget_resume_token(session_dir: &Path) -> bool {
|
|
let path = session_dir.join(RESUME_FILE);
|
|
match std::fs::remove_file(&path) {
|
|
Ok(()) => true,
|
|
Err(err) if err.kind() == std::io::ErrorKind::NotFound => true,
|
|
Err(err) => {
|
|
tracing::error!(
|
|
"couldn't forget the refused resume token at {}: {err}",
|
|
path.display()
|
|
);
|
|
false
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Whether the CLI refused `--resume` because the token names nothing it has.
|
|
///
|
|
/// Two sentences say it and both recover the same way; matched on the phrases
|
|
/// that carry the fact, since the rest names the id. Measured against 2.1.237:
|
|
/// a valid UUID with nothing behind it gives the first, a non-UUID matching no
|
|
/// title the second.
|
|
fn missing_conversation(detail: &str) -> bool {
|
|
let detail = detail.to_ascii_lowercase();
|
|
detail.contains("no conversation found with session id")
|
|
|| (detail.contains("--resume requires a valid session id")
|
|
&& detail.contains("does not match any session title"))
|
|
}
|
|
|
|
/// Where an uploaded attachment is, as a path the CLI can be told.
|
|
///
|
|
/// Absolute, because the CLI's working directory is the session's and the
|
|
/// attachments are not in it. Refused rather than resolved when the id is not
|
|
/// one this server would have written, so a crafted id cannot name a file
|
|
/// outside the session.
|
|
fn attachment_path(session_dir: &Path, id: &str) -> Result<PathBuf> {
|
|
let path = super::driver::attachment_path(session_dir, id)?;
|
|
// A file copied to the session's own machine is named where it landed there
|
|
// -- `routes::upload_attachment` writes that down beside it -- because the
|
|
// path has to be one the CLI can open, not one this server can.
|
|
let shipped = path.with_file_name(format!("{id}.remote"));
|
|
if let Ok(remote) = std::fs::read_to_string(&shipped) {
|
|
return Ok(PathBuf::from(remote.trim()));
|
|
}
|
|
std::fs::canonicalize(&path).with_context(|| format!("find {}", path.display()))
|
|
}
|
|
|
|
/// Reads an uploaded image into an API image content block.
|
|
fn attachment_block(session_dir: &Path, id: &str) -> Result<Value> {
|
|
let path = attachment_path(session_dir, 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::*;
|
|
|
|
#[test]
|
|
fn an_attachment_is_named_where_the_cli_can_open_it() {
|
|
let dir = tempfile::tempdir().expect("temp dir");
|
|
let attachments = dir.path().join("attachments");
|
|
std::fs::create_dir(&attachments).unwrap();
|
|
std::fs::write(attachments.join("ab12-x.bin"), b"x").unwrap();
|
|
assert_eq!(
|
|
attachment_path(dir.path(), "ab12-x.bin").unwrap(),
|
|
attachments.join("ab12-x.bin").canonicalize().unwrap()
|
|
);
|
|
// Shipped to the session's machine: the path there, not here.
|
|
std::fs::write(
|
|
attachments.join("ab12-x.bin.remote"),
|
|
"/home/t/in/ab12-x.bin\n",
|
|
)
|
|
.unwrap();
|
|
assert_eq!(
|
|
attachment_path(dir.path(), "ab12-x.bin").unwrap(),
|
|
PathBuf::from("/home/t/in/ab12-x.bin")
|
|
);
|
|
assert!(attachment_path(dir.path(), "../config.ron").is_err());
|
|
assert!(attachment_path(dir.path(), "missing.bin").is_err());
|
|
}
|
|
|
|
/// 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(),
|
|
Arc::new(Subagents::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(),
|
|
Arc::new(Subagents::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 emitted after it was typed still belongs to a model call that
|
|
/// had not read it. `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 for this
|
|
// message and clears the one with this id. Matching on the text 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 only recorded when announced would vanish, leaving
|
|
/// a phone drawing it as waiting forever.
|
|
#[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 a side effect; the
|
|
/// announcement says so directly and arrives first, so the divider lands
|
|
/// above the new conversation.
|
|
#[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 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 as a clear would tell
|
|
/// the reader a 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(),
|
|
Arc::new(Subagents::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.
|
|
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(),
|
|
Arc::new(Subagents::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);
|
|
}
|
|
|
|
/// Both refusals verbatim from 2.1.237, and the exits that must *not* be
|
|
/// read as one: a session that merely failed still has a conversation, and
|
|
/// forgetting its token would discard the model's context for nothing.
|
|
#[test]
|
|
fn both_refused_resume_tokens_are_recognised() {
|
|
assert!(missing_conversation(
|
|
"No conversation found with session ID: a1c6c855-86cd-4f47-8b50-eaea85be3579"
|
|
));
|
|
assert!(missing_conversation(
|
|
"Error: --resume requires a valid session ID or session title when used with \
|
|
--print. Usage: claude -p --resume <session-id|title>. Provided value \
|
|
\"not-a-real-session\" is not a UUID and does not match any session title."
|
|
));
|
|
assert!(!missing_conversation("Error: connection closed"));
|
|
assert!(!missing_conversation(
|
|
"Credit balance is too low to run this request"
|
|
));
|
|
// The usage line alone is a different complaint: `--resume` was passed
|
|
// wrongly, not given a token that named nothing.
|
|
assert!(!missing_conversation(
|
|
"Error: --resume requires a valid session ID or session title when used with --print."
|
|
));
|
|
}
|
|
|
|
/// The composition the reader performs, not just the matcher: `stderr_tail`
|
|
/// both truncates and trims, so the phrase can survive the CLI and still
|
|
/// not reach `missing_conversation`.
|
|
#[test]
|
|
fn the_refusal_survives_the_stderr_tail_that_carries_it() {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
let path = dir.path().join("stderr.log");
|
|
// Verbatim from 2.1.237, trailing newline included -- a shell's error
|
|
// ends with one, which is exactly what `tail_of` exists to trim.
|
|
std::fs::write(
|
|
&path,
|
|
"No conversation found with session ID: a1c6c855-86cd-4f47-8b50-eaea85be3579\n",
|
|
)
|
|
.expect("write stderr");
|
|
assert!(missing_conversation(&stderr_tail(&path)));
|
|
|
|
// And still found when noise precedes it, since the refusal is the last
|
|
// thing the CLI writes and the tail is taken from the end.
|
|
let mut noisy = "some earlier warning\n".repeat(STDERR_LINES_KEPT * 2);
|
|
noisy.push_str("No conversation found with session ID: a1c6c855\n");
|
|
std::fs::write(&path, noisy).expect("write noisy stderr");
|
|
assert!(missing_conversation(&stderr_tail(&path)));
|
|
}
|
|
|
|
/// A dead process, a refusal in its stderr log, and a token on disk --
|
|
/// driven through the reader that performs the recovery rather than through
|
|
/// its parts, because the wiring is what the parts cannot check.
|
|
async fn exit_with_stderr(stderr: &str) -> (tempfile::TempDir, Vec<Event>) {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
// Empty but present: an unreadable stdout is a different path that
|
|
// returns before any of this.
|
|
std::fs::write(dir.path().join(STDOUT_LOG), "").expect("stdout log");
|
|
std::fs::write(dir.path().join(STDERR_LOG), stderr).expect("stderr log");
|
|
write_resume_token(dir.path(), "a1c6c855-86cd-4f47-8b50-eaea85be3579");
|
|
|
|
let (sink, mut events) = mpsc::unbounded_channel();
|
|
follow(
|
|
dir.path().to_path_buf(),
|
|
// No `/proc` entry, so `liveness()` reads `Dead` -- the state this
|
|
// arm exists for, without having to kill anything.
|
|
process::Record {
|
|
pid: u32::MAX,
|
|
started: 0,
|
|
detail: process::Detail::Stdio { stdout_read: 0 },
|
|
},
|
|
0,
|
|
Arc::new(Mutex::new(Translator::new(
|
|
dir.path().to_path_buf(),
|
|
Arc::new(Subagents::new(dir.path().to_path_buf())),
|
|
))),
|
|
sink,
|
|
Arc::new(Mutex::new(Queue::default())),
|
|
Arc::new(AtomicBool::new(true)),
|
|
"claude-cli on vm".to_string(),
|
|
)
|
|
.await;
|
|
let seen = std::iter::from_fn(|| events.try_recv().ok()).collect();
|
|
(dir, seen)
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn a_session_whose_conversation_vanished_is_left_able_to_start_again() {
|
|
let (dir, seen) = exit_with_stderr(
|
|
"No conversation found with session ID: a1c6c855-86cd-4f47-8b50-eaea85be3579\n",
|
|
)
|
|
.await;
|
|
|
|
let reported = seen
|
|
.iter()
|
|
.find_map(|event| match event {
|
|
Event::Error { message } => Some(message.clone()),
|
|
_ => None,
|
|
})
|
|
.expect("the exit is reported");
|
|
// The CLI's own words, and then what to do about them.
|
|
assert!(
|
|
reported.contains("No conversation found with session ID"),
|
|
"{reported}"
|
|
);
|
|
assert!(
|
|
reported.contains("Send a message to carry on"),
|
|
"{reported}"
|
|
);
|
|
// The divider, so the transcript says where the model's context ended.
|
|
assert!(seen.contains(&Event::Cleared), "{seen:?}");
|
|
// And the point of all of it: the next start has no token to repeat.
|
|
assert_eq!(read_resume_token(dir.path()), None);
|
|
}
|
|
|
|
/// The same reader on an exit that is *not* a refusal. The token is what
|
|
/// the session is still worth resuming from, so it has to survive.
|
|
#[tokio::test]
|
|
async fn an_ordinary_failure_keeps_the_token_it_can_still_resume_from() {
|
|
let (dir, seen) = exit_with_stderr("Error: connection closed\n").await;
|
|
|
|
assert!(!seen.contains(&Event::Cleared), "{seen:?}");
|
|
assert_eq!(
|
|
read_resume_token(dir.path()).as_deref(),
|
|
Some("a1c6c855-86cd-4f47-8b50-eaea85be3579")
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_refused_token_is_forgotten_so_the_next_start_makes_a_session() {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
write_resume_token(dir.path(), "a1c6c855-86cd-4f47-8b50-eaea85be3579");
|
|
assert!(read_resume_token(dir.path()).is_some());
|
|
|
|
assert!(forget_resume_token(dir.path()));
|
|
|
|
// Gone, so `launch` pushes `--name` instead of `--resume` and the CLI
|
|
// is asked for a session it can actually make.
|
|
assert_eq!(read_resume_token(dir.path()), None);
|
|
// Removing an absent one succeeds too -- the check and the remove are
|
|
// separate, so the file can go between them.
|
|
assert!(forget_resume_token(dir.path()));
|
|
assert_eq!(read_resume_token(dir.path()), None);
|
|
}
|
|
}
|