diff --git a/crates/agent-core/ARCHITECTURE.md b/crates/agent-core/ARCHITECTURE.md index 8702b93..015d91d 100644 --- a/crates/agent-core/ARCHITECTURE.md +++ b/crates/agent-core/ARCHITECTURE.md @@ -30,9 +30,16 @@ Everything above the wire is testable with mocks because of two trait boundaries - **`Tool`** — tests register a mock (`EchoTool`) to exercise dispatch without a real capability. - **`ModelTransport`** — tests replay scripted `StreamEvent`s to exercise the loop without a network. +## Observation surface + +`Agent::run` exposes only streamed model events (`FnMut(&StreamEvent)`); `Agent::run_events` exposes +the full [`AgentEvent`] stream — `Stream(StreamEvent)`, `ToolStart`/`ToolEnd` (tool boundaries), and +`TurnEnd`. The headless `serve` in the `beyond-ai-agent` crate serializes these to its clients. + ## Milestone status -Scaffolded: the type model, the `Tool`/`ToolRegistry` seam, the `ModelTransport` seam, `Session`, -and errors — all unit-tested. **Not yet built:** the wire dialect adapters (OpenAI + Anthropic), the -HTTP-to-gateway transport, the agent loop itself, and the `serve` control API. See the repo plan for -the milestone sequence (M2–M8). +Built and tested: the type model, the `Tool`/`ToolRegistry` seam, the `ModelTransport` seam + +`GatewayClient` HTTP transport, `MockTransport`, the OpenAI/Anthropic wire dialects, the agent loop +(`run`/`run_events`), and `Session`. The coding tools, the `run` CLI, and the headless `serve` +control protocol live in the `beyond-ai-agent` crate. Remaining: the Beyond platform tools +(fork/sync/logs) and the full gateway e2e (M8). diff --git a/crates/agent-core/src/agent.rs b/crates/agent-core/src/agent.rs index 90741fa..a476822 100644 --- a/crates/agent-core/src/agent.rs +++ b/crates/agent-core/src/agent.rs @@ -10,6 +10,7 @@ use std::sync::Arc; use futures::StreamExt; +use serde::Serialize; use serde_json::{Value, json}; use crate::error::{Error, Result}; @@ -18,6 +19,31 @@ use crate::session::Session; use crate::tool::ToolRegistry; use crate::transport::{ModelRequest, ModelTransport}; +/// An observable event from a run: a streamed model event, a tool-invocation boundary, or a turn +/// boundary. The headless server serializes these to its clients; [`Agent::run`] exposes only the +/// `Stream` events. +#[derive(Debug, Clone, Serialize)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum AgentEvent { + /// A streamed model event (text/tool deltas, usage, stop). + Stream(StreamEvent), + /// A tool is about to run, with the arguments the model supplied. + ToolStart { + id: String, + name: String, + input: Value, + }, + /// A tool finished (or errored); `result` is what's fed back to the model. + ToolEnd { + id: String, + name: String, + result: String, + is_error: bool, + }, + /// One model turn completed. + TurnEnd { stop_reason: StopReason, step: u32 }, +} + /// Default per-turn output token ceiling. const DEFAULT_MAX_TOKENS: u32 = 4096; /// Default ceiling on loop iterations before bailing — a runaway-tool-call backstop. @@ -77,6 +103,21 @@ impl Agent { pub async fn run(&self, session: &mut Session, mut on_event: F) -> Result<()> where F: FnMut(&StreamEvent), + { + self.run_events(session, move |ev| { + if let AgentEvent::Stream(s) = &ev { + on_event(s); + } + }) + .await + } + + /// Drive the loop to completion, emitting an [`AgentEvent`] for every streamed model event, tool + /// invocation, and turn boundary — the full observation surface the headless server streams to + /// its clients. Returns when the model ends its turn without tools, or [`Error::MaxSteps`]. + pub async fn run_events(&self, session: &mut Session, mut sink: F) -> Result<()> + where + F: FnMut(AgentEvent), { loop { if session.steps >= self.max_steps { @@ -93,10 +134,17 @@ impl Agent { req = req.with_system(system.clone()); } - let turn = self.run_turn(req, &mut on_event).await?; + let turn = { + let mut emit = |ev: StreamEvent| sink(AgentEvent::Stream(ev)); + self.run_turn(req, &mut emit).await? + }; session.push(Message::assistant(turn.blocks)); session.record_usage(turn.input_tokens, turn.output_tokens); session.steps += 1; + sink(AgentEvent::TurnEnd { + stop_reason: turn.stop_reason, + step: session.steps, + }); // Collect the tool calls the assistant just made. let calls: Vec<(String, String, Value)> = session @@ -116,6 +164,11 @@ impl Agent { // Run each tool and feed results back as a user turn. A tool's own failure becomes an // error `tool_result`, not an aborted run — the model can react to it next turn. for (id, name, input) in calls { + sink(AgentEvent::ToolStart { + id: id.clone(), + name: name.clone(), + input: input.clone(), + }); let (content, is_error) = match self.tools.get(&name) { Some(tool) => match tool.run(input).await { Ok(out) => (out, false), @@ -123,21 +176,24 @@ impl Agent { }, None => (format!("unknown tool: {name}"), true), }; + sink(AgentEvent::ToolEnd { + id: id.clone(), + name: name.clone(), + result: content.clone(), + is_error, + }); session.push(Message::tool_result(id, content, is_error)); } } } /// Stream and assemble a single model turn into content blocks + accounting. - async fn run_turn(&self, req: ModelRequest, on_event: &mut F) -> Result - where - F: FnMut(&StreamEvent), - { + async fn run_turn(&self, req: ModelRequest, emit: &mut dyn FnMut(StreamEvent)) -> Result { let mut stream = self.transport.stream(req).await?; let mut acc = Accumulator::default(); while let Some(ev) = stream.next().await { let ev = ev?; - on_event(&ev); + emit(ev.clone()); acc.apply(ev); } Ok(acc.finish()) diff --git a/crates/agent-core/src/lib.rs b/crates/agent-core/src/lib.rs index ee64d5d..c84b4aa 100644 --- a/crates/agent-core/src/lib.rs +++ b/crates/agent-core/src/lib.rs @@ -28,7 +28,7 @@ pub mod session; pub mod tool; pub mod transport; -pub use agent::Agent; +pub use agent::{Agent, AgentEvent}; pub use client::GatewayClient; pub use error::{Error, Result, ToolError}; pub use message::{ContentBlock, Message, Role, StopReason, StreamEvent, ToolDef}; diff --git a/crates/agent/src/main.rs b/crates/agent/src/main.rs index b11591e..605b4ad 100644 --- a/crates/agent/src/main.rs +++ b/crates/agent/src/main.rs @@ -1,8 +1,9 @@ //! Beyond agent harness — CLI. //! -//! `run` drives a one-shot coding task to completion through the gateway. `serve` (M7) will expose -//! the headless control API. `tools` lists the advertised tool set. Model traffic always flows -//! through the gateway (`AI_GATEWAY_URL`) authenticated with a `bai_v1` key (`AI_AGENT_KEY`). +//! `run` drives a one-shot coding task to completion through the gateway. `serve` exposes the +//! headless control protocol (newline-delimited JSON over stdio). `tools` lists the advertised tool +//! set. Model traffic always flows through the gateway (`AI_GATEWAY_URL`) authenticated with a +//! `bai_v1` key (`AI_AGENT_KEY`). // Unit tests assert preconditions with `.unwrap()`; allow that under `test` (matches the gateway and // agent-core crate roots). Production paths stay panic-free per the workspace lints. @@ -14,6 +15,7 @@ use std::sync::Arc; use agent_core::{Agent, GatewayClient, Session, StreamEvent}; use clap::{Parser, Subcommand}; +mod serve; mod tools; /// Default model when neither `--model` nor `AI_AGENT_MODEL` is set. @@ -51,8 +53,24 @@ enum Command { #[arg(long, default_value_t = 24)] max_steps: u32, }, - /// Run the headless agent server exposing an attachable control API. (M7.) - Serve, + /// Run the headless agent server: a newline-delimited JSON control protocol over stdio. + Serve { + /// Model id (default `claude-opus-4-8`, or `AI_AGENT_MODEL`). + #[arg(long, env = "AI_AGENT_MODEL")] + model: Option, + /// Gateway base URL (default `http://ai.internal`, or `AI_GATEWAY_URL`). + #[arg(long, env = "AI_GATEWAY_URL")] + gateway_url: Option, + /// Virtual key (`bai_v1…`) or BYO provider key. Required; or set `AI_AGENT_KEY`. + #[arg(long, env = "AI_AGENT_KEY")] + key: Option, + /// Persist/restore session state here so a later `serve` reattaches with the transcript. + #[arg(long, env = "AI_AGENT_SESSION_FILE")] + session_file: Option, + /// Max loop iterations per prompt before bailing. + #[arg(long, default_value_t = 24)] + max_steps: u32, + }, /// List the tools the agent advertises to the model. Tools, } @@ -73,8 +91,24 @@ async fn main() -> Result<(), Box> { } => { run_task(task, model, gateway_url, key, max_steps).await?; } - Command::Serve => { - println!("scaffold: `serve` is not wired yet (headless control API = M7)."); + Command::Serve { + model, + gateway_url, + key, + session_file, + max_steps, + } => { + let key = key + .ok_or("no gateway key: pass --key or set AI_AGENT_KEY (a bai_v1… virtual key)")?; + serve::serve(serve::ServeConfig { + gateway: gateway_url.unwrap_or_else(|| DEFAULT_GATEWAY.to_string()), + key, + model: model.unwrap_or_else(|| DEFAULT_MODEL.to_string()), + max_steps, + system: SYSTEM_PROMPT.to_string(), + session_file, + }) + .await?; } Command::Tools => { let reg = tools::default_registry(); diff --git a/crates/agent/src/serve.rs b/crates/agent/src/serve.rs new file mode 100644 index 0000000..14d8aff --- /dev/null +++ b/crates/agent/src/serve.rs @@ -0,0 +1,229 @@ +//! Headless `serve` — a newline-delimited JSON control protocol over stdio. +//! +//! The server is the source of truth; any client (a TUI, an editor, or an `ssh` pipe) drives it by +//! writing one JSON command per line to stdin and reading one JSON frame per line from stdout. The +//! shape mirrors pi's `rpc` mode and opencode's session server: commands get a `response` frame, +//! and a `prompt` streams `event` frames (the agent's `AgentEvent`s) before its response. +//! +//! Session state is `serde`-persisted to `--session-file` after each turn, so a client can detach +//! and a later `serve` over the same file reattaches with the full transcript (`get_messages`). +//! +//! Commands (stdin): `{id?, type, …}` +//! - `{type:"prompt", message}` run a turn; streams `event` frames, then a `response` +//! - `{type:"get_state"}` → `data: {session_id, model, steps, message_count, …}` +//! - `{type:"get_messages"}` → `data: {messages: [...]}` +//! - `{type:"new_session"}` reset transcript → `data: {session_id}` +//! +//! Frames (stdout) are either `{type:"response", id?, command, success, data?, error?}` or +//! `{type:"event", event: }`. + +use std::sync::Arc; +use std::time::{SystemTime, UNIX_EPOCH}; + +use agent_core::{Agent, AgentEvent, GatewayClient, Session}; +use serde_json::{Map, Value, json}; +use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; +use tokio::sync::mpsc; + +use crate::tools; + +/// Options for the headless server (mirrors `run`, plus a session file). +pub struct ServeConfig { + pub gateway: String, + pub key: String, + pub model: String, + pub max_steps: u32, + pub system: String, + pub session_file: Option, +} + +/// Run the control loop until stdin closes. +pub async fn serve(cfg: ServeConfig) -> Result<(), Box> { + let client = GatewayClient::new(cfg.gateway, cfg.key)?; + let agent = Agent::new(Arc::new(client), cfg.model.clone()) + .with_tools(tools::default_registry()) + .with_system(cfg.system) + .with_max_steps(cfg.max_steps); + + let mut session = match &cfg.session_file { + Some(path) => load_session(path), + None => Session::new(), + }; + let mut session_id = make_id(); + + // One writer task owns stdout; every frame (events + responses) is serialized through it in FIFO + // order, so output never interleaves. + let (out_tx, mut out_rx) = mpsc::unbounded_channel::(); + let writer = tokio::spawn(async move { + let mut out = tokio::io::stdout(); + while let Some(frame) = out_rx.recv().await { + if let Ok(line) = serde_json::to_string(&frame) { + let _ = out.write_all(line.as_bytes()).await; + let _ = out.write_all(b"\n").await; + let _ = out.flush().await; + } + } + }); + + // Announce readiness so a client can sync before issuing commands. + let _ = out_tx.send(json!({ "type": "ready", "session_id": session_id, "model": cfg.model })); + + let mut lines = BufReader::new(tokio::io::stdin()).lines(); + while let Some(line) = lines.next_line().await? { + let line = line.trim(); + if line.is_empty() { + continue; + } + let cmd: Value = match serde_json::from_str(line) { + Ok(v) => v, + Err(e) => { + let _ = out_tx.send(response( + None, + "?", + false, + None, + Some(&format!("invalid JSON: {e}")), + )); + continue; + } + }; + let id = cmd.get("id").and_then(Value::as_str).map(str::to_string); + let ctype = cmd + .get("type") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + + match ctype.as_str() { + "prompt" => { + let message = cmd + .get("message") + .and_then(Value::as_str) + .unwrap_or("") + .to_string(); + session.user(message); + let tx = out_tx.clone(); + let result = agent + .run_events(&mut session, move |ev| { + let _ = tx.send(event_frame(ev)); + }) + .await; + + if let Some(path) = &cfg.session_file { + save_session(path, &session); + } + let frame = match result { + Ok(()) => response( + id.clone(), + "prompt", + true, + Some( + json!({ "steps": session.steps, "input_tokens": session.input_tokens, "output_tokens": session.output_tokens }), + ), + None, + ), + Err(e) => response(id.clone(), "prompt", false, None, Some(&e.to_string())), + }; + let _ = out_tx.send(frame); + } + "get_state" => { + let data = json!({ + "session_id": session_id, + "model": cfg.model, + "steps": session.steps, + "message_count": session.messages.len(), + "input_tokens": session.input_tokens, + "output_tokens": session.output_tokens, + }); + let _ = out_tx.send(response(id, "get_state", true, Some(data), None)); + } + "get_messages" => { + let messages = serde_json::to_value(&session.messages).unwrap_or(Value::Null); + let _ = out_tx.send(response( + id, + "get_messages", + true, + Some(json!({ "messages": messages })), + None, + )); + } + "new_session" => { + session = Session::new(); + session_id = make_id(); + if let Some(path) = &cfg.session_file { + save_session(path, &session); + } + let _ = out_tx.send(response( + id, + "new_session", + true, + Some(json!({ "session_id": session_id })), + None, + )); + } + other => { + let _ = out_tx.send(response(id, other, false, None, Some("unknown command"))); + } + } + } + + drop(out_tx); + let _ = writer.await; + Ok(()) +} + +/// Build a `response` frame. +fn response( + id: Option, + command: &str, + success: bool, + data: Option, + error: Option<&str>, +) -> Value { + let mut m = Map::new(); + m.insert("type".into(), json!("response")); + if let Some(id) = id { + m.insert("id".into(), json!(id)); + } + m.insert("command".into(), json!(command)); + m.insert("success".into(), json!(success)); + if let Some(d) = data { + m.insert("data".into(), d); + } + if let Some(e) = error { + m.insert("error".into(), json!(e)); + } + Value::Object(m) +} + +/// Wrap an `AgentEvent` in an `event` frame. +fn event_frame(ev: AgentEvent) -> Value { + let mut m = Map::new(); + m.insert("type".into(), json!("event")); + if let Ok(v) = serde_json::to_value(&ev) { + m.insert("event".into(), v); + } + Value::Object(m) +} + +fn load_session(path: &str) -> Session { + std::fs::read_to_string(path) + .ok() + .and_then(|s| serde_json::from_str(&s).ok()) + .unwrap_or_default() +} + +fn save_session(path: &str, session: &Session) { + if let Ok(s) = serde_json::to_string(session) { + let _ = std::fs::write(path, s); + } +} + +/// A short, monotonic-ish session id (no extra deps; uniqueness across a process is sufficient). +fn make_id() -> String { + let nanos = SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|d| d.as_nanos()) + .unwrap_or(0); + format!("sess_{nanos:x}") +} diff --git a/crates/agent/tests/common/mod.rs b/crates/agent/tests/common/mod.rs new file mode 100644 index 0000000..aaf7fc4 --- /dev/null +++ b/crates/agent/tests/common/mod.rs @@ -0,0 +1,91 @@ +//! Shared test helpers: a mock model server speaking Anthropic SSE. +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic, dead_code)] + +use std::io::{Read, Write}; +use std::net::{TcpListener, TcpStream}; +use std::sync::{Arc, Mutex}; +use std::thread; + +use serde_json::{Value, json}; + +/// An Anthropic SSE turn that calls one tool with the given JSON-argument string. +pub fn turn_tool_use(id: &str, name: &str, args_json: &str) -> String { + sse(&[ + json!({ "type": "message_start", "message": { "usage": { "input_tokens": 10, "output_tokens": 1 } } }), + json!({ "type": "content_block_start", "index": 0, "content_block": { "type": "tool_use", "id": id, "name": name, "input": {} } }), + json!({ "type": "content_block_delta", "index": 0, "delta": { "type": "input_json_delta", "partial_json": args_json } }), + json!({ "type": "content_block_stop", "index": 0 }), + json!({ "type": "message_delta", "delta": { "stop_reason": "tool_use" }, "usage": { "output_tokens": 8 } }), + json!({ "type": "message_stop" }), + ]) +} + +/// An Anthropic SSE turn that emits text and ends. +pub fn turn_text(text: &str) -> String { + sse(&[ + json!({ "type": "message_start", "message": { "usage": { "input_tokens": 12, "output_tokens": 1 } } }), + json!({ "type": "content_block_start", "index": 0, "content_block": { "type": "text", "text": "" } }), + json!({ "type": "content_block_delta", "index": 0, "delta": { "type": "text_delta", "text": text } }), + json!({ "type": "content_block_stop", "index": 0 }), + json!({ "type": "message_delta", "delta": { "stop_reason": "end_turn" }, "usage": { "output_tokens": 6 } }), + json!({ "type": "message_stop" }), + ]) +} + +fn sse(events: &[Value]) -> String { + events.iter().map(|e| format!("data: {e}\n\n")).collect() +} + +fn read_http_request(stream: &mut TcpStream) -> String { + let mut buf = Vec::new(); + let mut tmp = [0u8; 2048]; + loop { + if let Some(pos) = buf.windows(4).position(|w| w == b"\r\n\r\n") { + let headers = String::from_utf8_lossy(&buf[..pos]).to_ascii_lowercase(); + let len = headers + .lines() + .find_map(|l| { + l.strip_prefix("content-length:") + .map(|v| v.trim().parse::().unwrap_or(0)) + }) + .unwrap_or(0); + let mut body = buf[pos + 4..].to_vec(); + while body.len() < len { + let n = stream.read(&mut tmp).unwrap_or(0); + if n == 0 { + break; + } + body.extend_from_slice(&tmp[..n]); + } + return String::from_utf8_lossy(&body).into_owned(); + } + let n = stream.read(&mut tmp).unwrap_or(0); + if n == 0 { + return String::from_utf8_lossy(&buf).into_owned(); + } + buf.extend_from_slice(&tmp[..n]); + } +} + +/// Spawn a model server answering `responses` in order, recording each request body. Returns the +/// base URL and the shared record of request bodies. +pub fn spawn_model_server(responses: Vec) -> (String, Arc>>) { + let listener = TcpListener::bind("127.0.0.1:0").unwrap(); + let addr = listener.local_addr().unwrap(); + let bodies = Arc::new(Mutex::new(Vec::new())); + let recorder = bodies.clone(); + thread::spawn(move || { + for resp in responses { + if let Ok((mut stream, _)) = listener.accept() { + let body = read_http_request(&mut stream); + recorder.lock().unwrap().push(body); + let http = format!( + "HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\nConnection: close\r\n\r\n{resp}" + ); + let _ = stream.write_all(http.as_bytes()); + let _ = stream.flush(); + } + } + }); + (format!("http://{addr}"), bodies) +} diff --git a/crates/agent/tests/serve_e2e.rs b/crates/agent/tests/serve_e2e.rs new file mode 100644 index 0000000..7c91801 --- /dev/null +++ b/crates/agent/tests/serve_e2e.rs @@ -0,0 +1,143 @@ +//! End-to-end: the real `beyond-ai-agent serve` binary over its stdio control protocol. +//! +//! Drives the headless server exactly as a remote client (or an SSH pipe) would: writes JSON command +//! lines to stdin, reads JSON frames from stdout. Proves (a) a `prompt` streams `event` frames for a +//! tool round-trip then a success `response`, (b) `get_messages` returns the transcript, and (c) a +//! fresh `serve` process **reattaches** to the persisted session and sees the prior transcript. +#![allow(clippy::unwrap_used, clippy::expect_used, clippy::panic)] + +mod common; + +use std::io::{BufRead, BufReader, Write}; +use std::process::{Command, Stdio}; + +use common::{spawn_model_server, turn_text, turn_tool_use}; +use serde_json::{Value, json}; + +/// Read stdout frames until the `response` frame for `command` arrives; return all frames seen. +fn read_until_response(reader: &mut impl BufRead, command: &str) -> Vec { + let mut frames = Vec::new(); + let mut line = String::new(); + loop { + line.clear(); + if reader.read_line(&mut line).unwrap() == 0 { + break; + } + let Ok(v) = serde_json::from_str::(line.trim()) else { + continue; + }; + let done = v.get("type").and_then(Value::as_str) == Some("response") + && v.get("command").and_then(Value::as_str) == Some(command); + frames.push(v); + if done { + break; + } + } + frames +} + +fn serve_cmd(bin: &str, base: &str, session_file: &str) -> Command { + let mut c = Command::new(bin); + c.args([ + "serve", + "--gateway-url", + base, + "--key", + "bai_v1.test", + "--model", + "claude-test", + "--session-file", + session_file, + ]) + .stdin(Stdio::piped()) + .stdout(Stdio::piped()) + .stderr(Stdio::null()); + c +} + +#[test] +fn serve_streams_events_and_reattaches() { + let dir = tempfile::tempdir().unwrap(); + std::fs::write(dir.path().join("hello.txt"), "secret-marker-77\n").unwrap(); + let abs = dir.path().join("hello.txt").to_string_lossy().into_owned(); + let session_file = dir + .path() + .join("session.json") + .to_string_lossy() + .into_owned(); + + // One prompt drives two model turns: read tool, then text. + let turn1 = turn_tool_use("toolu_1", "read", &json!({ "path": abs }).to_string()); + let turn2 = turn_text("Read complete."); + let (base, _bodies) = spawn_model_server(vec![turn1, turn2]); + + let bin = env!("CARGO_BIN_EXE_beyond-ai-agent"); + + // --- First session: prompt, observe streamed events, read transcript --- + let mut child = serve_cmd(bin, &base, &session_file).spawn().unwrap(); + let mut stdin = child.stdin.take().unwrap(); + let mut stdout = BufReader::new(child.stdout.take().unwrap()); + + writeln!( + stdin, + "{}", + json!({ "type": "prompt", "message": "read hello.txt" }) + ) + .unwrap(); + stdin.flush().unwrap(); + let frames = read_until_response(&mut stdout, "prompt"); + + // A `ready` frame was emitted on startup. + assert!( + frames + .iter() + .any(|f| f.get("type").and_then(Value::as_str) == Some("ready")) + ); + // Tool-call boundaries streamed as events. + let events: Vec<&Value> = frames.iter().filter(|f| f["type"] == "event").collect(); + assert!( + events + .iter() + .any(|e| e["event"]["kind"] == "tool_start" && e["event"]["name"] == "read"), + "expected a tool_start event for `read`; frames: {frames:#?}" + ); + assert!( + events + .iter() + .any(|e| e["event"]["kind"] == "tool_end" && e["event"]["name"] == "read") + ); + // Final response is a success. + let resp = frames.last().unwrap(); + assert_eq!(resp["type"], "response"); + assert_eq!(resp["success"], true); + + // get_messages returns the transcript including the tool result and final text. + writeln!(stdin, "{}", json!({ "type": "get_messages" })).unwrap(); + stdin.flush().unwrap(); + let frames2 = read_until_response(&mut stdout, "get_messages"); + let dump = frames2.last().unwrap()["data"]["messages"].to_string(); + assert!( + dump.contains("secret-marker-77"), + "transcript should hold the tool result: {dump}" + ); + assert!(dump.contains("Read complete.")); + + drop(stdin); // close stdin → server exits + assert!(child.wait().unwrap().success()); + + // --- Reattach: a fresh process over the same session file sees the prior transcript --- + let mut child2 = serve_cmd(bin, &base, &session_file).spawn().unwrap(); + let mut stdin2 = child2.stdin.take().unwrap(); + let mut stdout2 = BufReader::new(child2.stdout.take().unwrap()); + writeln!(stdin2, "{}", json!({ "type": "get_messages" })).unwrap(); + stdin2.flush().unwrap(); + let frames3 = read_until_response(&mut stdout2, "get_messages"); + let dump3 = frames3.last().unwrap()["data"]["messages"].to_string(); + assert!( + dump3.contains("secret-marker-77"), + "reattached session must restore the transcript: {dump3}" + ); + + drop(stdin2); + child2.wait().unwrap(); +}