iris/ai-app
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Let sessions outlive the backend, and never resume one twice

Browse Source 
Three `claude` processes ended up running against this checkout on
2026-08-29, and the account hit its session limit. One cause, several
ways in.

An agent imported the Claude Code session it was *itself* running in.
That is an ordinary import, and importing runs `--resume` -- so a second
CLI attached to a file the first was still writing. The whole 65 MB
conversation, 154 embedded screenshots included, was re-appended to the
transcript under a new prompt id; both copies then read each other's
writes as work done elsewhere, and the adopted one was billed for
re-reading all of it. Meanwhile `shutdown_all` asked each session to stop
and the process exited immediately, so the SIGKILL timer died with the
runtime, the stop was unreliable, and whatever survived was orphaned with
nothing written down to find it by.

The processes leaked either way. So leak them on purpose, and be able to
pick them back up.

A session's process now outlives the backend and is adopted again on the
way up, which is worth having for its own sake: restarting the server no
longer ends a turn somebody is waiting on. Its stdio lives in the session
directory -- a fifo opened read-write so the process is its own last
writer and never reads EOF, plus stdout/stderr logs read from a byte
offset. `session::process` records the pid *and* the kernel's start time
for it, because a pid alone is reused and adopting a stranger's would mean
never resuming the real conversation.

That makes the fix structural rather than a check: everything goes through
`ClaudeDriver::launch`, which adopts if it can and starts if it cannot,
and `--resume` is reachable only on the second path. `Driver` gains two
ways out where it had one -- `detach` (coming back) and `stop` (the
session is being deleted, so the process must not survive).

Importing a session that is open is now refused outright. Claude Code
keeps `~/.claude/sessions/<pid>.json` for every live session, so this is a
measurement rather than a guess; it reports no/yes/unknown, because a
machine that keeps no such record cannot answer and "could not check" is
not "nobody is using it". `SessionStatus` gains `Unknown` for the same
reason.

Also here, found on the way:

- A reconnecting phone was sent the entire backlog. Opening a session was
  bounded to a page but reconnecting was not, so a long disconnect
  delivered thousands of events one frame at a time. Past `CATCH_UP_LIMIT`
  the stream sends a `reset` frame and the newest window, and the client
  rebuilds from it as it does on open -- without the reset the window is
  spliced onto rows no longer adjacent to it.
- A session's status was assumed idle at launch. Read from the transcript
  instead, so a restart stops claiming an exited session is waiting for
  you.
- `llama-server`'s stdout was piped and never drained, so a chatty one
  blocked on a full pipe buffer mid-load. It goes to a log now.
- A turn that exited or errored never emitted `Idle`, so the queue stayed
  "running" for good: every later message was held forever and, since a
  message is only recorded when taken, vanished with nothing on screen.
- Two doc comments had drifted onto the wrong functions.

Verified by killing the server mid-turn: the process survived, finished
its turn unattended (12.8 KB of output nothing was reading), and the
restarted server adopted it -- one process, all 700 lines in the
transcript, no hole, and it still took a new message afterwards. Deleting
a session stops its process; a 266-event backlog resets while a 16-event
one streams. 46 tests, clippy and rustfmt clean, app compiles and lints.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VETa8afmpWaYezLCqJhDB8
This commit is contained in:
irisandClaude Opus 5 committed 2026-08-29 04:47:43 -04:00
1 parent 9791afcfd6
commit 362d436d4f
23 files changed
+1934 -345
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/session/claude.rs
+629 -153
View File
@@ -1,11 +1,19 @@
//! The Claude Code driver: `claude -p` speaking stream-json on stdio,
//! translated into the common event model.
//!
//! This half owns the process -- asking a transport to start it, resuming
//! it after a crash, writing lines to it, and shutting it down. 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.
//! 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.
//!
@@ -39,16 +47,19 @@
//! 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::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::sync::{mpsc, oneshot};
use tokio::io::AsyncWriteExt;
use tokio::sync::mpsc;
use super::driver::{Driver, Event, EventSink, ImageRef, SessionStatus};
use super::transport::{Launch, Transport};
use super::process;
use super::transport::{Launch, Streams, Transport};
use crate::config::{ProviderConfig, SessionConfig};
use translate::{AnswerOutcome, Translator};
@@ -89,28 +100,219 @@ mod translate;
const RESUME_FILE: &str = "claude-session.json";
/// Grace period between closing stdin (the polite exit) and SIGKILL.
const SHUTDOWN_GRACE: std::time::Duration = std::time::Duration::from_secs(5);
/// 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 sent while a turn was already running, and whether one is.
///
/// The CLI does not inject a message into a turn in flight: a line written
/// to its stdin mid-turn simply becomes the next turn, and it says nothing
/// on stdout about having read it. So the wait is held here instead, where
/// the moment it ends is a line *this* driver writes -- which is what lets
/// the phone be told, and what puts the message in the transcript where it
/// was read rather than where it was typed.
#[derive(Default)]
struct Queue {
/// A turn is in flight, so anything sent now waits for it.
running: bool,
/// Each held message as the text to report and the line to write.
waiting: VecDeque<(String, 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: the queue
/// then held every later message forever, and since a message is only
/// recorded when it is *taken*, each 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) {
self.closed = true;
self.running = false;
let lost: Vec<String> = self.waiting.drain(..).map(|(text, _)| text).collect();
if lost.is_empty() {
return;
}
let _ = sink.send(Event::Error {
message: format!(
"the session ended before it read {}: {}",
if lost.len() == 1 {
"this message".to_string()
} else {
format!("{} queued messages", lost.len())
},
lost.join(" / ")
),
});
}
}
/// The session directory's copies of the process's standard streams.
///
/// Named once rather than built at each use, because the spawn path and
/// the attach path must agree about which file is which; if they drift,
/// a reattached session reads a file nothing is writing and simply looks
/// idle forever.
const STDIN_FIFO: &str = "stdin.fifo";
const STDOUT_LOG: &str = "stdout.log";
const STDERR_LOG: &str = "stderr.log";
/// How often a reader with nothing to read looks again.
///
/// A poll rather than a watch: the alternative is an inotify dependency
/// for one file per session, and at this interval the streaming text is
/// already arriving faster than a phone renders it.
const POLL: std::time::Duration = std::time::Duration::from_millis(50);
pub struct ClaudeDriver {
sink: EventSink,
/// Lines for the child's stdin; `None` after shutdown started (taking
/// it closes stdin, which is the CLI's graceful exit signal).
to_child: Mutex<Option<mpsc::UnboundedSender<String>>>,
/// Fires SIGKILL if the process outlives the shutdown grace period.
kill: Mutex<Option<oneshot::Sender<()>>>,
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 {
pub fn spawn(
/// 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),
to_child.clone(),
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());
@@ -132,119 +334,50 @@ impl ClaudeDriver {
}
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 mut child = transport.spawn(&launch)?;
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} {}",
"session {} running {program} {} as pid {pid}",
meta.id,
transport.describe()
);
let stdin = child.stdin.take().expect("piped stdin");
let stdout = child.stdout.take().expect("piped stdout");
let stderr = child.stderr.take().expect("piped stderr");
let state = Arc::new(Mutex::new(Translator::new(session_dir.to_path_buf())));
// Writer: everything for the child funnels through one channel so
// driver methods stay sync and writes can't interleave.
let (to_child, mut from_driver) = mpsc::unbounded_channel::<String>();
// 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 stdin = 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;
}
}
// Sender dropped/taken: stdin drops here, closing it -- the
// CLI's signal to finish up and exit.
let mut child = child;
let _ = child.wait().await;
});
tokio::spawn(read_stdout(
stdout,
Arc::clone(&state),
sink.clone(),
session_dir.to_path_buf(),
));
// stderr is diagnostics only; surface it in the log, and keep the
// tail of it for the exit report below. For a remote provider this
// is also where ssh's own failures arrive ("Permission denied",
// "Could not resolve hostname"), which are the ones a person
// actually needs to see.
//
// A ring of the last lines rather than the last line alone. Keeping
// one line meant keeping whatever happened to come last, and what
// comes last is very often blank -- a shell's error message ends
// with one -- so the report was a bare exit status and the actual
// complaint existed only in the server's log, which is not where
// the person holding the phone is looking. A failing `cd` cost an
// evening to exactly that.
let recent_stderr = Arc::new(Mutex::new(VecDeque::<String>::new()));
{
let recent_stderr = Arc::clone(&recent_stderr);
let label = provider.name.clone();
tokio::spawn(async move {
let mut lines = BufReader::new(stderr).lines();
while let Ok(Some(line)) = lines.next_line().await {
tracing::warn!("{label} stderr: {line}");
let mut kept = recent_stderr.lock().unwrap();
kept.push_back(line);
while kept.len() > STDERR_LINES_KEPT {
kept.pop_front();
}
}
});
}
// Monitor: reports process death as an event (with stderr context
// when it died complaining), and carries the SIGKILL escape hatch.
let (kill_tx, kill_rx) = oneshot::channel::<()>();
{
let sink = sink.clone();
let label = format!("{} {}", provider.name, transport.describe());
tokio::spawn(async move {
let status = tokio::select! {
status = child.wait() => status.ok(),
_ = kill_rx => {
let _ = child.kill().await;
None
}
};
if let Some(status) = status
&& !status.success()
{
let detail = tail_of(&recent_stderr.lock().unwrap());
let _ = sink.send(Event::Error {
message: if detail.is_empty() {
format!("{label} exited with {status}")
} else {
format!("{label} exited with {status}:\n{detail}")
},
});
}
let _ = sink.send(Event::Status {
state: SessionStatus::Exited,
});
});
}
Ok(Self {
sink,
to_child: Mutex::new(Some(to_child)),
kill: Mutex::new(Some(kill_tx)),
state,
session_dir: session_dir.to_path_buf(),
})
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) {
if let Some(sender) = self.to_child.lock().unwrap().as_ref() {
let _ = sender.send(line);
}
let _ = self.to_child.send(line);
}
fn send_control(&self, request: Value) {
@@ -271,14 +404,31 @@ impl Driver for ClaudeDriver {
if !text.is_empty() {
content.push(json!({"type": "text", "text": text}));
}
// Sent mid-turn this queues for injection at the next tool
// boundary; sent while idle it starts a turn.
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 {
queue.waiting.push_back((text, line));
return;
}
queue.running = true;
drop(queue);
let _ = self.sink.send(Event::MessageTaken { text });
let _ = self.sink.send(Event::Status {
state: SessionStatus::Running,
});
self.send_line(
json!({"type": "user", "message": {"role": "user", "content": content}}).to_string(),
);
self.send_line(line);
}
fn answer_question(&self, id: &str, answer: &str) {
@@ -303,6 +453,9 @@ impl Driver for ClaudeDriver {
}
}
/// 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"}));
}
@@ -325,52 +478,292 @@ impl Driver for ClaudeDriver {
);
}
fn shutdown(&self) {
// Closing stdin is the polite exit; the kill timer is the escape
// hatch for a CLI that doesn't oblige.
self.to_child.lock().unwrap().take();
if let Some(kill) = self.kill.lock().unwrap().take() {
tokio::spawn(async move {
tokio::time::sleep(SHUTDOWN_GRACE).await;
let _ = kill.send(());
});
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);
}
}
async fn read_stdout(
stdout: tokio::process::ChildStdout,
/// 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,
session_dir: PathBuf,
queue: Arc<Mutex<Queue>>,
to_child: mpsc::UnboundedSender<String>,
reading: Arc<AtomicBool>,
label: String,
) {
let mut lines = BufReader::new(stdout).lines();
while let Ok(Some(line)) = lines.next_line().await {
let Ok(message) = serde_json::from_str::<Value>(&line) else {
tracing::warn!(
"unparseable claude output line: {}",
&line[..line.len().min(200)]
);
continue;
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 = match process::read_from(&stderr_path, u64::MAX) {
Ok((_, at)) => at,
Err(_) => 0,
};
let mut said_unknown = false;
while reading.load(Ordering::SeqCst) {
let (bytes, _) = match process::read_from(&stdout_path, offset) {
Ok(found) => found,
Err(err) => {
tracing::error!("couldn't read {}: {err:#}", stdout_path.display());
return;
}
};
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 {
if sink.send(event).is_err() {
// 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, &to_child) {
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);
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.
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>>,
to_child: &mpsc::UnboundedSender<String>,
) -> bool {
let Ok(message) = serde_json::from_str::<Value>(line) else {
tracing::warn!(
"unparseable claude output line: {}",
&line[..line.len().min(200)]
);
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 {
// A turn ending is when a held message is taken up, and the
// session is then not idle at all -- it is about to start the
// turn that message asked for. Reporting the idle would show a
// phone a finished session for as long as it took the next
// turn to produce anything, with the message it is holding
// still drawn as waiting.
if matches!(
event,
Event::Status {
state: SessionStatus::Idle
}
) {
let next = {
let mut queue = queue.lock().unwrap();
let next = queue.waiting.pop_front();
queue.running = next.is_some();
next
};
if let Some((text, line)) = next {
if sink.send(Event::MessageTaken { text }).is_err() || to_child.send(line).is_err()
{
return false;
}
continue;
}
}
if sink.send(event).is_err() {
return false; // session torn down
}
}
true
}
/// The end of the stderr log, for an exit report a person reads.
///
/// Bounded because this is held in a message; trimmed of blank lines at
/// both ends because a shell's error ends with one, so anything reporting
/// "the last line" reports nothing at all. A failing `cd` cost an evening
/// to exactly that.
fn stderr_tail(path: &Path) -> String {
let Ok(text) = std::fs::read_to_string(path) else {
return String::new();
};
let kept: VecDeque<String> = text
.lines()
.rev()
.take(STDERR_LINES_KEPT)
.map(str::to_string)
.collect::<Vec<_>>()
.into_iter()
.rev()
.collect();
tail_of(&kept)
}
/// Creates the stdin fifo if it is not already there, and opens it
/// read-write for the process to inherit.
///
/// Read-write is the whole trick, and it is not an accident of
/// convenience: a fifo opened read-only delivers EOF as soon as the last
/// writer closes, so the process would exit the moment this server did --
/// which is exactly what leaving it running has to prevent. Holding it
/// open for writing as well means the process is its own last writer and
/// never sees the end of its input.
fn make_fifo(path: &Path) -> Result<std::fs::File> {
if !path.exists() {
let c_path = std::ffi::CString::new(path.as_os_str().as_encoded_bytes())
.with_context(|| format!("{} is not a usable path", path.display()))?;
// SAFETY: a nul-terminated path this call only reads, and a mode
// with no bits the kernel can object to. Owner-only, like
// everything else in a session directory: this carries what the
// person typed.
let made = unsafe { libc::mkfifo(c_path.as_ptr(), 0o600) };
if made != 0 {
return Err(std::io::Error::last_os_error())
.with_context(|| format!("creating the fifo {}", path.display()));
}
}
std::fs::OpenOptions::new()
.read(true)
.write(true)
.open(path)
.with_context(|| format!("opening the fifo {}", path.display()))
}
/// A fresh, empty, owner-only log for one of the process's output streams.
fn create_log(path: &Path) -> Result<std::fs::File> {
std::fs::OpenOptions::new()
.create(true)
.write(true)
.truncate(true)
.mode(0o600)
.open(path)
.with_context(|| format!("creating {}", path.display()))
}
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)
@@ -454,4 +847,87 @@ mod tests {
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.waiting.push_back(("first".into(), "{}".into()));
queue.waiting.push_back(("second".into(), "{}".into()));
queue.close(&sink);
// 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.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 closing_an_empty_queue_says_nothing() {
let (sink, mut received) = mpsc::unbounded_channel();
let mut queue = Queue::default();
queue.close(&sink);
// 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);
}
}