irisandClaude Opus 5
797513bb86
100k is the cheapest window on tokens and the wrong one to sit in front of. Measured on the session this was written against: context returns to 70-85k within ten calls of a compaction, so a 100k window compacts about every thirteen calls, and a compaction takes roughly two minutes (durationMs 104,346 to 147,671 across the six recorded). A 130-call request would have spent some twenty minutes compacting -- optimising the number that was asked about while making the thing somebody actually waits for on a phone considerably worse. 200k keeps most of the saving against the 1M ceiling and halves the stalls. The comment now also says what the window does not do, because measuring this turned up the opposite of what the byte counts suggested. Images are 93% of the bytes that tool calls put into that transcript but only 8% of the context growth -- the adb wrapper's downscaling holds a screenshot to a median of 476 tokens, while text-only calls add a median of 740 and a mean of 1,139. So the file is large because of screenshots and the context is large because of ordinary tool output, and only the second one is what this constant governs. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VETa8afmpWaYezLCqJhDB8
1230 lines
52 KiB
Rust
1230 lines
52 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};
|
|
use super::process;
|
|
use super::transport::{Launch, Streams, Transport};
|
|
use crate::config::{ProviderConfig, SessionConfig};
|
|
use translate::{AnswerOutcome, Setting, Translator};
|
|
|
|
/// 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";
|
|
|
|
/// Grace period between asking a process to stop and killing it.
|
|
///
|
|
/// Only [`Driver::stop`] uses it -- a session being deleted. Detaching
|
|
/// does not stop anything, so it has no grace period and needs none.
|
|
const STOP_GRACE: std::time::Duration = std::time::Duration::from_secs(5);
|
|
|
|
/// 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:
|
|
/// the next assistant text or tool call is proof another model call has
|
|
/// happened, and the message was in it. 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.
|
|
#[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.
|
|
awaiting: VecDeque<String>,
|
|
/// 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(..).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";
|
|
|
|
/// The context size the CLI compacts at, rather than letting it choose.
|
|
///
|
|
/// Left to `auto` a phone session drifts: these run for hours, nobody
|
|
/// closes them, and the transcript carries screenshots. Measured on one
|
|
/// here, the window the CLI chose was 1M and it compacted at the ceiling
|
|
/// -- its only automatic compaction fired at 1,000,184 tokens, and the
|
|
/// context peaked at 999,668. Every API call re-reads the whole context,
|
|
/// so near that ceiling a single tool call bills about 100k tokens
|
|
/// before it does anything.
|
|
///
|
|
/// What kept that session usable was the person in it running `/compact`
|
|
/// by hand, four times. That is the state this constant is really for:
|
|
/// even between those manual compactions, at the merely-large contexts
|
|
/// they left behind, one ordinary request ("can you make it so you can
|
|
/// rename a session?") cost 4.2 million tokens across its 130 calls.
|
|
///
|
|
/// 200k rather than the 100k floor, because the cheapest window on tokens
|
|
/// is not the best one to sit in front of. Measured on the same session:
|
|
/// context comes back to 70-85k within ten calls of a compaction, so a
|
|
/// 100k window compacts about every thirteen calls, and a compaction
|
|
/// takes roughly two minutes (`durationMs` 104,346 to 147,671 across the
|
|
/// six recorded). A 130-call request would spend some twenty minutes
|
|
/// compacting. 200k keeps most of the saving against the 1M ceiling and
|
|
/// halves the stalls.
|
|
///
|
|
/// Worth knowing before tuning this: the window decides how often the
|
|
/// context is thrown away, not how fast it fills. What fills it is ~7k
|
|
/// per call of tool output, and no value here touches that.
|
|
const AUTOCOMPACT_WINDOW: &str = "200k";
|
|
|
|
/// 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.
|
|
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
|
|
);
|
|
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
|
|
);
|
|
record
|
|
}
|
|
Some((_, process::Liveness::Dead)) | None => {
|
|
Self::start(meta, provider, transport, session_dir)?
|
|
}
|
|
};
|
|
// 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");
|
|
push("--autocompact", AUTOCOMPACT_WINDOW);
|
|
// 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());
|
|
|
|
// 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) {
|
|
self.queue.lock().unwrap().running = true;
|
|
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`.
|
|
queue.awaiting.push_back(text);
|
|
drop(queue);
|
|
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.
|
|
let _ = self.sink.send(Event::MessageTaken { text });
|
|
let _ = self.sink.send(Event::Status {
|
|
state: SessionStatus::Running,
|
|
});
|
|
self.send_line(line);
|
|
}
|
|
|
|
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) {
|
|
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 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, 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 (events, new_session_id) = {
|
|
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 })
|
|
};
|
|
if let Some(session_id) = new_session_id {
|
|
write_resume_token(session_dir, &session_id);
|
|
}
|
|
for event in events {
|
|
// Anything the CLI says after a steer was written is proof it has
|
|
// been round the model again, and the steer went with it -- so
|
|
// that is the moment it is announced, and the moment a phone can
|
|
// stop drawing it as still waiting. The announcement goes out
|
|
// *before* the event that proves it, so the message is above the
|
|
// output it produced rather than below it.
|
|
//
|
|
// A turn ending counts too, and is the case that must not be
|
|
// missed: a message written after the last model call of a turn
|
|
// has no later output to prove anything, and without this it would
|
|
// never be announced at all.
|
|
if announces_a_steer(&event) {
|
|
let taken: Vec<String> = {
|
|
let mut queue = queue.lock().unwrap();
|
|
queue.awaiting.drain(..).collect()
|
|
};
|
|
for text in taken {
|
|
if sink.send(Event::MessageTaken { text }).is_err() {
|
|
return false;
|
|
}
|
|
}
|
|
}
|
|
// 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
|
|
}
|
|
) {
|
|
queue.lock().unwrap().running = false;
|
|
}
|
|
if sink.send(event).is_err() {
|
|
return false; // session torn down
|
|
}
|
|
}
|
|
true
|
|
}
|
|
|
|
/// Whether this event could only have come from a turn in flight.
|
|
///
|
|
/// The turn this side starts is announced where it is started, and that
|
|
/// covers the common case and nothing else. Everything below happens
|
|
/// without a phone asking for it: a compaction the CLI decided on by
|
|
/// itself, a session adopted while it was already mid-turn, a message
|
|
/// that reached the conversation by some route other than this server --
|
|
/// another agent writing to it, or somebody at the terminal. In all of
|
|
/// them the CLI is plainly working and the only thing that would ever
|
|
/// have said so is a `Running` nobody sent, so the session sits there
|
|
/// reading as idle until the turn ends.
|
|
///
|
|
/// So the driver says it from what it observes rather than from what it
|
|
/// was asked to do, and this is the same set as [`announces_a_steer`]
|
|
/// with the ends swapped: that one takes the `Idle` that closes a turn
|
|
/// and this one takes the states that open one. `Idle` is the pair to
|
|
/// this -- it is where `running` goes back to false, a few lines above
|
|
/// where it is set here.
|
|
fn proves_a_turn(event: &Event) -> bool {
|
|
matches!(
|
|
event,
|
|
Event::AssistantText { .. }
|
|
| Event::ToolStart { .. }
|
|
| Event::ToolUpdate { .. }
|
|
| Event::ToolEnd { .. }
|
|
| Event::Question { .. }
|
|
| Event::Compacted { .. }
|
|
| Event::Status {
|
|
state: SessionStatus::Compacting | SessionStatus::AwaitingInput
|
|
}
|
|
)
|
|
}
|
|
|
|
/// Whether this event proves the CLI has consumed anything written to it
|
|
/// since the last one did.
|
|
///
|
|
/// Assistant output and a tool call both mean another model call happened;
|
|
/// an idle means the turn is over and nothing further is coming. Status
|
|
/// changes that are not idle prove nothing -- a turn can go `running`
|
|
/// without having read a line written a moment ago.
|
|
fn announces_a_steer(event: &Event) -> bool {
|
|
matches!(
|
|
event,
|
|
Event::AssistantText { .. }
|
|
| Event::ToolStart { .. }
|
|
| Event::Status {
|
|
state: SessionStatus::Idle
|
|
}
|
|
)
|
|
}
|
|
|
|
/// 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()))
|
|
}
|
|
|
|
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::*;
|
|
|
|
/// 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("first".into());
|
|
queue.awaiting.push_back("second".into());
|
|
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);
|
|
}
|
|
}
|