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

Show a session's subagents as subcards, each with a read-only transcript

Browse Source 
A subagent is a second transcript owned by a session, in the same event
model, with no process and no controls. The claude translator routes lines
carrying parent_tool_use_id to a per-subagent translator and transcript
under <session>/subagents/<tool_use_id>; three routes expose the list, a
transcript page and the SSE stream. Echo grows /subagent [n] as the rig.

On the phone a card with subagents ends in a chevron expander, collapsed by
default, opening to outlined subcards styled like dev-updater's components;
a subcard opens SessionScreen in read-only form, addressed through
TranscriptAddress so paging, cache and stream are shared.

Design in SUBAGENTS.md; choices awaiting review in DECISIONS.md.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
irisandClaude Fable 5.1 committed 2026-09-05 13:41:15 -04:00
1 parent eff5c8b0c0
commit 9fa09b0af1
21 files changed
+1953 -332
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

server/src/routes.rs
+120 -13
View File
@@ -29,6 +29,12 @@
//! GET /sessions/{id}/transcript a page of history: ?before=N (newest when absent),
//! ?limit=N, ?coalesce=true to count rows not deltas,
//! ?after=N to floor it at what the caller already holds
//! GET /sessions/{id}/subagents [{id, title, status, created, lastActivity}], oldest
//! first -- see SUBAGENTS.md
//! GET /sessions/{id}/subagents/{sub}/transcript exactly the transcript route above,
//! against that subagent's own transcript
//! GET /sessions/{id}/subagents/{sub}/events?after=N exactly the events route above,
//! against that subagent's own stream
//! POST /sessions/{id}/message {text, attachmentIds?}
//! (starts the process first if it has exited)
//! POST /sessions/{id}/unqueue {messageId} -- take back one not read yet
@@ -99,6 +105,7 @@ use tokio_stream::wrappers::{BroadcastStream, ReceiverStream};
use crate::session::driver::{SessionCommand, Unqueued};
use crate::session::pending::Operation;
use crate::session::subagent::{Subagent, SubagentInfo};
use crate::session::transcript::{CATCH_UP_LIMIT, CatchUp, SeqEvent, catch_up};
use crate::session::{LiveSession, SessionInfo, SessionManager, SpawnSpec};
@@ -130,6 +137,15 @@ pub fn router(manager: Arc<SessionManager>) -> Router {
.route("/sessions/{id}", get(read_session).delete(delete_session))
.route("/sessions/{id}/events", get(events))
.route("/sessions/{id}/transcript", get(transcript))
.route("/sessions/{id}/subagents", get(list_subagents))
.route(
"/sessions/{id}/subagents/{sub}/transcript",
get(subagent_transcript),
)
.route(
"/sessions/{id}/subagents/{sub}/events",
get(subagent_events),
)
.route("/sessions/{id}/message", post(message))
.route("/sessions/{id}/unqueue", post(unqueue))
.route("/sessions/{id}/answer", post(answer))
@@ -209,6 +225,18 @@ fn lookup(manager: &SessionManager, id: &str) -> Result<Arc<LiveSession>, ApiErr
.ok_or_else(|| ApiError::NotFound(format!("no session {id}")))
}
/// A session's subagent by id -- the second half of the lookup every
/// `/sessions/{id}/subagents/{sub}/...` route needs. `Arc` because reopening
/// one from disk (a subagent this process has not touched yet) inserts it
/// into the registry, and a route holding a borrow across that would be
/// holding the registry's lock the whole request.
fn lookup_subagent(session: &LiveSession, sub: &str) -> Result<Arc<Subagent>, ApiError> {
session
.subagents()
.get(sub)
.ok_or_else(|| ApiError::NotFound(format!("no subagent {sub}")))
}
async fn list_sessions(State(manager): State<Arc<SessionManager>>) -> axum::Json<Vec<SessionInfo>> {
axum::Json(manager.sessions())
}
@@ -1747,8 +1775,36 @@ async fn transcript(
Query(query): Query<TranscriptQuery>,
) -> Result<axum::Json<Vec<crate::session::transcript::SeqEvent>>, ApiError> {
let session = lookup(&manager, &id)?;
transcript_page(session.transcript_path(), &id, query)
}
/// Exactly [`transcript`]'s route and answer, against one subagent's own
/// transcript instead of its session's -- see `SUBAGENTS.md`'s wire shape.
async fn subagent_transcript(
State(manager): State<Arc<SessionManager>>,
UrlPath((id, sub)): UrlPath<(String, String)>,
Query(query): Query<TranscriptQuery>,
) -> Result<axum::Json<Vec<crate::session::transcript::SeqEvent>>, ApiError> {
let session = lookup(&manager, &id)?;
let subagent = lookup_subagent(&session, &sub)?;
transcript_page(
&subagent.transcript_path(),
&format!("{id}/subagents/{sub}"),
query,
)
}
/// A page of history at `path`, newest first to open with -- the one
/// implementation [`transcript`] and [`subagent_transcript`] share, since a
/// subagent's transcript is read exactly the way a session's is. `label` is
/// only for the debug line below.
fn transcript_page(
path: &Path,
label: &str,
query: TranscriptQuery,
) -> Result<axum::Json<Vec<crate::session::transcript::SeqEvent>>, ApiError> {
let events = crate::session::transcript::read_window(
session.transcript_path(),
path,
query.before,
query.after,
query.limit,
@@ -1760,7 +1816,7 @@ async fn transcript(
// for events and draws rows, and the ratio between them is a property of
// the conversation. `RUST_LOG=ai_server=debug`.
tracing::debug!(
session = %id,
session = %label,
before = ?query.before,
after = ?query.after,
limit = query.limit,
@@ -1778,23 +1834,74 @@ async fn events(
headers: HeaderMap,
) -> Result<Sse<impl tokio_stream::Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
let session = lookup(&manager, &id)?;
let cursor = headers
.get("last-event-id")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse().ok())
.unwrap_or(query.after);
let cursor = cursor_of(&headers, query.after);
// Subscribe before reading the file so nothing can land in the gap
// between replay and live; overlap is deduplicated by seq.
let live = session.subscribe();
let (tx, stream) = mpsc::channel(64);
tokio::spawn(stream_session(
Ok(sse_stream(
session.transcript_path().to_path_buf(),
cursor,
live,
tx,
));
Ok(Sse::new(ReceiverStream::new(stream).map(Ok)).keep_alive(KeepAlive::default()))
))
}
/// Exactly [`events`]'s route and answer, against one subagent's own stream
/// instead of its session's -- see `SUBAGENTS.md`'s wire shape.
async fn subagent_events(
State(manager): State<Arc<SessionManager>>,
UrlPath((id, sub)): UrlPath<(String, String)>,
Query(query): Query<EventsQuery>,
headers: HeaderMap,
) -> Result<Sse<impl tokio_stream::Stream<Item = Result<SseEvent, Infallible>>>, ApiError> {
let session = lookup(&manager, &id)?;
let subagent = lookup_subagent(&session, &sub)?;
let cursor = cursor_of(&headers, query.after);
let live = subagent.subscribe();
Ok(sse_stream(subagent.transcript_path(), cursor, live))
}
/// The cursor an SSE reconnect resumes from: the native `Last-Event-ID`
/// takes precedence over the query parameter, same cursor either way.
fn cursor_of(headers: &HeaderMap, query_after: u64) -> u64 {
headers
.get("last-event-id")
.and_then(|value| value.to_str().ok())
.and_then(|value| value.parse().ok())
.unwrap_or(query_after)
}
/// Spawns the backlog-then-live task and wraps it as the response, the one
/// piece [`events`] and [`subagent_events`] share.
fn sse_stream(
transcript: PathBuf,
cursor: u64,
live: broadcast::Receiver<SeqEvent>,
) -> Sse<impl tokio_stream::Stream<Item = Result<SseEvent, Infallible>>> {
let (tx, stream) = mpsc::channel(64);
tokio::spawn(stream_session(transcript, cursor, live, tx));
Sse::new(ReceiverStream::new(stream).map(Ok)).keep_alive(KeepAlive::default())
}
/// `GET /sessions/{id}/subagents`: every subagent this session has started,
/// oldest first, with a status read from its own transcript -- see
/// `SUBAGENTS.md`'s wire shape. A subagent whose last status is `Running` is
/// reported `Unknown` instead when the session itself is not running: its
/// process was the session's, and a session with none has nothing left to
/// ask.
async fn list_subagents(
State(manager): State<Arc<SessionManager>>,
UrlPath(id): UrlPath<String>,
) -> Result<axum::Json<Vec<SubagentInfo>>, ApiError> {
let session = lookup(&manager, &id)?;
// Anything but `Exited` or `Unknown` has a process behind it, which is
// what decides whether a subagent still reading `Running` from its own
// transcript can be believed -- see `SUBAGENTS.md`'s wire shape.
let running = !matches!(
session.status(),
crate::session::driver::SessionStatus::Exited
| crate::session::driver::SessionStatus::Unknown
);
Ok(axum::Json(session.subagents().list(running)))
}
/// Every session's attention-wanting moments, on one stream.
server/src/session/claude.rs
+22 -5
View File
@@ -59,6 +59,7 @@ use tokio::sync::mpsc;
use super::driver::{AttachmentRef, Driver, Event, EventSink, SessionStatus, Unqueued};
use super::process;
use super::subagent::Subagents;
use super::transport::{Launch, Streams, Transport};
use crate::config::{ProviderConfig, SessionConfig};
use translate::{AnswerOutcome, Setting, Translator, starts_a_model_call};
@@ -213,8 +214,12 @@ impl ClaudeDriver {
transport: &Transport,
session_dir: &Path,
sink: EventSink,
subagents: Arc<Subagents>,
) -> Result<Self> {
let state = Arc::new(Mutex::new(Translator::new(session_dir.to_path_buf())));
let state = Arc::new(Mutex::new(Translator::new(
session_dir.to_path_buf(),
subagents,
)));
let queue = Arc::new(Mutex::new(Queue::default()));
let reading = Arc::new(AtomicBool::new(true));
@@ -1169,7 +1174,10 @@ mod tests {
/// this" and "the transcript records that".
fn events_from_lines(lines: &[&str]) -> Vec<Event> {
let dir = tempfile::tempdir().expect("temp dir");
let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
let state = Arc::new(Mutex::new(Translator::new(
dir.path().to_path_buf(),
Arc::new(Subagents::new(dir.path().to_path_buf())),
)));
let queue = Arc::new(Mutex::new(Queue::default()));
let (sink, mut out) = mpsc::unbounded_channel::<Event>();
for line in lines {
@@ -1193,7 +1201,10 @@ mod tests {
interject: impl FnOnce(&Arc<Mutex<Queue>>),
) -> Vec<Event> {
let dir = tempfile::tempdir().expect("temp dir");
let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
let state = Arc::new(Mutex::new(Translator::new(
dir.path().to_path_buf(),
Arc::new(Subagents::new(dir.path().to_path_buf())),
)));
let queue = Arc::new(Mutex::new(Queue::default()));
let (sink, mut out) = mpsc::unbounded_channel::<Event>();
let mut interject = Some(interject);
@@ -1487,7 +1498,10 @@ mod tests {
// 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 state = Arc::new(Mutex::new(Translator::new(
dir.path().to_path_buf(),
Arc::new(Subagents::new(dir.path().to_path_buf())),
)));
let queue = Arc::new(Mutex::new(Queue::default()));
let text = r#"{"type":"stream_event","event":{"type":"content_block_delta","index":0,"delta":{"type":"text_delta","text":"working"}},"parent_tool_use_id":null}"#;
@@ -1532,7 +1546,10 @@ mod tests {
// session back to work.
let dir = tempfile::tempdir().expect("tempdir");
let (sink, mut received) = mpsc::unbounded_channel();
let state = Arc::new(Mutex::new(Translator::new(dir.path().to_path_buf())));
let state = Arc::new(Mutex::new(Translator::new(
dir.path().to_path_buf(),
Arc::new(Subagents::new(dir.path().to_path_buf())),
)));
let queue = Arc::new(Mutex::new(Queue::default()));
queue.lock().unwrap().close(&sink, "the session ended");
server/src/session/claude/translate.rs
+268 -36
View File
@@ -11,10 +11,12 @@
use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use serde_json::{Value, json};
use super::super::driver::{Event, QuestionOption, SessionStatus, context_tokens};
use super::super::subagent::Subagents;
/// Whether this line is the CLI opening a fresh model call.
///
@@ -92,10 +94,20 @@ pub(super) struct Translator {
/// reports none rather than repeating the previous turn's.
context: Option<u64>,
session_dir: PathBuf,
/// This session's subagents, shared with every child translator below --
/// see `SUBAGENTS.md`. One registry per session, so a subagent started
/// through this translator or any of its children lands in the same
/// place a route reads it back from.
subagents: Arc<Subagents>,
/// One translator per subagent id, holding *its* streaming and
/// tool-tracking state -- separate from the parent's because tool ids
/// are unique but a `stream_event`'s content-block index is not, and
/// parallel subagents interleave their deltas on one stdout.
children: HashMap<String, Arc<Mutex<Translator>>>,
}
impl Translator {
pub(super) fn new(session_dir: PathBuf) -> Self {
pub(super) fn new(session_dir: PathBuf, subagents: Arc<Subagents>) -> Self {
Self {
session_id: None,
pending: HashMap::new(),
@@ -103,6 +115,8 @@ impl Translator {
interrupting: false,
context: None,
session_dir,
subagents,
children: HashMap::new(),
}
}
@@ -122,13 +136,54 @@ impl Translator {
pub(super) fn translate(&mut self, message: &Value) -> Vec<Event> {
// Events from subagents (Task tool internals) carry a
// parent_tool_use_id; the transcript shows the Task tool's own
// start/end instead of every nested step.
if message
.get("parent_tool_use_id")
.is_some_and(|id| !id.is_null())
{
return Vec::new();
// start/end instead of every nested step. Routed into that
// subagent's own transcript rather than dropped -- see
// `SUBAGENTS.md`.
if let Some(parent_id) = message.get("parent_tool_use_id").and_then(Value::as_str) {
return self.translate_child(parent_id, message);
}
self.dispatch(message)
}
/// A line belonging to a subagent rather than to this translator's own
/// session. Always returns nothing to the *caller*: everything it
/// produces goes into the subagent's own transcript instead.
fn translate_child(&mut self, id: &str, message: &Value) -> Vec<Event> {
match self.subagents.get(id) {
Some(subagent) if !subagent.is_open() => {
// The Task call already ended (or this line is stale from a
// resumed conversation) -- see `SUBAGENTS.md`'s lifecycle.
tracing::debug!("dropping a line for subagent {id}, which has already finished");
return Vec::new();
}
Some(_) => {}
None => {
// Nobody has heard of this id yet: the Task call itself
// either has not been seen or never will be. Started here
// with the best title available -- the tool name of this
// first line -- since SUBAGENTS.md's real title only
// arrives with the Task call.
self.subagents.start(id, &fallback_title(message), None);
}
}
let child = self
.children
.entry(id.to_string())
.or_insert_with(|| {
Arc::new(Mutex::new(Translator::new(
self.session_dir.clone(),
Arc::clone(&self.subagents),
)))
})
.clone();
let events = child.lock().unwrap().dispatch(message);
for event in events {
self.subagents.record(id, event);
}
Vec::new()
}
fn dispatch(&mut self, message: &Value) -> Vec<Event> {
match message.get("type").and_then(Value::as_str) {
Some("system") => self.translate_system(message),
// The CLI's own announcement that `/clear` took effect, sent just
@@ -379,22 +434,47 @@ impl Translator {
content
.iter()
.filter(|block| block.get("type").and_then(Value::as_str) == Some("tool_use"))
.map(|block| Event::ToolStart {
id: block
.map(|block| {
let id = block
.get("id")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string(),
tool: block
.to_string();
let tool = block
.get("name")
.and_then(Value::as_str)
.unwrap_or_default()
.to_string(),
input: block.get("input").cloned().unwrap_or(Value::Null),
.to_string();
let input = block.get("input").cloned().unwrap_or(Value::Null);
// A subagent this call is about to start -- see
// `SUBAGENTS.md`'s lifecycle #1. The parent's own transcript
// still shows only the Task call itself, below.
if tool == "Task" || tool == "Agent" {
self.start_subagent_from_task(&id, &input);
}
Event::ToolStart { id, tool, input }
})
.collect()
}
/// Starts the subagent a Task call names, with the title and prompt
/// SUBAGENTS.md describes: the call's `description`, then
/// `(<subagent_type>)` when one is given, falling back to the tool's own
/// name when there is no description to build one from.
fn start_subagent_from_task(&self, id: &str, input: &Value) {
let description = text_field(input, "description");
let subagent_type = text_field(input, "subagent_type");
let prompt = input.get("prompt").and_then(Value::as_str);
let title = match (description, subagent_type) {
(Some(description), Some(subagent_type)) => {
format!("{description} ({subagent_type})")
}
(Some(description), None) => description,
(None, _) => "Task".to_string(),
};
self.subagents.start(id, &title, prompt);
}
fn translate_control_request(&mut self, message: &Value) -> Vec<Event> {
let request = &message["request"];
if request.get("subtype").and_then(Value::as_str) != Some("can_use_tool") {
@@ -598,11 +678,33 @@ impl Translator {
id: about.clone(),
output: texts.join("\n"),
});
// A no-op unless `about` is a subagent's own id -- see
// `SUBAGENTS.md`'s lifecycle #3: the parent gets this `ToolEnd`
// like any other tool result, and the subagent it names (if it
// names one) gets its `Status::Exited`.
self.subagents.finish(&about);
}
events
}
}
/// The title to start a subagent under when its own first line arrives
/// before (or without) its Task call ever being seen: the tool name of that
/// first line, which is the only thing known about it yet. `"subagent"` for
/// a line this cannot even find a tool name in, such as one that opens with
/// something other than a tool call.
fn fallback_title(message: &Value) -> String {
message["message"]["content"]
.as_array()
.into_iter()
.flatten()
.find(|block| block.get("type").and_then(Value::as_str) == Some("tool_use"))
.and_then(|block| block.get("name"))
.and_then(Value::as_str)
.unwrap_or("subagent")
.to_string()
}
/// Whether a failed turn failed because the account is out of quota, and when
/// the CLI said the limit lifts.
///
@@ -696,10 +798,18 @@ mod tests {
.collect()
}
/// A fresh, empty subagent registry over the same temp dir a test's
/// translator writes into -- every test here is about the parent's own
/// events, so what a registry does with a subagent is `subagent.rs`'s
/// tests to make, not these.
fn test_subagents(dir: &tempfile::TempDir) -> Arc<Subagents> {
Arc::new(Subagents::new(dir.path().to_path_buf()))
}
#[test]
fn captures_the_resume_token_and_the_settings_from_init() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -721,7 +831,7 @@ mod tests {
#[test]
fn a_setting_is_reported_when_the_cli_accepts_it_and_not_before() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
// What `set_model` does: remember, send, and say nothing yet.
translator.expect_setting("req-a".to_string(), Setting::Model("sonnet".to_string()));
@@ -798,7 +908,7 @@ mod tests {
// The line it sends just after answering `set_permission_mode`, which is
// also how a mode changed from the terminal arrives.
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -818,7 +928,7 @@ mod tests {
fn streams_text_deltas_and_skips_the_consolidated_copy() {
// Real lines (trimmed) from the 2.1.237 probe.
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -838,7 +948,7 @@ mod tests {
#[test]
fn tool_use_and_result_become_tool_events() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -865,7 +975,7 @@ mod tests {
#[test]
fn subagent_events_are_not_duplicated_into_the_transcript() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -875,10 +985,132 @@ mod tests {
assert!(events.is_empty());
}
/// A child line does not just vanish from the parent -- it lands in its
/// own subagent's transcript, with that transcript's own sequence
/// numbers, starting at 1 like any other.
#[test]
fn a_child_line_lands_in_its_own_subagents_transcript() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = test_subagents(&dir);
let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
translate_lines(
&mut translator,
&[
r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_c1","name":"Bash","input":{"command":"echo hi"}}]},"parent_tool_use_id":"toolu_parent"}"#,
],
);
let subagent = subagents.get("toolu_parent").expect("subagent started");
let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0)
.expect("read subagent transcript");
assert_eq!(lines[0].seq, 1);
assert_eq!(
lines[0].event,
Event::Status {
state: SessionStatus::Running
}
);
assert!(
lines.iter().any(
|entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Bash")
)
);
}
/// The title and prompt shown for a subagent come from the Task call
/// that started it, not from anything guessed at its first line.
#[test]
fn the_subagent_takes_its_title_and_prompt_from_the_task_call() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = test_subagents(&dir);
let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
translate_lines(
&mut translator,
&[
r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_task","name":"Task","input":{"description":"Investigate the bug","prompt":"Find why X fails","subagent_type":"general-purpose"}}]},"parent_tool_use_id":null}"#,
],
);
let rows = subagents.list(true);
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].title, "Investigate the bug (general-purpose)");
let subagent = subagents.get(&rows[0].id).expect("subagent");
let lines = crate::session::transcript::read_after(&subagent.transcript_path(), 0)
.expect("read subagent transcript");
assert!(lines.iter().any(
|entry| matches!(&entry.event, Event::UserMessage { text, .. } if text == "Find why X fails")
));
}
/// The parent's `tool_result` for the Task id is what ends the
/// subagent -- SUBAGENTS.md's lifecycle #3 -- and nothing else does.
#[test]
fn the_parents_tool_result_finishes_the_subagent() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = test_subagents(&dir);
let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
translate_lines(
&mut translator,
&[
r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_task2","name":"Task","input":{"description":"helper"}}]},"parent_tool_use_id":null}"#,
],
);
let subagent = subagents.get("toolu_task2").expect("subagent started");
assert!(subagent.is_open());
translate_lines(
&mut translator,
&[
r#"{"type":"user","message":{"role":"user","content":[{"type":"tool_result","tool_use_id":"toolu_task2","content":"done","is_error":false}]},"parent_tool_use_id":null}"#,
],
);
assert!(!subagent.is_open());
}
/// Two subagents running at once keep two separate transcripts: tool ids
/// are unique but a `stream_event`'s content-block index is not, so
/// sharing translation state between them would cross their streams.
#[test]
fn two_parallel_subagents_keep_separate_transcripts() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = test_subagents(&dir);
let mut translator = Translator::new(dir.path().to_path_buf(), Arc::clone(&subagents));
translate_lines(
&mut translator,
&[
r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_a","name":"Bash","input":{}}]},"parent_tool_use_id":"toolu_task_a"}"#,
r#"{"type":"assistant","message":{"content":[{"type":"tool_use","id":"toolu_b","name":"Read","input":{}}]},"parent_tool_use_id":"toolu_task_b"}"#,
],
);
let a = subagents.get("toolu_task_a").expect("subagent a");
let b = subagents.get("toolu_task_b").expect("subagent b");
let a_events = crate::session::transcript::read_after(&a.transcript_path(), 0)
.expect("read a's transcript");
let b_events = crate::session::transcript::read_after(&b.transcript_path(), 0)
.expect("read b's transcript");
assert!(
a_events.iter().any(
|entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Bash")
)
);
assert!(
b_events.iter().any(
|entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Read")
)
);
assert!(
!a_events.iter().any(
|entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Read")
)
);
assert!(
!b_events.iter().any(
|entry| matches!(&entry.event, Event::ToolStart { tool, .. } if tool == "Bash")
)
);
}
#[test]
fn a_permission_request_becomes_an_allow_deny_question() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -927,7 +1159,7 @@ mod tests {
#[test]
fn denying_a_permission_sends_deny() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
translate_lines(
&mut translator,
&[
@@ -945,7 +1177,7 @@ mod tests {
// The real 2.1.237 shape, verified live: answers go back inside
// updatedInput, keyed by the question text.
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -997,7 +1229,7 @@ mod tests {
// in the event: a phone that had to read this dialect's tool input to
// find them would be the only place that knew how.
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1045,7 +1277,7 @@ mod tests {
#[test]
fn images_in_tool_results_are_saved_and_referenced() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
// A 1x1 PNG, the smallest real payload worth round-tripping.
let png = "iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAYAAAAfFcSJAAAADUlEQVR42mNk+M9QDwADhgGAWjR9awAAAABJRU5ErkJggg==";
let line = format!(
@@ -1074,7 +1306,7 @@ mod tests {
#[test]
fn a_turn_result_reports_usage_and_returns_to_idle() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1109,7 +1341,7 @@ mod tests {
#[test]
fn a_turn_started_by_another_agent_records_who_and_what() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1143,7 +1375,7 @@ mod tests {
#[test]
fn an_ordinary_turn_carries_no_peer_note() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1168,7 +1400,7 @@ mod tests {
#[test]
fn the_context_is_what_the_last_message_held_not_the_turn_added_up() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1208,7 +1440,7 @@ mod tests {
// Note the snake_case keys -- the CLI's transcript file writes the same
// records in camelCase.
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1238,7 +1470,7 @@ mod tests {
#[test]
fn a_failed_compaction_says_why_and_leaves_the_turn_running() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1265,7 +1497,7 @@ mod tests {
#[test]
fn a_boundary_without_counts_says_so_rather_than_inventing_them() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1285,7 +1517,7 @@ mod tests {
#[test]
fn an_error_result_surfaces_the_message() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1315,7 +1547,7 @@ mod tests {
#[test]
fn a_turn_stopped_by_the_usage_limit_says_so_and_carries_the_reset() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1333,7 +1565,7 @@ mod tests {
#[test]
fn a_limit_the_cli_gave_no_reset_for_is_reported_without_one() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
@@ -1366,7 +1598,7 @@ mod tests {
#[test]
fn a_turn_stopped_on_purpose_is_not_an_error() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let stopped_result = r#"{"type":"result","subtype":"error_during_execution","is_error":true,"result":"Interrupted by user","usage":{}}"#;
translator.expect_interrupt();
@@ -1398,7 +1630,7 @@ mod tests {
#[test]
fn replayed_and_synthetic_user_text_is_skipped() {
let dir = tempfile::tempdir().expect("tempdir");
let mut translator = Translator::new(dir.path().to_path_buf());
let mut translator = Translator::new(dir.path().to_path_buf(), test_subagents(&dir));
let events = translate_lines(
&mut translator,
&[
server/src/session/echo.rs
+110 -1
View File
@@ -48,6 +48,10 @@
//! and height the app draws, in one session, which is what a scrolling
//! problem needs in order to be reproduced twice the same way.
//! - `/table [columns]` -- a markdown table with cells too long for one line.
//! - `/subagent [n]` -- n subagents at once (default 1), each named
//! "helper k", its prompt recorded as its own first user message: a
//! streamed reply, one Bash call, then it finishes about three seconds
//! later, the same lifecycle a real Task call has -- see `SUBAGENTS.md`.
//!
//! `/slow` earns its place: a queued message, a Stop button and a spinner are
//! states that only exist mid-turn, and the obvious way to get one -- ask a
@@ -62,6 +66,7 @@ use std::time::Duration;
use super::driver::{
AttachmentRef, Driver, Event, EventSink, QuestionOption, SessionStatus, Unqueued,
};
use super::subagent::Subagents;
/// Delay between streamed deltas -- long enough that streaming is visibly
/// streaming, short enough that tests waiting on a full turn stay fast.
@@ -108,6 +113,10 @@ pub struct EchoDriver {
/// says it recovered, and a clear leaves it unmeasured. What is real is
/// which way the numbers move.
context: Arc<AtomicU64>,
/// This session's subagents -- see `SUBAGENTS.md`. `/subagent` is the
/// test rig for the same registry the claude driver routes real Task
/// calls into.
subagents: Arc<Subagents>,
}
impl EchoDriver {
@@ -384,6 +393,55 @@ impl EchoDriver {
return;
}
// `n` subagents at once, each with its own transcript in the
// registry a real Task call routes into -- see `SUBAGENTS.md`. The
// parent's own Task calls end when their subagent does, three
// seconds later, which is long enough to see the running state on
// the phone before it finishes.
if let Some(rest) = text.strip_prefix("/subagent") {
let n = rest.trim().parse::<usize>().unwrap_or(1).clamp(1, 8);
if announce {
self.emit(Event::MessageTaken {
id: None,
text: text.clone(),
attachments,
});
}
self.emit(Event::Status {
state: SessionStatus::Running,
});
let sink = self.sink.clone();
let subagents = Arc::clone(&self.subagents);
tokio::spawn(async move {
let mut helpers = Vec::new();
for k in 1..=n {
let id = format!("echo-subagent-{k}-{}", super::random_hex());
let title = format!("helper {k}");
let prompt = format!(
"You are helper {k} of {n}. Say a few words, run a command, then stop."
);
let _ = sink.send(Event::ToolStart {
id: id.clone(),
tool: "Task".to_string(),
input: serde_json::json!({
"description": title,
"prompt": prompt,
"subagent_type": "general-purpose",
}),
});
subagents.start(&id, &title, Some(&prompt));
helpers.push((id, sink.clone(), Arc::clone(&subagents)));
}
for (id, sink, subagents) in helpers {
tokio::spawn(run_helper(id, sink, subagents));
}
let _ = sink.send(Event::Status {
state: SessionStatus::Idle,
});
});
return;
}
// The same word the real CLI takes, so a phone drives both the same way.
// `Driver::compact` is what the manager's route calls; this is the typed
// path onto it.
@@ -679,7 +737,12 @@ impl EchoDriver {
});
}
pub fn new(sink: EventSink, session_dir: PathBuf, usage: crate::usage::Fixture) -> Self {
pub fn new(
sink: EventSink,
session_dir: PathBuf,
usage: crate::usage::Fixture,
subagents: Arc<Subagents>,
) -> Self {
let driver = Self {
sink,
pending_questions: Mutex::new(Vec::new()),
@@ -688,6 +751,7 @@ impl EchoDriver {
queued: Arc::new(Mutex::new(Vec::new())),
session_dir,
usage,
subagents,
};
driver.emit(Event::Status {
state: SessionStatus::Idle,
@@ -809,6 +873,51 @@ async fn write_beat(sink: &EventSink, session_dir: &Path, beat: usize) {
tokio::time::sleep(Duration::from_millis(120)).await;
}
/// One `/subagent` helper: a few streamed words, one Bash call, then
/// `Status::Exited` about three seconds after it started -- long enough that
/// its `Running` state can be seen on the phone before it finishes. The
/// parent's own Task call for it ends at the same moment, the same way a
/// real Task's `tool_result` ends it.
async fn run_helper(id: String, sink: EventSink, subagents: Arc<Subagents>) {
let start = tokio::time::Instant::now();
for word in "Working on it now.".split_inclusive(' ') {
subagents.record(
&id,
Event::AssistantText {
delta: word.to_string(),
},
);
tokio::time::sleep(DELTA_DELAY).await;
}
let tool_id = format!("{id}-bash");
subagents.record(
&id,
Event::ToolStart {
id: tool_id.clone(),
tool: "Bash".to_string(),
input: serde_json::json!({ "command": "echo helper done" }),
},
);
tokio::time::sleep(DELTA_DELAY).await;
subagents.record(
&id,
Event::ToolEnd {
id: tool_id,
output: "helper done".to_string(),
},
);
let target = Duration::from_secs(3);
let elapsed = start.elapsed();
if elapsed < target {
tokio::time::sleep(target - elapsed).await;
}
subagents.finish(&id);
let _ = sink.send(Event::ToolEnd {
id,
output: "subagent finished".to_string(),
});
}
/// A message written during a turn and waiting for it to end: the id of the
/// `MessageQueued` that announced it, what it said, and what was attached. All
/// three, because all three are what the `MessageTaken` at the other end owes.
server/src/session/llama.rs
+5
View File
@@ -77,6 +77,7 @@ impl LlamaDriver {
/// in a different currency: two servers holding the same model is twice the
/// memory, and the second would bind a different port while the phone kept
/// talking to the first.
#[allow(clippy::too_many_arguments)]
pub fn launch(
meta: &SessionConfig,
provider: &ProviderConfig,
@@ -85,6 +86,10 @@ impl LlamaDriver {
transcript: &Path,
session_dir: &Path,
sink: EventSink,
// llama.cpp has no notion of a Task call, so this is accepted only
// to keep one shape across every driver's launch -- see
// `SUBAGENTS.md`'s "Server layout".
_subagents: Arc<super::subagent::Subagents>,
) -> Result<Self> {
let model = meta.model.as_deref().context(
"a llama.cpp session needs a model -- one of the downloaded ones, by its key",
server/src/session/mod.rs
+127 -2
View File
@@ -15,6 +15,7 @@ pub mod import;
pub mod llama;
pub mod pending;
pub mod process;
pub mod subagent;
pub mod transcript;
pub mod transport;
@@ -37,6 +38,7 @@ use driver::{
};
use echo::EchoDriver;
use llama::LlamaDriver;
use subagent::Subagents;
use transcript::{SeqEvent, Transcript};
use transport::Transport;
@@ -265,6 +267,11 @@ pub struct SessionInfo {
pub status: SessionStatus,
pub last_activity: f64,
pub created: f64,
/// How many subagents this session has started, from a directory
/// listing rather than reading each one's status -- see
/// `GET /sessions/{id}/subagents` for that. 0 when it has none, not
/// absent: every session can say this without asking anything.
pub subagents: usize,
}
/// What is running a session at this moment, and `None` when nothing is.
@@ -293,6 +300,10 @@ pub struct LiveSession {
events: broadcast::Sender<SeqEvent>,
transcript_path: PathBuf,
shared: Arc<Shared>,
/// This session's subagents -- see `SUBAGENTS.md`. Built once at launch
/// and handed to whichever driver replaces it across a stop/start, so a
/// subagent started before a Stop is still there to read after a Start.
subagents: Arc<Subagents>,
}
/// Commands waiting for the session to be between turns.
@@ -502,6 +513,19 @@ impl LiveSession {
&self.transcript_path
}
pub fn subagents(&self) -> &Arc<Subagents> {
&self.subagents
}
/// What this session is doing right now, as the pump last recorded it --
/// the same word `SessionInfo::status` reports. Read here rather than
/// only through `SessionManager::sessions` for
/// `GET /sessions/{id}/subagents`, which needs exactly this and nothing
/// else `SessionInfo` carries.
pub fn status(&self) -> SessionStatus {
*self.shared.status.lock().unwrap()
}
/// The session's directory (attachments in, produced files out live in
/// `attachments/` and `files/` under it).
pub fn dir(&self) -> &Path {
@@ -578,6 +602,7 @@ impl LiveSession {
status: *self.shared.status.lock().unwrap(),
last_activity: *self.shared.last_activity.lock().unwrap(),
created: self.meta.created,
subagents: subagent::count(self.dir()),
}
}
}
@@ -1065,6 +1090,7 @@ impl SessionManager {
status: status_of_unlaunched(&self.data_dir.join(&meta.id)),
last_activity: meta.created,
created: meta.created,
subagents: subagent::count(&self.data_dir.join(&meta.id)),
},
})
.collect()
@@ -1828,6 +1854,7 @@ impl SessionManager {
session.dir(),
session.transcript_path(),
&session.sink,
session.subagents(),
)?);
}
// Nothing is live for this one -- a session whose launch failed
@@ -2289,6 +2316,11 @@ fn launch(
let (sink, source) = mpsc::unbounded_channel();
let (events, _) = broadcast::channel(EVENT_BUFFER);
// Built once per session, here, rather than per driver: a subagent
// started before a Stop has to still be there to read after a Start,
// and only `launch` runs once across that boundary -- `start_if_exited`
// replaces the driver alone.
let subagents = Arc::new(subagent::Subagents::new(dir.clone()));
let shared = Arc::new(Shared {
// What it was last known to be doing, not an assumption. A driver
// that has something to say corrects this within its first poll.
@@ -2350,7 +2382,18 @@ fn launch(
let driver = Arc::new(Mutex::new(
driving
.then(|| make_driver(&meta, setup, provider, env, &dir, &transcript_path, &sink))
.then(|| {
make_driver(
&meta,
setup,
provider,
env,
&dir,
&transcript_path,
&sink,
&subagents,
)
})
.transpose()?,
));
@@ -2368,6 +2411,7 @@ fn launch(
events.clone(),
Arc::clone(&commands),
announce,
Arc::clone(&subagents),
));
Ok(Arc::new(LiveSession {
@@ -2378,6 +2422,7 @@ fn launch(
events,
transcript_path,
shared,
subagents,
}))
}
@@ -2388,6 +2433,7 @@ fn launch(
/// what [`SessionManager::start_session`] builds. That path replaces the
/// driver and nothing else, so it has to construct one the same way rather
/// than becoming a second answer to "what runs this".
#[allow(clippy::too_many_arguments)]
fn make_driver(
meta: &SessionConfig,
setup: &SetupConfig,
@@ -2396,13 +2442,17 @@ fn make_driver(
dir: &Path,
transcript_path: &Path,
sink: &EventSink,
subagents: &Arc<Subagents>,
) -> Result<Arc<dyn Driver>> {
Ok(match provider.kind {
DriverKind::Echo => Arc::new(EchoDriver::new(
sink.clone(),
dir.to_path_buf(),
env.usage.clone(),
Arc::clone(subagents),
)),
// llama.cpp has no notion of a Task call, so it takes the registry
// and never touches it -- see `SUBAGENTS.md`'s "Server layout".
DriverKind::LlamaCpp => Arc::new(LlamaDriver::launch(
meta,
provider,
@@ -2411,6 +2461,7 @@ fn make_driver(
transcript_path,
dir,
sink.clone(),
Arc::clone(subagents),
)?),
DriverKind::ClaudeCli => Arc::new(ClaudeDriver::launch(
meta,
@@ -2418,6 +2469,7 @@ fn make_driver(
&Transport::for_setup(setup),
dir,
sink.clone(),
Arc::clone(subagents),
)?),
})
}
@@ -2482,6 +2534,7 @@ fn notification_for(
}
}
#[allow(clippy::too_many_arguments)]
async fn pump(
id: String,
mut transcript: Transcript,
@@ -2490,6 +2543,7 @@ async fn pump(
events: broadcast::Sender<SeqEvent>,
commands: Arc<Commands>,
announce: Announcements,
subagents: Arc<Subagents>,
) {
// Messages the session has been given and not started reading, which is
// what makes a turn ending not the same thing as the work ending.
@@ -2611,7 +2665,12 @@ async fn pump(
} => commands.take_one(),
Event::Status {
state: SessionStatus::Exited,
} => commands.abandon("this session's process has exited"),
} => {
commands.abandon("this session's process has exited");
// The process behind every open subagent was this
// session's own -- see `SUBAGENTS.md`'s lifecycle #4.
subagents.finish_all();
}
// The two ends of a message's wait. A `UserMessage` with
// no id never waited -- it was sent between turns, and
// counting it would take the total below zero.
@@ -2752,6 +2811,7 @@ mod tests {
sink.clone(),
dir.path().to_path_buf(),
crate::usage::Fixture::new(),
Arc::new(subagent::Subagents::new(dir.path().to_path_buf())),
))))),
sink,
waiting: Mutex::new(VecDeque::new()),
@@ -4399,4 +4459,69 @@ mod tests {
let seen = collect_turn(&mut rx).await;
assert!(seen.first().expect("events").seq > last_seq);
}
/// `/subagent 2` is the test rig for `SUBAGENTS.md`'s whole feature:
/// each helper gets its own transcript with its prompt as its first
/// user message, `SessionInfo::subagents` counts them from the
/// directory, and each finishes on its own a few seconds later.
#[tokio::test]
async fn subagent_helpers_get_their_own_transcripts_and_finish() {
let dir = tempfile::tempdir().expect("tempdir");
let config_path = dir.path().join("config.ron");
let data_dir = dir.path().join("sessions");
seed_echo_only(&config_path);
let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
.expect("manager");
let info = manager.spawn_session(echo_spec()).expect("spawn");
let session = manager.session(&info.id).expect("live");
session.send_message("/subagent 2".to_string(), Vec::new());
// Both helpers exist as soon as their Task calls go out, well before
// either finishes.
let deadline = tokio::time::Instant::now() + Duration::from_secs(2);
loop {
if session.subagents().list(true).len() == 2 {
break;
}
assert!(
tokio::time::Instant::now() < deadline,
"both helpers should have started by now"
);
tokio::time::sleep(Duration::from_millis(20)).await;
}
let rows = session.subagents().list(true);
let mut titles: Vec<&str> = rows.iter().map(|row| row.title.as_str()).collect();
titles.sort_unstable();
assert_eq!(titles, ["helper 1", "helper 2"]);
assert!(rows.iter().all(|row| row.status == SessionStatus::Running));
assert_eq!(manager.sessions()[0].subagents, 2);
// Each subagent's own transcript opens with its prompt.
let first = session.subagents().get(&rows[0].id).expect("subagent");
let events =
transcript::read_after(&first.transcript_path(), 0).expect("read subagent transcript");
assert!(
events
.iter()
.any(|entry| matches!(&entry.event, Event::UserMessage { text, .. } if text.contains("helper")))
);
// Each finishes on its own about three seconds after it started.
let deadline = tokio::time::Instant::now() + Duration::from_secs(5);
loop {
if session
.subagents()
.list(true)
.iter()
.all(|row| row.status == SessionStatus::Exited)
{
break;
}
assert!(
tokio::time::Instant::now() < deadline,
"both helpers should have finished by now"
);
tokio::time::sleep(Duration::from_millis(50)).await;
}
}
}
server/src/session/subagent.rs
+490
View File
@@ -0,0 +1,490 @@
//! A session's subagents -- see `SUBAGENTS.md`.
//!
//! **A subagent is a second transcript owned by a session, in the same event
//! model, with no process and no controls.** It shares the transcript file
//! format, the paging routes, and the SSE stream with a session by
//! addressing, not by copying: `Transcript`, `read_window` and `catch_up`
//! work on a subagent's file unchanged.
//!
//! Storage is `<session dir>/subagents/<id>/{meta.json,transcript.jsonl}`,
//! where `<id>` is the Task tool_use id that started it -- unique, stable
//! across a backend restart, and already the key the parent side uses. Only
//! ids matching [`is_subagent_id`] are ever turned into a path.
use std::collections::HashMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use anyhow::{Context, Result};
use serde::{Deserialize, Serialize};
use tokio::sync::broadcast;
use super::driver::{Event, SessionStatus};
use super::transcript::{SeqEvent, Transcript};
/// Fan-out buffer for one subagent's SSE subscribers. Smaller than a
/// session's: a subagent's whole conversation is usually a handful of tool
/// calls, not an hours-long session.
const EVENT_BUFFER: usize = 64;
/// Whether `id` is safe to become a path segment under a session's
/// `subagents/` directory. Mirrors `import::is_session_id`'s reasoning: the
/// id arrives as a value inside JSON the CLI sent, and it becomes a
/// directory name, so a `/` or `..` in it must never be trusted.
fn is_subagent_id(id: &str) -> bool {
!id.is_empty()
&& id.len() <= 200
&& id
.bytes()
.all(|b| b.is_ascii_alphanumeric() || b == b'_' || b == b'-')
}
/// What a subagent's directory holds beside its transcript. Small and
/// separate from `Subagent` itself because this is exactly what survives a
/// backend restart on disk, and nothing else does.
#[derive(Debug, Clone, Serialize, Deserialize)]
struct Meta {
title: String,
/// Epoch seconds. Absent from `SubagentInfo`'s sort key deliberately:
/// `list` sorts by this rather than by directory order, which a
/// filesystem does not promise.
created: f64,
}
/// One row of `GET /sessions/{id}/subagents`.
#[derive(Debug, Clone, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct SubagentInfo {
pub id: String,
pub title: String,
pub status: SessionStatus,
pub created: f64,
pub last_activity: f64,
}
/// One subagent: its own transcript and broadcast, same shape as a
/// session's but with no driver behind it.
pub struct Subagent {
dir: PathBuf,
transcript: Mutex<Transcript>,
events: broadcast::Sender<SeqEvent>,
/// Mirrors the transcript's last `Status` event, kept live rather than
/// read back from `Transcript::last_status` -- that answers "as of
/// opening" (see its own doc comment) and never moves for an append made
/// through *this* object, which is every append a live subagent ever
/// makes. Without this, `finish` immediately after `start` in the same
/// process read the file's stale opening status and reported itself
/// still open.
status: Mutex<SessionStatus>,
}
impl Subagent {
pub fn transcript_path(&self) -> PathBuf {
self.dir.join("transcript.jsonl")
}
pub fn subscribe(&self) -> broadcast::Receiver<SeqEvent> {
self.events.subscribe()
}
/// Whether this subagent's last recorded status is not `Exited` --
/// what decides whether a further child line still belongs in its
/// transcript. See `SUBAGENTS.md`'s lifecycle: "a child line whose
/// subagent finished already... is ignored".
pub fn is_open(&self) -> bool {
*self.status.lock().unwrap() != SessionStatus::Exited
}
fn append(&self, event: Event) {
let mut transcript = self.transcript.lock().unwrap();
match transcript.append(event, super::now()) {
Ok(entry) => {
if let Event::Status { state } = &entry.event {
*self.status.lock().unwrap() = *state;
}
// No subscribers is fine; the transcript already has it,
// same as a session's pump.
let _ = self.events.send(entry);
}
Err(err) => tracing::error!("subagent transcript append failed: {err:#}"),
}
}
}
/// Every subagent one session has started, keyed by the Task tool_use id
/// that names it.
///
/// Lives beside a session's driver rather than inside it: a claude driver
/// holds an `Arc` to this and routes child lines into it; echo uses it for
/// its `/subagent` rig; llama ignores it, since it has no notion of a Task
/// call. One instance per live session, built at launch and handed to
/// whichever driver replaces it across a stop/start.
pub struct Subagents {
/// The session's own directory; subagents live under `<dir>/subagents`.
dir: PathBuf,
live: Mutex<HashMap<String, Arc<Subagent>>>,
}
impl Subagents {
pub fn new(session_dir: PathBuf) -> Self {
Self {
dir: session_dir,
live: Mutex::new(HashMap::new()),
}
}
fn subagents_dir(&self) -> PathBuf {
self.dir.join("subagents")
}
/// Opens the subagent named `id`, creating it if this is the first
/// anyone has heard of it -- on disk as well as in memory, so a
/// subagent from before a backend restart is reopened rather than
/// recreated. `title`/`prompt` are used only at creation: reopening an
/// existing one keeps its original title and never repeats the prompt
/// into its transcript a second time.
fn open_or_create(&self, id: &str, title: &str, prompt: Option<&str>) -> Result<Arc<Subagent>> {
let dir = self.subagents_dir().join(id);
let meta_path = dir.join("meta.json");
let existed = meta_path.is_file();
let meta = if existed {
let text = fs::read_to_string(&meta_path)
.with_context(|| format!("read {}", meta_path.display()))?;
serde_json::from_str::<Meta>(&text).context("parse subagent meta")?
} else {
wg_app_link::private::create_dir(&dir)?;
let meta = Meta {
title: title.to_string(),
created: super::now(),
};
wg_app_link::private::write_file(
&meta_path,
serde_json::to_string(&meta)
.context("serialize subagent meta")?
.as_bytes(),
)?;
meta
};
let mut transcript = Transcript::open(&dir.join("transcript.jsonl"))?;
if !existed {
// First lines, in order: the subagent is running the moment it
// exists, and its prompt -- when known -- is genuinely its first
// user turn. Written once, here, so a reopen never repeats them.
transcript.append(
Event::Status {
state: SessionStatus::Running,
},
meta.created,
)?;
if let Some(prompt) = prompt {
transcript.append(
Event::UserMessage {
id: None,
text: prompt.to_string(),
attachments: Vec::new(),
},
meta.created,
)?;
}
}
// A freshly created subagent is running by construction (its only
// lines so far are `Status::Running` and maybe its prompt); a
// reopened one takes whatever the file last said, since this
// `Transcript` has not been appended to yet in this process.
let status = if existed {
transcript.last_status().unwrap_or(SessionStatus::Running)
} else {
SessionStatus::Running
};
let (events, _) = broadcast::channel(EVENT_BUFFER);
Ok(Arc::new(Subagent {
dir,
transcript: Mutex::new(transcript),
events,
status: Mutex::new(status),
}))
}
/// Starts a subagent unless one is already known by this id -- see
/// `SUBAGENTS.md`'s lifecycle: created at the Task call or at the first
/// child line, whichever comes first, and never twice. A bad id is
/// refused rather than turned into a path.
pub fn start(&self, id: &str, title: &str, prompt: Option<&str>) {
if !is_subagent_id(id) {
tracing::debug!("refusing to start a subagent with a bad id {id:?}");
return;
}
let mut live = self.live.lock().unwrap();
if live.contains_key(id) {
return;
}
match self.open_or_create(id, title, prompt) {
Ok(subagent) => {
live.insert(id.to_string(), subagent);
}
Err(err) => tracing::error!("couldn't start subagent {id}: {err:#}"),
}
}
/// The subagent named `id`, reopening it from disk on first use in this
/// process if one is there. `None` for an id nothing has ever started --
/// deliberately not created here, since a route or a routing decision is
/// not the Task call that is supposed to be the only way one begins.
pub fn get(&self, id: &str) -> Option<Arc<Subagent>> {
if !is_subagent_id(id) {
return None;
}
if let Some(existing) = self.live.lock().unwrap().get(id).cloned() {
return Some(existing);
}
if !self.subagents_dir().join(id).join("meta.json").is_file() {
return None;
}
// Title and prompt are ignored: the directory already exists, so
// `open_or_create` reads its own meta rather than using either.
match self.open_or_create(id, "", None) {
Ok(subagent) => {
self.live
.lock()
.unwrap()
.insert(id.to_string(), Arc::clone(&subagent));
Some(subagent)
}
Err(err) => {
tracing::error!("couldn't reopen subagent {id}: {err:#}");
None
}
}
}
/// Appends one event to a subagent's own transcript. A no-op, with a
/// debug log, for an id nothing was started under -- a child line for a
/// subagent this registry never opened is dropped rather than guessed
/// at.
pub fn record(&self, id: &str, event: Event) {
match self.live.lock().unwrap().get(id).cloned() {
Some(subagent) => subagent.append(event),
None => tracing::debug!("dropping an event for unknown subagent {id}"),
}
}
/// The parent's `tool_result` for this Task id arrived: the subagent's
/// own `Status::Exited`. A no-op for an id that is not a subagent's, so
/// callers can call this for every `tool_result` without first checking
/// whether it belongs to one.
pub fn finish(&self, id: &str) {
if let Some(subagent) = self.live.lock().unwrap().get(id).cloned()
&& subagent.is_open()
{
subagent.append(Event::Status {
state: SessionStatus::Exited,
});
}
}
/// The parent session's process is gone, so nothing still open here has
/// a process behind it either -- see `SUBAGENTS.md`'s lifecycle #4.
pub fn finish_all(&self) {
let subagents: Vec<Arc<Subagent>> = self.live.lock().unwrap().values().cloned().collect();
for subagent in subagents {
if subagent.is_open() {
subagent.append(Event::Status {
state: SessionStatus::Exited,
});
}
}
}
/// Every subagent under this session's directory, oldest first --
/// `GET /sessions/{id}/subagents`. Read straight from disk rather than
/// from `live`, so a subagent from before this process started (or one
/// this run has not yet touched) still shows up; one file read per
/// subagent, which is fine at the handful a session usually has.
///
/// `session_running` is what turns a subagent whose last status is
/// `Running` into `Unknown`: its process was the session's, and the
/// session has none.
pub fn list(&self, session_running: bool) -> Vec<SubagentInfo> {
let mut rows: Vec<SubagentInfo> = match fs::read_dir(self.subagents_dir()) {
Ok(entries) => entries
.filter_map(Result::ok)
.filter_map(|entry| info_of(&entry.path(), session_running))
.collect(),
// No directory is no subagents, not a fault worth reporting.
Err(_) => Vec::new(),
};
rows.sort_by(|a, b| {
a.created
.partial_cmp(&b.created)
.unwrap_or(std::cmp::Ordering::Equal)
});
rows
}
}
fn info_of(subagent_dir: &Path, session_running: bool) -> Option<SubagentInfo> {
let id = subagent_dir.file_name()?.to_str()?.to_string();
let meta_path = subagent_dir.join("meta.json");
let text = fs::read_to_string(&meta_path).ok()?;
let meta: Meta = serde_json::from_str(&text).ok()?;
let transcript = Transcript::open(&subagent_dir.join("transcript.jsonl")).ok()?;
// The subagent's first line is always `Status::Running`, written before
// this directory is discoverable at all, so `None` here is not a state a
// reader can actually observe -- but it is not this function's place to
// invent one, so a status this build does not expect to see falls back
// to the word the lifecycle promises it started in.
let last_status = transcript.last_status().unwrap_or(SessionStatus::Running);
let status = if last_status == SessionStatus::Running && !session_running {
SessionStatus::Unknown
} else {
last_status
};
Some(SubagentInfo {
id,
title: meta.title,
status,
created: meta.created,
last_activity: transcript.last_activity().unwrap_or(meta.created),
})
}
/// How many subagents a session has, for `SessionInfo::subagents`: a
/// directory listing, so the session list stays cheap and only the
/// dedicated route pays for reading a status out of each one.
pub fn count(session_dir: &Path) -> usize {
fs::read_dir(session_dir.join("subagents"))
.map(|entries| entries.filter_map(Result::ok).count())
.unwrap_or(0)
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn a_bad_id_is_refused_rather_than_turned_into_a_path() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = Subagents::new(dir.path().to_path_buf());
subagents.start("../../etc", "escape", None);
assert!(subagents.get("../../etc").is_none());
assert!(!dir.path().join("subagents").exists());
}
#[test]
fn starting_twice_keeps_the_first_title_and_does_not_repeat_the_prompt() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = Subagents::new(dir.path().to_path_buf());
subagents.start("toolu_1", "first title", Some("do the thing"));
subagents.start("toolu_1", "second title", Some("do the thing"));
let rows = subagents.list(true);
assert_eq!(rows.len(), 1);
assert_eq!(rows[0].title, "first title");
let events = crate::session::transcript::read_after(
&subagents.get("toolu_1").unwrap().transcript_path(),
0,
)
.expect("read");
assert_eq!(
events
.iter()
.filter(|e| matches!(e.event, Event::UserMessage { .. }))
.count(),
1
);
}
#[test]
fn a_reopened_subagent_continues_its_own_transcript() {
let dir = tempfile::tempdir().expect("tempdir");
{
let subagents = Subagents::new(dir.path().to_path_buf());
subagents.start("toolu_2", "helper", Some("go"));
subagents.record(
"toolu_2",
Event::AssistantText {
delta: "working".to_string(),
},
);
}
// A fresh registry, the way a backend restart builds one.
let subagents = Subagents::new(dir.path().to_path_buf());
let subagent = subagents.get("toolu_2").expect("reopened");
assert!(subagent.is_open());
subagents.record(
"toolu_2",
Event::AssistantText {
delta: " more".to_string(),
},
);
let events =
crate::session::transcript::read_after(&subagent.transcript_path(), 0).expect("read");
// Status, UserMessage, two AssistantText deltas, seq continuing.
assert_eq!(events.len(), 4);
assert_eq!(events.last().unwrap().seq, 4);
}
#[test]
fn finishing_appends_exited_and_further_lines_are_droppable() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = Subagents::new(dir.path().to_path_buf());
subagents.start("toolu_3", "helper", None);
subagents.finish("toolu_3");
let subagent = subagents.get("toolu_3").unwrap();
assert!(!subagent.is_open());
// On disk too, not only in the live cache `is_open` reads.
assert_eq!(
Transcript::open(&subagent.transcript_path())
.expect("reopen")
.last_status(),
Some(SessionStatus::Exited)
);
// Finishing an id that was never a subagent is a no-op, not a panic.
subagents.finish("never-started");
}
#[test]
fn finish_all_closes_only_what_is_still_open() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = Subagents::new(dir.path().to_path_buf());
subagents.start("toolu_4", "one", None);
subagents.start("toolu_5", "two", None);
subagents.finish("toolu_4");
subagents.finish_all();
let rows = subagents.list(false);
assert_eq!(rows.len(), 2);
for row in rows {
assert_eq!(row.status, SessionStatus::Exited);
}
}
#[test]
fn a_subagent_still_running_when_the_session_is_not_reports_unknown() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = Subagents::new(dir.path().to_path_buf());
subagents.start("toolu_6", "helper", None);
assert_eq!(subagents.list(true)[0].status, SessionStatus::Running);
assert_eq!(subagents.list(false)[0].status, SessionStatus::Unknown);
}
#[test]
fn list_is_oldest_first_and_the_count_matches_the_directory() {
let dir = tempfile::tempdir().expect("tempdir");
let subagents = Subagents::new(dir.path().to_path_buf());
assert_eq!(count(dir.path()), 0);
subagents.start("toolu_a", "a", None);
std::thread::sleep(std::time::Duration::from_millis(2));
subagents.start("toolu_b", "b", None);
let rows = subagents.list(true);
assert_eq!(
rows.iter().map(|r| r.id.as_str()).collect::<Vec<_>>(),
["toolu_a", "toolu_b"]
);
assert_eq!(count(dir.path()), 2);
}
}