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

Merge branch 'main' of git.arirex.me:iris/ai-app

Browse Source
This commit is contained in:
iris committed 2026-09-01 00:51:21 -04:00
2 parents 28902ac834 7525fc925a
commit 4add4fd176
4 files changed
+230 -79
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/routes.rs
+64 -18
View File
@@ -549,28 +549,68 @@ struct ImportableRow {
/// id is registered as in flight before the 202 goes back, so the answer to
/// "did my batch start" is one answer for the batch.
///
/// Registering is what has to be atomic; the work itself does not. Each id
/// runs on its own task and settles on its own event, because six deletes
/// that must all succeed or all roll back is not something a filesystem
/// offers, and pretending otherwise would mean holding five sessions
/// hostage to the one that failed.
/// Registering is what has to be atomic; the work itself does not. The
/// batch runs as one command on the machine -- see
/// [`crate::session::import::delete`] for why it is not one per id -- but
/// each row still settles on its own event from its own outcome, because
/// six deletes that must all succeed or all roll back is not something a
/// filesystem offers, and pretending otherwise would mean holding five
/// sessions hostage to the one that failed.
async fn delete_importable(
State(manager): State<Arc<SessionManager>>,
UrlPath(id): UrlPath<String>,
axum::Json(body): axum::Json<DeleteBatch>,
) -> Result<StatusCode, ApiError> {
let setup = setup_by_id(&manager, &id)?;
for session in body.sessions {
let transport = crate::session::transport::Transport::for_setup(&setup);
let target = session.clone();
in_background(
&manager,
id.clone(),
session,
Operation::Deleting,
async move { crate::session::import::delete(&transport, &target).await },
);
}
let transport = crate::session::transport::Transport::for_setup(&setup);
// Registered before anything is spawned, so the 202 is only sent once
// every row is already showing "deleting" -- a phone that refetches the
// instant it gets the reply cannot catch a row that has not started.
let running: Vec<(String, crate::session::pending::InFlight)> = body
.sessions
.iter()
.map(|session| {
(
session.clone(),
manager.pending().begin(&id, session, Operation::Deleting),
)
})
.collect();
let sessions = body.sessions;
tokio::spawn(async move {
// One failure here is the machine being unreachable, which is true
// of every row rather than of any one of them, so they all say so.
let outcomes = match crate::session::import::delete(&transport, &sessions).await {
Ok(outcomes) => outcomes,
Err(err) => {
let message = format!("{err:#}");
tracing::warn!(
"deleting {} sessions on {id} failed: {message}",
running.len()
);
for (_, flight) in running {
flight.failed(message.clone());
}
return;
}
};
for (session, flight) in running {
match outcomes.get(&session) {
Some(Ok(())) => {
tracing::info!("deleting {session} on {id}: done");
flight.succeeded();
}
Some(Err(message)) => {
tracing::warn!("deleting {session} on {id} failed: {message}");
flight.failed(message.clone());
}
// `delete` promises an entry per id, so this is a bug
// rather than a state -- but a row stuck on "deleting"
// for ever is a worse answer than one that says so.
None => flight.failed(format!("nothing was reported about {session}")),
}
}
});
Ok(StatusCode::ACCEPTED)
}
@@ -872,9 +912,15 @@ async fn delete_session(
if let Some((setup, session)) = &foreign {
let setup = setup_by_id(&manager, setup)?;
let transport = crate::session::transport::Transport::for_setup(&setup);
crate::session::import::delete(&transport, session)
// A batch of one: the same call, so there is one description of
// what deleting a foreign transcript means. Its outcome is this
// request's outcome, since there is only the one row.
crate::session::import::delete(&transport, std::slice::from_ref(session))
.await
.map_err(bad_request)?;
.map_err(bad_request)?
.remove(session)
.unwrap_or_else(|| Err(format!("nothing was reported about {session}")))
.map_err(|message| bad_request(anyhow::anyhow!("{message}")))?;
tracing::info!("deleted Claude Code session {session} with ai-app session {id}");
}
manager.delete_session(&id).map_err(bad_request)?;
server/src/session/import.rs
+166 -35
View File
@@ -17,7 +17,9 @@
//! reason: an enrolled token must not be able to turn into "read me this
//! arbitrary path".
use anyhow::{Context, Result, bail, ensure};
use std::collections::HashMap;
use anyhow::{Context, Result};
use serde::Serialize;
use serde_json::Value;
@@ -684,9 +686,42 @@ fn push_assistant(events: &mut Vec<Event>, content: &Value) {
}
}
/// Deletes one of the sessions [`list`] reported.
/// Removes each id it is given and prints one `<id>\t<state>` line per id.
///
/// By id, resolved here against what the machine actually has, so the
/// The three states are every way removing one id can end: `deleted` if at
/// least one file went, `missing` if the glob matched nothing, `failed` if
/// an `rm` refused. "Not there" is deliberately kept apart from "it broke"
/// rather than folded together by the shell -- only one of them is worth
/// retrying, and the caller is what knows how to word either.
///
/// Every copy of each id, not the first. The same id can name a file under
/// two project directories -- see the de-duplication in `parse_listing` --
/// and stopping at the first left the other behind, so the row came back on
/// the next listing after a delete that had reported success. `failed`
/// therefore sticks once set: one copy removed and another refused is not a
/// success.
///
/// Ids arrive as arguments rather than in the script text, so nothing here
/// is shell syntax; `is_session_id` is what keeps one from globbing its way
/// out of the projects directory.
const DELETE_SCRIPT: &str = r#"
for id do
state=missing
for f in "$HOME"/.claude/projects/*/"$id".jsonl; do
[ -f "$f" ] || continue
if rm -f "$f"; then
[ "$state" = failed ] || state=deleted
else
state=failed
fi
done
printf '%s\t%s\n' "$id" "$state"
done
"#;
/// Deletes sessions [`list`] reported, and says what happened to each.
///
/// By id, resolved on the machine against what it actually has, so the
/// caller never names a file -- the same rule importing follows, and it
/// matters more here: this one removes something.
///
@@ -694,7 +729,48 @@ fn push_assistant(events: &mut Vec<Event>, content: &Value) {
/// keeps no copy: the JSONL *is* the session, so deleting it ends any
/// chance of resuming that conversation, including from an ai-app session
/// that was already importing it.
pub async fn delete(transport: &Transport, id: &str) -> Result<()> {
///
/// The whole batch in one invocation, which over ssh is the difference
/// between one connection and one per session. Six deletes started in the
/// same tick were six `ssh` processes racing to authenticate, and a batch
/// big enough to pass the remote sshd's `MaxStartups` (10 unauthenticated
/// connections, by default, before it begins refusing) had rows come back
/// as `Connection closed by … port 2222` -- a row reporting a delete that
/// never ran, for a reason that has nothing to do with the session. One
/// connection cannot exceed that however many ids are selected.
///
/// Still one outcome per id, because a batch is not a transaction: six
/// removals that must all succeed or all roll back is not something a
/// filesystem offers, and the caller settles each row from its own line.
/// Every requested id gets an entry, so an id the machine said nothing
/// about is reported as such rather than defaulting to either answer.
pub async fn delete(
transport: &Transport,
ids: &[String],
) -> Result<HashMap<String, Result<(), String>>> {
// Refused here rather than on the machine: `is_session_id` is what
// keeps an id from walking out of the projects directory, and a bad
// one must never reach the glob. It fails only itself -- one malformed
// id is not a reason to leave the other five in place.
let (safe, mut outcomes): (Vec<&String>, HashMap<String, Result<(), String>>) =
ids.iter().fold(
(Vec::new(), HashMap::new()),
|(mut safe, mut outcomes), id| {
if is_session_id(id) {
safe.push(id);
} else {
outcomes.insert(
id.clone(),
Err(format!("not a Claude Code session id: {id}")),
);
}
(safe, outcomes)
},
);
if safe.is_empty() {
return Ok(outcomes);
}
// The file name *is* the id, so the machine can find it by name. This
// used to call `list` and search its output, which is correct and costs
// a full read of every transcript on the machine -- around four seconds
@@ -702,40 +778,58 @@ pub async fn delete(transport: &Transport, id: &str) -> Result<()> {
// best part of a minute doing nothing but re-reading the same files.
// `context_of` below already resolved an id the cheap way; this is the
// same lookup, and the two now agree.
ensure!(is_session_id(id), "not a Claude Code session id: {id}");
// Every copy, not the first. The same id can name a file under two
// project directories -- see the de-duplication in `parse_listing` --
// and stopping at the first left the other behind, so the row came back
// on the next listing after a delete that had reported success.
let script = r#"
for f in "$HOME"/.claude/projects/*/"$1".jsonl; do
[ -f "$f" ] || continue
rm -f "$f" || exit 1
printf '%s\n' "$f"
done
"#;
let launch = Launch::new(
"sh",
vec![
"-c".to_string(),
script.to_string(),
"sh".to_string(),
id.to_string(),
],
None,
);
// Nothing on stdout means the loop found no such file. Said here rather
// than by exiting non-zero, because a non-zero exit is reported as the
// machine being unreachable -- which is a different thing from the
// session not being there, and only one of them is worth retrying.
let removed = transport
//
// Every copy of each id, not the first. The same id can name a file
// under two project directories -- see the de-duplication in
// `parse_listing` -- and stopping at the first left the other behind,
// so the row came back on the next listing after a delete that had
// reported success.
//
// Each id prints its own verdict rather than the loop exiting on the
// first failure: with a batch, exiting would leave every id after it
// unexplained. See [`DELETE_SCRIPT`] for what the words mean.
let mut args = vec![
"-c".to_string(),
DELETE_SCRIPT.to_string(),
"sh".to_string(),
];
args.extend(safe.iter().map(|id| (*id).clone()));
let launch = Launch::new("sh", args, None);
// A failure to run the script at all is the machine being unreachable,
// which is true of every id in the batch rather than of any one of
// them -- so it is returned as the error, not written into each row.
let reported = transport
.capture(&launch)
.await
.with_context(|| format!("deleting Claude Code session {id}"))?;
if removed.trim().is_empty() {
bail!("no Claude Code session {id} on that machine");
.with_context(|| format!("deleting {} Claude Code sessions", safe.len()))?;
for line in reported.lines() {
let Some((id, state)) = line.trim().split_once('\t') else {
continue;
};
outcomes.insert(
id.to_string(),
match state {
"deleted" => Ok(()),
"missing" => Err(format!("no Claude Code session {id} on that machine")),
_ => Err(format!("couldn't remove Claude Code session {id}")),
},
);
}
Ok(())
// Anything the machine did not mention. The connection can drop
// part-way through the loop, and an id whose line never arrived is one
// nobody knows the fate of -- which is its own answer, and must not be
// read as either a success or a clean "not there".
for id in safe {
outcomes.entry(id.clone()).or_insert_with(|| {
Err(format!(
"couldn't tell whether Claude Code session {id} was deleted -- the machine \
stopped answering part-way through the batch"
))
});
}
Ok(outcomes)
}
/// Whether an id is one of ours to put in a shell glob.
@@ -940,6 +1034,43 @@ mod tests {
assert!(!is_session_id(&"a".repeat(65)));
}
/// One batch, one invocation, one verdict per id -- including for the
/// two cases a single-id delete never had to keep apart from the rest:
/// an id recorded under two project directories (both copies must go,
/// and it still reports once) and an id that is not there at all.
///
/// Runs the real script against a temporary `$HOME`, because what is
/// being checked is the shell, not the Rust around it.
#[test]
fn a_batch_deletes_every_copy_and_reports_each_id_once() {
let home = tempfile::tempdir().expect("tempdir");
let projects = home.path().join(".claude/projects");
let one = "5ecf21da-d53f-4a11-9c0d-000000000100";
let twice = "5ecf21da-d53f-4a11-9c0d-000000000200";
let absent = "5ecf21da-d53f-4a11-9c0d-000000000300";
for (project, id) in [("a", one), ("a", twice), ("b", twice)] {
let dir = projects.join(project);
std::fs::create_dir_all(&dir).expect("project dir");
std::fs::write(dir.join(format!("{id}.jsonl")), "{}\n").expect("transcript");
}
let output = std::process::Command::new("sh")
.args(["-c", DELETE_SCRIPT, "sh", one, twice, absent])
.env("HOME", home.path())
.output()
.expect("run the delete script");
assert!(output.status.success());
assert_eq!(
String::from_utf8_lossy(&output.stdout),
format!("{one}\tdeleted\n{twice}\tdeleted\n{absent}\tmissing\n"),
);
// The second copy is the one a per-id delete used to leave behind.
assert!(!projects.join("b").join(format!("{twice}.jsonl")).exists());
assert!(!projects.join("a").join(format!("{twice}.jsonl")).exists());
assert!(!projects.join("a").join(format!("{one}.jsonl")).exists());
}
/// A 1x1 PNG, base64 -- the smallest thing with a real header.
const PNG: &str = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk\
YPhfDwAChwGA60e6kgAAAABJRU5ErkJggg==";
server/src/session/transport.rs
-19
View File
@@ -21,31 +21,12 @@
use std::path::{Path, PathBuf};
use std::process::Stdio;
use std::sync::LazyLock;
use anyhow::{Context, Result};
use tokio::process::Child;
use tokio::sync::Semaphore;
use crate::config::SshConfig;
/// How many [`Transport::capture`] calls may have an ssh connection open at
/// once, across every setup.
///
/// `capture` is what a batch (deleting several imports, listing several
/// machines) fans out over -- one child `ssh` process per call, all started
/// within the same tick. Nothing here throttled that, so a batch large
/// enough to open more connections than the remote sshd's default
/// `MaxStartups` (10, before it starts randomly refusing) has some of
/// them come back as "Connection closed" -- not a real failure of the
/// operation, just too many handshakes landing on the listener at once. Four
/// keeps a batch comfortably under that ceiling while still overlapping the
/// network round trips. Long-lived processes (`Transport::spawn`, a
/// session's own child) don't take a permit: they hold it for the session's
/// lifetime rather than for one round trip, which would starve every other
/// probe behind it.
pub(crate) static CAPTURE_PERMITS: LazyLock<Semaphore> = LazyLock::new(|| Semaphore::new(4));
/// What a driver needs run in order to exist as a process.
///
/// Deliberately just the three things every transport can carry. Anything
server/src/setups.rs
-7
View File
@@ -163,13 +163,6 @@ pub fn tidy(value: &str) -> Option<String> {
/// Runs a launch to completion and returns its stdout.
impl Transport {
pub async fn capture(&self, launch: &Launch) -> Result<String> {
// See `CAPTURE_PERMITS`: caps how many of these run their ssh
// connection at once, so a batch doesn't open more than the remote
// sshd tolerates before it starts dropping them.
let _permit = super::session::transport::CAPTURE_PERMITS
.acquire()
.await
.expect("capture semaphore is never closed");
let child = self.spawn(launch, super::session::transport::Streams::Piped)?;
let output = child
.wait_with_output()