From df1149aad3a53273e81ae6324ea36cbe0556bc99 Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:08:12 -0600 Subject: [PATCH 01/21] feat(host): bounded rendezvous for goose delegate children A goose delegate child sends inference under its own session, and nothing orders its first request against the ACP update that names it. A goose generation now holds the children its parent's live delegate calls registered, frozen to it, with caps on children, calls and parked requests. Inference naming an unknown opaque session waits once, holding no upstream connection, until its child is registered (then admission runs again in full), the generation is sealed, revoked or closed, or a 5 s deadline refuses it with 504 child_unregistered. A full queue gets 429 child_pending_full. Expiry never admits and nothing falls back to the parent's identity. Membership checks, parking and registration share the registry lock, so a wake cannot be lost, and each parked request frees only its own slot however it ends. `plugin_host_forward_register_child` and `registerForwardChild` add the registration, with five refusal reasons; contract violations reject. `admitted` events say whether a child sent the request. `AcpSession` gains an optional `observe` hook, and `forwardChildren` registers only a completed `delegate` call announced earlier in the same live prompt, surfacing every refusal and rejection. Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Philip Azar --- src-tauri/build.rs | 1 + src-tauri/capabilities/default.json | 1 + src-tauri/src/browser_permissions_tests.rs | 1 + src-tauri/src/host_forward.rs | 238 ++++++++++++- src-tauri/src/host_forward/tests.rs | 332 +++++++++++++++++- src-tauri/src/lib.rs | 4 +- src/bundled/goose/forward-children.test.ts | 95 +++++ src/bundled/goose/forward-children.ts | 48 +++ src/features/agent-harness/acp/client.test.ts | 24 ++ src/features/agent-harness/acp/session.ts | 6 + src/features/host/forward.test.ts | 33 ++ src/features/host/forward.ts | 38 +- 12 files changed, 804 insertions(+), 17 deletions(-) create mode 100644 src/bundled/goose/forward-children.test.ts create mode 100644 src/bundled/goose/forward-children.ts diff --git a/src-tauri/build.rs b/src-tauri/build.rs index 4b23bf34ed..23fef61852 100644 --- a/src-tauri/build.rs +++ b/src-tauri/build.rs @@ -179,6 +179,7 @@ fn main() { "plugin_host_forward_admission", "plugin_host_forward_open", "plugin_host_forward_associate", + "plugin_host_forward_register_child", "plugin_host_forward_seal", "plugin_host_forward_close", "plugin_host_read_file", diff --git a/src-tauri/capabilities/default.json b/src-tauri/capabilities/default.json index 046f8ef44b..5101184f27 100644 --- a/src-tauri/capabilities/default.json +++ b/src-tauri/capabilities/default.json @@ -125,6 +125,7 @@ "allow-plugin-host-forward-admission", "allow-plugin-host-forward-open", "allow-plugin-host-forward-associate", + "allow-plugin-host-forward-register-child", "allow-plugin-host-forward-seal", "allow-plugin-host-forward-close", "allow-plugin-host-read-file", diff --git a/src-tauri/src/browser_permissions_tests.rs b/src-tauri/src/browser_permissions_tests.rs index 0204d6eea2..3f6cd7cdcb 100644 --- a/src-tauri/src/browser_permissions_tests.rs +++ b/src-tauri/src/browser_permissions_tests.rs @@ -166,6 +166,7 @@ fn native_command_permissions_isolate_webviews() { "plugin_host_forward_admission", "plugin_host_forward_open", "plugin_host_forward_associate", + "plugin_host_forward_register_child", "plugin_host_forward_seal", "plugin_host_forward_close", "plugin_host_read_file", diff --git a/src-tauri/src/host_forward.rs b/src-tauri/src/host_forward.rs index 0a4f21aa5a..d70c0052df 100644 --- a/src-tauri/src/host_forward.rs +++ b/src-tauri/src/host_forward.rs @@ -35,7 +35,7 @@ use hyper::header::{self, HeaderMap, HeaderName, HeaderValue}; use hyper::{server::conn::http1, service::service_fn, Method, Request, Response, StatusCode}; use hyper_util::rt::{TokioIo, TokioTimer}; use serde::{Deserialize, Serialize}; -use std::collections::{BTreeMap, HashMap}; +use std::collections::{BTreeMap, HashMap, HashSet}; use std::convert::Infallible; use std::future::Future; use std::net::Ipv4Addr; @@ -45,7 +45,7 @@ use std::task::{Context, Poll}; use std::time::Duration; use tauri::ipc::Channel; use tokio::net::TcpListener; -use tokio::sync::{mpsc, Semaphore}; +use tokio::sync::{mpsc, oneshot, Semaphore}; use tokio_util::sync::{CancellationToken, WaitForCancellationFutureOwned}; /// Generations open at once across all plugins. @@ -59,6 +59,15 @@ const BODY_LIMIT: usize = 64 * 1024 * 1024; const CONNECT_TIMEOUT: Duration = Duration::from_secs(30); /// The whole reserved header block, counted as `name: value\r\n` lines. const HEADER_BUDGET: usize = 2048; +/// A goose generation's child rendezvous bounds: inference requests parked +/// for a child not yet registered, registered children, and the delegate +/// calls that named them (a resume adds a call, not a child). +const PENDING_CAP: usize = 5; +const CHILD_CAP: usize = 32; +const CALL_CAP: usize = 64; +/// How long a parked request waits for its child's registration, by +/// default. It only bounds resources: expiry rejects, never admits. +const RENDEZVOUS_DEADLINE: Duration = Duration::from_secs(5); /// Never forwarded: the connection's own, the address and anything a browser /// adds, plus whatever `Connection` names. Credentials and every other header /// pass through unchanged. @@ -217,8 +226,13 @@ fn origin(origin: &str) -> Option { #[derive(Clone, Debug, Serialize, PartialEq, Eq)] #[serde(tag = "type", rename_all = "camelCase")] pub(crate) enum ForwardEvent { - /// A request passed admission and was sent upstream. - Admitted { request: u64, class: Class }, + /// A request passed admission and was sent upstream. `child`: it named a + /// registered goose child, not the generation's own session. + Admitted { + request: u64, + class: Class, + child: bool, + }, /// An admitted request's upstream response ended: relayed whole, or /// aborted by the harness hanging up, a revocation or a close. `status` is /// what the harness was answered, if anything yet. @@ -267,6 +281,18 @@ pub(crate) enum Pushed { Stale, } +/// A goose child registration's outcome. Contract violations are errors. +#[derive(Clone, Copy, Debug, Serialize, PartialEq, Eq)] +#[serde(tag = "type", rename_all = "camelCase")] +pub(crate) enum Registered { + Registered, + /// `child_call_conflict`, `child_cap`, `call_cap`, `generation_ended` + /// (sealed, closed or unknown) or `generation_revoked`. + Refused { + reason: &'static str, + }, +} + #[derive(Clone, Copy, Debug, PartialEq, Eq)] enum State { Open, @@ -290,6 +316,30 @@ struct Generation { abort: CancellationToken, events: Arc, requests: u64, + /// goose children registered for this generation's parent session, + /// frozen to it; each delegate call names one child. + children: HashSet, + calls: HashMap, + /// Inference requests parked for a child, by their own waiter ID. + pending: HashMap, + next_waiter: u64, +} +impl Generation { + /// Answers every parked request; none can outlive admission. + fn release(&mut self, status: StatusCode, reason: &'static str) { + for (_, waiter) in self.pending.drain() { + let _ = waiter.wake.send(Wake::Released(status, reason)); + } + } +} + +struct Waiter { + session: String, + wake: oneshot::Sender, +} +enum Wake { + Registered, + Released(StatusCode, &'static str), } #[derive(Clone, Copy, Default)] @@ -319,6 +369,7 @@ impl Registry { if let Some(generation) = self.generations.get_mut(handle) { generation.state = State::Revoked; generation.abort.cancel(); + generation.release(StatusCode::GONE, "generation_revoked"); } } } @@ -334,6 +385,7 @@ pub(crate) struct HostForward { inner: Arc>, /// The shared listener's address, bound on first use for the app's life. listener: Arc>>, + rendezvous: Duration, } impl Default for HostForward { fn default() -> Self { @@ -346,6 +398,7 @@ impl HostForward { gateway: gateway.map(Arc::new), inner: Arc::default(), listener: Arc::default(), + rendezvous: RENDEZVOUS_DEADLINE, } } fn registry(&self) -> MutexGuard<'_, Registry> { @@ -479,6 +532,10 @@ impl HostForward { abort: CancellationToken::new(), events: Arc::new(Events(events)), requests: 0, + children: HashSet::new(), + calls: HashMap::new(), + pending: HashMap::new(), + next_waiter: 0, }, ); Ok(Opened::Opened { @@ -511,10 +568,71 @@ impl HostForward { let generation = registry.generations.get_mut(&handle).expect("found"); if generation.state == State::Open { generation.state = State::Sealed; + generation.release(StatusCode::GONE, "generation_ended"); } Ok(()) } + /// Registers the goose child `child` that delegate `call` named, waking + /// every request parked for it in the same critical section. + fn register_child( + &self, + plugin: &str, + id: u64, + call: String, + child: String, + ) -> Result { + let refused = |reason| Ok(Registered::Refused { reason }); + let mut registry = self.registry(); + let Ok(handle) = registry.handle_of(plugin, id) else { + return refused("generation_ended"); + }; + let generation = ®istry.generations[&handle]; + let current = registry + .routes + .get(&(generation.plugin.clone(), generation.route.clone())) + .copied() + .unwrap_or_default(); + let generation = registry.generations.get_mut(&handle).expect("found"); + if generation.harness != Harness::Goose { + return Err("Only a goose generation has children".into()); + } + let Some(parent) = generation.session.as_deref() else { + return Err("The generation's session is not associated".into()); + }; + if !opaque(&call) || !opaque(&child) || child == parent { + return Err("Invalid child session".into()); + } + match generation.state { + State::Open if current.admits && current.revision == generation.revision => {} + State::Open | State::Revoked => return refused("generation_revoked"), + State::Sealed => return refused("generation_ended"), + } + match generation.calls.get(&call) { + Some(named) if *named == child => return Ok(Registered::Registered), + Some(_) => return refused("child_call_conflict"), + None if generation.calls.len() >= CALL_CAP => return refused("call_cap"), + None => {} + } + if !generation.children.contains(&child) && generation.children.len() >= CHILD_CAP { + return refused("child_cap"); + } + generation.calls.insert(call, child.clone()); + generation.children.insert(child.clone()); + let woken: Vec = generation + .pending + .iter() + .filter(|(_, waiter)| waiter.session == child) + .map(|(waiter, _)| *waiter) + .collect(); + for waiter in woken { + if let Some(waiter) = generation.pending.remove(&waiter) { + let _ = waiter.wake.send(Wake::Registered); + } + } + Ok(Registered::Registered) + } + fn close(&self, plugin: &str, id: u64) -> Result<(), String> { let mut registry = self.registry(); let handle = registry.handle_of(plugin, id)?; @@ -553,8 +671,32 @@ impl HostForward { Ok(authority) } - /// Admits one request under the lock that revocation takes, or refuses it. - fn admit(&self, request: &Request) -> Result { + /// Admits one request, or refuses it. goose inference naming a child not + /// yet registered waits, once and bounded, then is admitted afresh. + async fn admission(&self, request: &Request) -> Result { + let mut parked = match self.admit(request, true)? { + Step::Admit(admitted) => return Ok(admitted), + Step::Park(parked) => parked, + }; + let refuse = |status, reason| Refusal { + status, + reason, + events: Some(parked.events.clone()), + }; + match tokio::time::timeout(self.rendezvous, &mut parked.wake).await { + Ok(Ok(Wake::Registered)) => match self.admit(request, false)? { + Step::Admit(admitted) => Ok(admitted), + Step::Park(_) => unreachable!("a re-check never parks"), + }, + Ok(Ok(Wake::Released(status, reason))) => Err(refuse(status, reason)), + Ok(Err(_)) => Err(refuse(StatusCode::GONE, "generation_ended")), + Err(_) => Err(refuse(StatusCode::GATEWAY_TIMEOUT, "child_unregistered")), + } + } + + /// Admits one request under the lock that revocation and registration + /// take, parks it for a goose child when `wait`, or refuses it. + fn admit(&self, request: &Request, wait: bool) -> Result { let authority = self .listener .lock() @@ -618,7 +760,15 @@ impl HostForward { (Class::Discovery, None, _) | (Class::Discovery, Some(_), None) => true, (_, Some(session), expected) => expected == Some(session), }; - if !matches { + // goose inference under another session may be a delegated child's. + let child = session.filter(|session| { + !matches + && class == Class::Inference + && generation.harness == Harness::Goose + && expected.is_some() + && opaque(session) + }); + if !matches && child.is_none() { let reason = if expected.is_none() { "session_unassociated" } else { @@ -629,26 +779,75 @@ impl HostForward { if !current.admits || current.revision != generation.revision { return Err(refuse(StatusCode::GONE, "generation_revoked")); } + if let Some(child) = child.filter(|child| !generation.children.contains(*child)) { + if !wait { + return Err(refuse(StatusCode::BAD_REQUEST, "session_mismatch")); + } + if generation.pending.len() >= PENDING_CAP { + return Err(refuse(StatusCode::TOO_MANY_REQUESTS, "child_pending_full")); + } + let (sender, wake) = oneshot::channel(); + let waiter = generation.next_waiter; + generation.next_waiter += 1; + generation.pending.insert( + waiter, + Waiter { + session: child.to_owned(), + wake: sender, + }, + ); + return Ok(Step::Park(Parked { + forward: self.clone(), + handle: handle.to_owned(), + waiter, + wake, + events: generation.events.clone(), + })); + } generation.requests += 1; let headers = match class { Class::Inference => generation.inference.clone(), Class::Discovery => generation.discovery.clone(), }; - Ok(Admitted { + Ok(Step::Admit(Admitted { request: generation.requests, class, + child: child.is_some(), harness: generation.harness, rest, headers, abort: generation.abort.clone(), events: generation.events.clone(), - }) + })) + } +} + +enum Step { + Admit(Admitted), + Park(Parked), +} + +/// A request parked for a goose child. However it ends, including hyper +/// dropping the request, it frees its own slot and no other. +struct Parked { + forward: HostForward, + handle: String, + waiter: u64, + wake: oneshot::Receiver, + events: Arc, +} +impl Drop for Parked { + fn drop(&mut self) { + if let Some(generation) = self.forward.registry().generations.get_mut(&self.handle) { + generation.pending.remove(&self.waiter); + } } } struct Admitted { request: u64, class: Class, + child: bool, harness: Harness, rest: String, headers: HeaderMap, @@ -738,7 +937,7 @@ fn error(status: StatusCode, reason: &str) -> Reply { } async fn forward(state: &HostForward, request: Request) -> Reply { - let admitted = match state.admit(&request) { + let admitted = match state.admission(&request).await { Ok(admitted) => admitted, Err(refusal) => { if let Some(events) = refusal.events { @@ -753,13 +952,18 @@ async fn forward(state: &HostForward, request: Request) -> Reply { let Admitted { request: id, class, + child, harness, rest, headers: bundle, abort, events, } = admitted; - events.send(ForwardEvent::Admitted { request: id, class }); + events.send(ForwardEvent::Admitted { + request: id, + class, + child, + }); let mut finished = Finished { events, request: id, @@ -1004,6 +1208,18 @@ pub(crate) fn plugin_host_forward_associate( forward.associate(&id, generation, session) } +/// Registers a goose child that a live delegate call named. +#[tauri::command] +pub(crate) fn plugin_host_forward_register_child( + forward: tauri::State<'_, HostForward>, + id: String, + generation: u64, + call: String, + child: String, +) -> Result { + forward.register_child(&id, generation, call, child) +} + /// Ends the generation's admission for good; admitted streams finish. #[tauri::command] pub(crate) fn plugin_host_forward_seal( diff --git a/src-tauri/src/host_forward/tests.rs b/src-tauri/src/host_forward/tests.rs index fff44610b2..63406c6d7f 100644 --- a/src-tauri/src/host_forward/tests.rs +++ b/src-tauri/src/host_forward/tests.rs @@ -581,7 +581,7 @@ async fn forwards_inference_with_only_the_bundle_as_provenance() { ); assert_eq!( recv(&mut generation.events).await, - serde_json::json!({ "type": "admitted", "request": 1, "class": "inference" }) + serde_json::json!({ "type": "admitted", "request": 1, "class": "inference", "child": false }) ); assert_eq!( recv(&mut generation.events).await, @@ -766,7 +766,8 @@ async fn only_allowlisted_paths_reach_the_gateway() { #[tokio::test] async fn inference_must_name_the_generation_session() { let mut gateway = fake_gateway().await; - let forward = ready(gateway.origin); + let mut forward = ready(gateway.origin); + forward.rendezvous = Duration::from_millis(100); let mut claude = open(&forward, Harness::Claude, 1); let client = reqwest::Client::new(); let post = || { @@ -841,9 +842,11 @@ async fn inference_must_name_the_generation_session() { assert!(forward .associate(PLUGIN, claude.id, "20261104_1".into()) .is_err()); + // Another session may be a child not yet registered: it waits, then is + // refused, and nothing reaches the gateway. assert_eq!( invoke("20261104_2").send().await.unwrap().status(), - StatusCode::BAD_REQUEST + StatusCode::GATEWAY_TIMEOUT ); assert_eq!( invoke("20261104_1").send().await.unwrap().status(), @@ -1357,8 +1360,9 @@ fn ipc_shapes_match_the_typescript_bindings() { ForwardEvent::Admitted { request: 1, class: Class::Discovery, + child: false, }, - serde_json::json!({ "type": "admitted", "request": 1, "class": "discovery" }), + serde_json::json!({ "type": "admitted", "request": 1, "class": "discovery", "child": false }), ), ( ForwardEvent::Finished { @@ -1382,6 +1386,16 @@ fn ipc_shapes_match_the_typescript_bindings() { ] { assert_eq!(json(serde_json::to_value(event)), expected); } + assert_eq!( + json(serde_json::to_value(Registered::Registered)), + serde_json::json!({ "type": "registered" }) + ); + assert_eq!( + json(serde_json::to_value(Registered::Refused { + reason: "child_cap" + })), + serde_json::json!({ "type": "refused", "reason": "child_cap" }) + ); assert_eq!( serde_json::from_value::(serde_json::json!("claude")).unwrap(), Harness::Claude @@ -1391,3 +1405,313 @@ fn ipc_shapes_match_the_typescript_bindings() { Harness::Goose ); } + +const PARENT: &str = "20261104_1"; +const CHILD: &str = "20261104_2"; + +/// A goose generation associated with `PARENT`. +fn goose(forward: &HostForward) -> Opening { + let generation = open(forward, Harness::Goose, 1); + forward + .associate(PLUGIN, generation.id, PARENT.into()) + .unwrap(); + generation +} +fn invoke(url: &str, session: &str) -> reqwest::RequestBuilder { + reqwest::Client::new() + .post(format!("{url}/serving-endpoints/model-a/invocations")) + .header("agent-session-id", session) + .body("{}") +} +fn register(forward: &HostForward, id: u64, call: &str, child: &str) -> Registered { + forward + .register_child(PLUGIN, id, call.into(), child.into()) + .unwrap() +} +fn refused(reason: &'static str) -> Registered { + Registered::Refused { reason } +} +fn pending(forward: &HostForward) -> usize { + let registry = forward.registry(); + registry.generations.values().map(|g| g.pending.len()).sum() +} +/// Waits until `count` requests are parked: an observed state, not a delay. +async fn parked(forward: &HostForward, count: usize) { + tokio::time::timeout(Duration::from_secs(5), async { + while pending(forward) != count { + tokio::time::sleep(Duration::from_millis(1)).await; + } + }) + .await + .expect("parked in time"); +} +fn send(request: reqwest::RequestBuilder) -> tokio::task::JoinHandle { + tokio::spawn(async move { request.send().await.unwrap() }) +} + +#[tokio::test] +async fn a_registered_child_is_admitted_under_its_own_session() { + let mut gateway = fake_gateway().await; + let forward = ready(gateway.origin); + let mut generation = goose(&forward); + assert_eq!( + register(&forward, generation.id, "call-1", CHILD), + Registered::Registered + ); + let response = invoke(&generation.url, CHILD).send().await.unwrap(); + assert_eq!(response.status(), StatusCode::OK); + let seen = recv(&mut gateway.seen).await; + assert_eq!(seen.headers["agent-session-id"], CHILD); + assert_eq!(seen.headers["x-gw-ctx-message-id"], MESSAGE); + assert_eq!( + recv(&mut generation.events).await, + serde_json::json!({ "type": "admitted", "request": 1, "class": "inference", "child": true }) + ); + // The parent's own traffic is not a child's. + invoke(&generation.url, PARENT).send().await.unwrap(); + recv(&mut generation.events).await; + assert_eq!(recv(&mut generation.events).await["child"], false); +} + +#[tokio::test] +async fn a_request_first_child_waits_dispatching_nothing_until_registered() { + let mut gateway = fake_gateway().await; + let forward = ready(gateway.origin); + let generation = goose(&forward); + let early = send(invoke(&generation.url, CHILD)); + parked(&forward, 1).await; + assert!(gateway.seen.try_recv().is_err()); + assert_eq!( + register(&forward, generation.id, "call-1", CHILD), + Registered::Registered + ); + assert_eq!(early.await.unwrap().status(), StatusCode::OK); + assert_eq!( + recv(&mut gateway.seen).await.headers["agent-session-id"], + CHILD + ); + assert!(gateway.seen.try_recv().is_err()); + assert_eq!(pending(&forward), 0); +} + +#[tokio::test] +async fn an_unregistered_child_is_refused_at_the_deadline() { + let mut gateway = fake_gateway().await; + let mut forward = ready(gateway.origin); + forward.rendezvous = Duration::from_millis(300); + let mut generation = goose(&forward); + let started = std::time::Instant::now(); + let early = send(invoke(&generation.url, CHILD)); + parked(&forward, 1).await; + let response = early.await.unwrap(); + assert!(started.elapsed() >= forward.rendezvous); + assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT); + assert!(response + .text() + .await + .unwrap() + .contains("gateway_child_unregistered")); + assert_eq!( + recv(&mut generation.events).await, + serde_json::json!({ "type": "refused", "status": 504, "reason": "child_unregistered" }) + ); + assert!(gateway.seen.try_recv().is_err()); + assert_eq!(pending(&forward), 0); + // A registration after expiry still admits a retry; expiry never did. + register(&forward, generation.id, "call-1", CHILD); + assert_eq!( + invoke(&generation.url, CHILD) + .send() + .await + .unwrap() + .status(), + StatusCode::OK + ); +} + +#[tokio::test] +async fn parked_requests_are_capped_and_free_only_their_own_slot() { + let gateway = fake_gateway().await; + let forward = ready(gateway.origin); + let mut generation = goose(&forward); + let waiting: Vec<_> = (0..PENDING_CAP) + .map(|n| send(invoke(&generation.url, &format!("child-{n}")))) + .collect(); + parked(&forward, PENDING_CAP).await; + let full = invoke(&generation.url, "child-x").send().await.unwrap(); + assert_eq!(full.status(), StatusCode::TOO_MANY_REQUESTS); + assert_eq!( + recv(&mut generation.events).await, + serde_json::json!({ "type": "refused", "status": 429, "reason": "child_pending_full" }) + ); + // A second request for child-0 cannot park; drop one harness request and + // only its slot is freed. + let mut waiting = waiting.into_iter(); + waiting.next().unwrap().abort(); + parked(&forward, PENDING_CAP - 1).await; + let again = send(invoke(&generation.url, "child-1")); + parked(&forward, PENDING_CAP).await; + // Both requests for child-1 are woken by one registration. + register(&forward, generation.id, "call-1", "child-1"); + assert_eq!(again.await.unwrap().status(), StatusCode::OK); + assert_eq!( + waiting.next().unwrap().await.unwrap().status(), + StatusCode::OK + ); + assert_eq!(pending(&forward), PENDING_CAP - 2); +} + +#[tokio::test] +async fn seal_revocation_and_close_release_every_parked_request() { + let gateway = fake_gateway().await; + let forward = ready(gateway.origin); + let generation = goose(&forward); + let early = send(invoke(&generation.url, CHILD)); + parked(&forward, 1).await; + forward.seal(PLUGIN, generation.id).unwrap(); + assert_eq!(early.await.unwrap().status(), StatusCode::GONE); + assert_eq!( + register(&forward, generation.id, "call-1", CHILD), + refused("generation_ended") + ); + assert_eq!( + invoke(&generation.url, CHILD) + .send() + .await + .unwrap() + .status(), + StatusCode::GONE + ); + + let revoked = goose(&forward); + let early = send(invoke(&revoked.url, CHILD)); + parked(&forward, 1).await; + forward + .push(forward.page(), PLUGIN, "route-a", 2, true) + .unwrap(); + let response = early.await.unwrap(); + assert_eq!(response.status(), StatusCode::GONE); + assert!(response + .text() + .await + .unwrap() + .contains("gateway_generation_revoked")); + assert_eq!( + register(&forward, revoked.id, "call-1", CHILD), + refused("generation_revoked") + ); + + let closed = open(&forward, Harness::Goose, 2); + forward.associate(PLUGIN, closed.id, PARENT.into()).unwrap(); + let early = send(invoke(&closed.url, CHILD)); + parked(&forward, 1).await; + forward.close(PLUGIN, closed.id).unwrap(); + assert_eq!(early.await.unwrap().status(), StatusCode::GONE); + assert_eq!( + register(&forward, closed.id, "call-1", CHILD), + refused("generation_ended") + ); + assert_eq!(pending(&forward), 0); +} + +#[tokio::test] +async fn registration_is_frozen_to_its_generation_and_parent() { + let gateway = fake_gateway().await; + let forward = ready(gateway.origin); + let claude = open(&forward, Harness::Claude, 1); + let unassociated = open(&forward, Harness::Goose, 1); + let a = goose(&forward); + let b = goose(&forward); + let contract = |id, call: &str, child: &str| { + forward + .register_child(PLUGIN, id, call.into(), child.into()) + .is_err() + }; + assert!(contract(claude.id, "call-1", CHILD)); + assert!(contract(unassociated.id, "call-1", CHILD)); + assert!(contract(a.id, "call-1", PARENT)); + assert!(contract(a.id, "call/1", CHILD)); + assert!(contract(a.id, "call-1", "")); + assert_eq!( + forward + .register_child("other.plugin", a.id, "call-1".into(), CHILD.into()) + .unwrap(), + refused("generation_ended") + ); + assert_eq!( + register(&forward, a.id, "call-1", CHILD), + Registered::Registered + ); + assert_eq!( + register(&forward, a.id, "call-1", CHILD), + Registered::Registered + ); + assert_eq!( + register(&forward, a.id, "call-1", "20261104_3"), + refused("child_call_conflict") + ); + // Generation B never learns A's child; unknown sessions are never + // admitted once B ends. + forward.seal(PLUGIN, b.id).unwrap(); + assert_eq!( + invoke(&b.url, CHILD).send().await.unwrap().status(), + StatusCode::GONE + ); + assert!(!forward + .registry() + .generations + .values() + .any(|g| g.id == b.id && g.children.contains(CHILD))); + // Discovery is never parked or widened to children. + let discovery = reqwest::Client::new() + .get(format!("{}/api/2.0/serving-endpoints/m", a.url)) + .header("agent-session-id", CHILD) + .send() + .await + .unwrap(); + assert_eq!(discovery.status(), StatusCode::BAD_REQUEST); +} + +#[tokio::test] +async fn children_and_calls_have_their_own_bounds() { + let gateway = fake_gateway().await; + let forward = ready(gateway.origin); + let generation = goose(&forward); + for n in 0..CHILD_CAP { + assert_eq!( + register( + &forward, + generation.id, + &format!("call-{n}"), + &format!("child-{n}") + ), + Registered::Registered + ); + } + assert_eq!( + register(&forward, generation.id, "call-new", "child-new"), + refused("child_cap") + ); + // A resume under a new call adds a call, not a child. + for n in CHILD_CAP..CALL_CAP { + assert_eq!( + register(&forward, generation.id, &format!("call-{n}"), "child-0"), + Registered::Registered + ); + } + let sizes = |forward: &HostForward| { + let registry = forward.registry(); + let g = registry.generations.values().next().unwrap(); + (g.calls.len(), g.children.len()) + }; + assert_eq!(sizes(&forward), (CALL_CAP, CHILD_CAP)); + assert_eq!( + register(&forward, generation.id, "call-0", "child-0"), + Registered::Registered + ); + assert_eq!( + register(&forward, generation.id, "call-resume", "child-0"), + refused("call_cap") + ); + assert_eq!(sizes(&forward), (CALL_CAP, CHILD_CAP)); +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index 0fcef5f93d..00c1334fc5 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -112,7 +112,8 @@ use host_files::{ }; use host_forward::{ plugin_host_forward_admission, plugin_host_forward_associate, plugin_host_forward_close, - plugin_host_forward_open, plugin_host_forward_seal, HostForward, + plugin_host_forward_open, plugin_host_forward_register_child, plugin_host_forward_seal, + HostForward, }; use host_mcp::{ plugin_host_mcp_authorize, plugin_host_mcp_close, plugin_host_mcp_pairing, @@ -762,6 +763,7 @@ fn commands() -> impl Fn(tauri::ipc::Invoke) -> bool + Sen plugin_host_forward_admission, plugin_host_forward_open, plugin_host_forward_associate, + plugin_host_forward_register_child, plugin_host_forward_seal, plugin_host_forward_close, plugin_host_read_file, diff --git a/src/bundled/goose/forward-children.test.ts b/src/bundled/goose/forward-children.test.ts new file mode 100644 index 0000000000..85f96b70c4 --- /dev/null +++ b/src/bundled/goose/forward-children.test.ts @@ -0,0 +1,95 @@ +// @vitest-environment node +import { describe, expect, it, vi } from "vitest"; +import type { ForwardChildRegistered } from "../../features/host/forward"; +import { forwardChildren } from "./forward-children"; +import { flush } from "./goose-testing"; + +const announced = (call: string, title = "delegate") => ({ + sessionUpdate: "tool_call", + toolCallId: call, + title, +}); +const completed = (call: string, child: unknown = "20261104_2") => ({ + sessionUpdate: "tool_call_update", + toolCallId: call, + status: "completed", + _meta: { goose: { subagentSessionId: child } }, +}); + +function tracked( + outcome: () => Promise = async () => ({ + type: "registered", + }), +) { + const register = vi.fn((_call: string, _child: string) => outcome()); + const failed = vi.fn(); + return { observe: forwardChildren({ register, failed }), register, failed }; +} + +describe("forwardChildren", () => { + it("registers a live delegate's child, and its resume under a new call", async () => { + const { observe, register, failed } = tracked(); + observe(announced("call-1"), true); + observe({ ...completed("call-1"), status: "in_progress" }, true); + observe(completed("call-1"), true); + observe(announced("call-2"), true); + observe(completed("call-2"), true); + await flush(); + expect(register.mock.calls).toEqual([ + ["call-1", "20261104_2"], + ["call-2", "20261104_2"], + ]); + expect(failed).not.toHaveBeenCalled(); + }); + + it("ignores replay, other tools and calls it never saw announced", async () => { + const { observe, register } = tracked(); + // A replay arrives outside any prompt, announcement included. + observe(announced("call-1"), false); + observe(completed("call-1"), false); + observe(completed("call-1"), true); + observe(announced("call-2", "shell"), true); + observe(completed("call-2"), true); + observe(announced("call-3"), true); + observe(completed("call-3", 7), true); + observe({ ...completed("call-3"), _meta: undefined }, true); + await flush(); + expect(register).not.toHaveBeenCalled(); + }); + + it("shows each refusal, and stops registering once revoked", async () => { + const outcomes: ForwardChildRegistered[] = [ + { type: "refused", reason: "child_cap" }, + { type: "refused", reason: "generation_revoked" }, + ]; + const { observe, register, failed } = tracked( + async () => outcomes.shift() ?? { type: "registered" }, + ); + for (const call of ["call-1", "call-2", "call-3"]) { + observe(announced(call), true); + observe(completed(call), true); + await flush(); + } + expect(register).toHaveBeenCalledTimes(2); + expect(failed.mock.calls).toEqual([["child_cap"], ["generation_revoked"]]); + }); + + it("surfaces a rejected registration, leaving no promise unhandled", async () => { + const unhandled = vi.fn(); + process.on("unhandledRejection", unhandled); + try { + const { observe, failed } = tracked(() => + Promise.reject("The generation's session is not associated"), + ); + observe(announced("call-1"), true); + observe(completed("call-1"), true); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(failed).toHaveBeenCalledExactlyOnceWith( + "The generation's session is not associated", + ); + expect(unhandled).not.toHaveBeenCalled(); + } finally { + process.off("unhandledRejection", unhandled); + } + }); +}); diff --git a/src/bundled/goose/forward-children.ts b/src/bundled/goose/forward-children.ts new file mode 100644 index 0000000000..710ef4fcdd --- /dev/null +++ b/src/bundled/goose/forward-children.ts @@ -0,0 +1,48 @@ +import type { ForwardChildRegistered } from "../../features/host/forward"; + +/** goose's native delegate tool, as its `tool_call` titles it (goose 1.57.0). */ +const DELEGATE = "delegate"; + +export type ForwardChildren = Readonly<{ + /** `registerForwardChild` bound to this generation's owner and id. */ + register: (call: string, child: string) => Promise; + /** Shows why a child could not be registered on the turn. */ + failed: (reason: string) => void; +}>; + +/** Registers the children goose delegates during one forwarded prompt, as + * an `AcpSession.observe` of the generation's parent session. One per + * generation, so the delegate calls it trusts are this prompt's. Only a live + * `delegate` call announced earlier in the prompt registers: goose strips + * `subagentSessionId` from every other tool's result (goose 1.57.0 + * `lease.rs`), and replayed or out-of-prompt updates are ignored. */ +export function forwardChildren({ register, failed }: ForwardChildren) { + const delegates = new Set(); + let revoked = false; + return (update: Record, prompting: boolean) => { + if (!prompting || revoked) return; + const call = update.toolCallId; + if (typeof call !== "string") return; + if (update.sessionUpdate === "tool_call") { + if (update.title === DELEGATE) delegates.add(call); + return; + } + const meta = update._meta as { goose?: { subagentSessionId?: unknown } }; + const child = meta?.goose?.subagentSessionId; + if ( + update.sessionUpdate !== "tool_call_update" || + update.status !== "completed" || + typeof child !== "string" || + !delegates.has(call) + ) + return; + register(call, child).then( + (registered) => { + if (registered.type === "registered") return; + if (registered.reason === "generation_revoked") revoked = true; + failed(registered.reason); + }, + (error: unknown) => failed(String(error)), + ); + }; +} diff --git a/src/features/agent-harness/acp/client.test.ts b/src/features/agent-harness/acp/client.test.ts index 00818b524e..bcf9cc7cd5 100644 --- a/src/features/agent-harness/acp/client.test.ts +++ b/src/features/agent-harness/acp/client.test.ts @@ -466,6 +466,30 @@ describe("a session with a steering hook", () => { }); describe("a session without hooks", () => { + it("lets an observer see each update and whether a prompt is in flight", async () => { + const harness = fakeHarness({ hold: true }); + const acp = await start(harness); + const session = await acp.open({ cwd: "/work", systemPrompt: "" }); + const seen: [unknown, boolean][] = []; + // The fake harness also announces its runs. + session.observe = (update, prompting) => { + if (update.sessionUpdate !== "run") + seen.push([update.sessionUpdate, prompting]); + }; + // Outside a prompt, as a replay would arrive. + session.update({ sessionUpdate: "tool_call_update" }); + const turn = session.send("first"); + await flush(); + session.update({ sessionUpdate: "tool_call" }); + harness.turns[0]?.finish(); + await turn; + session.update({ sessionUpdate: "tool_call_update" }); + expect(seen).toEqual([ + ["tool_call_update", false], + ["tool_call", true], + ["tool_call_update", false], + ]); + }); it("leads the first prompt with the instructions, once, when ACP's systemPrompt is not advertised", async () => { const harness = fakeHarness(); const acp = await start(harness); diff --git a/src/features/agent-harness/acp/session.ts b/src/features/agent-harness/acp/session.ts index f39247b717..23f88d8f18 100644 --- a/src/features/agent-harness/acp/session.ts +++ b/src/features/agent-harness/acp/session.ts @@ -39,6 +39,11 @@ export class AcpSession { /** Takes the first name the harness generates for it; a titled session gets * none. */ named: ((title: string) => void) | undefined; + /** Sees each update first, and whether a `session/prompt` is in flight + * (not, for a `session/load` replay). */ + observe: + | ((update: Record, prompting: boolean) => void) + | undefined; /** The harness's id for the running turn, which a steer must name. */ private runId: string | undefined; /** Steers waiting for the running turn's id, which the harness sends just @@ -211,6 +216,7 @@ export class AcpSession { /** Called by the process with each `session/update` for this session. */ update(update: Record) { + this.observe?.(update, this.prompting); const kind = update.sessionUpdate; if (kind === "agent_message_chunk" && !this.acted) { const content = update.content as { text?: unknown } | undefined; diff --git a/src/features/host/forward.test.ts b/src/features/host/forward.test.ts index a5be4ca4fd..9932d739ed 100644 --- a/src/features/host/forward.test.ts +++ b/src/features/host/forward.test.ts @@ -6,6 +6,7 @@ import { closeForward, openForward, pushForwardAdmission, + registerForwardChild, sealForward, type ForwardEvent, } from "./forward"; @@ -118,3 +119,35 @@ it("opens a goose generation without a session, then associates it", async () => session: "20261104_1", }); }); + +it("registers a goose child and passes each outcome through", async () => { + const outcomes = [ + { type: "registered" }, + ...[ + "child_call_conflict", + "child_cap", + "call_cap", + "generation_ended", + "generation_revoked", + ].map((reason) => ({ type: "refused", reason })), + ]; + for (const outcome of outcomes) { + vi.mocked(invoke).mockResolvedValueOnce(outcome); + await expect( + registerForwardChild(owner, 7, "call-1", "20261104_2"), + ).resolves.toEqual(outcome); + } + expect(invoke).toHaveBeenLastCalledWith( + "plugin_host_forward_register_child", + { + id: "buzz.agents", + generation: 7, + call: "call-1", + child: "20261104_2", + }, + ); + vi.mocked(invoke).mockRejectedValueOnce("Invalid child session"); + await expect( + registerForwardChild(owner, 7, "call-1", "20261104_1"), + ).rejects.toBe("Invalid child session"); +}); diff --git a/src/features/host/forward.ts b/src/features/host/forward.ts index 88ab73dac2..3cce9df662 100644 --- a/src/features/host/forward.ts +++ b/src/features/host/forward.ts @@ -8,7 +8,13 @@ export type ForwardHarness = "claude" | "goose"; export type ForwardClass = "discovery" | "inference"; export type ForwardEvent = - | Readonly<{ type: "admitted"; request: number; class: ForwardClass }> + | Readonly<{ + type: "admitted"; + request: number; + class: ForwardClass; + /** It named a registered goose child, not the generation's session. */ + child: boolean; + }> | Readonly<{ type: "finished"; request: number; @@ -25,6 +31,20 @@ export type ForwardOpened = /** Not at the route's current revision, or it doesn't admit. */ | Readonly<{ type: "refused"; current: number; admits: boolean }>; +/** A goose child registration. `generation_ended`: sealed, closed or + * unknown; `generation_revoked`: its revision or page was revoked. */ +export type ForwardChildRegistered = + | Readonly<{ type: "registered" }> + | Readonly<{ + type: "refused"; + reason: + | "child_call_conflict" + | "child_cap" + | "call_cap" + | "generation_ended" + | "generation_revoked"; + }>; + export type ForwardPushed = | Readonly<{ type: "applied"; revoked: number }> | Readonly<{ type: "stale" }>; @@ -82,6 +102,22 @@ export function associateForward( }); } +/** Registers the goose child a live delegate `call` named, admitting its + * requests. Rejects on contract, owner or transport errors. */ +export function registerForwardChild( + owner: ForwardOwner, + generation: number, + call: string, + child: string, +): Promise { + return invoke("plugin_host_forward_register_child", { + id: owner.id, + generation, + call, + child, + }); +} + /** Ends the generation's admission for good; admitted streams finish. */ export function sealForward( owner: ForwardOwner, From 15d4e30f3c35f485da6ab0a8c428bf84831662a7 Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:12:44 -0600 Subject: [PATCH 02/21] fix(goose): keep child registration failures inside the observer A registration that throws synchronously, or whose result handling throws, no longer escapes the session update or leaves a rejected promise unhandled; each is reported through `failed`. Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Philip Azar --- src/bundled/goose/forward-children.test.ts | 13 +++++++++++++ src/bundled/goose/forward-children.ts | 22 ++++++++++++++-------- 2 files changed, 27 insertions(+), 8 deletions(-) diff --git a/src/bundled/goose/forward-children.test.ts b/src/bundled/goose/forward-children.test.ts index 85f96b70c4..6c4c91e3f6 100644 --- a/src/bundled/goose/forward-children.test.ts +++ b/src/bundled/goose/forward-children.test.ts @@ -93,3 +93,16 @@ describe("forwardChildren", () => { } }); }); + +it("surfaces a registration that throws instead of returning a promise", () => { + const failed = vi.fn(); + const observe = forwardChildren({ + register: () => { + throw new Error("ipc unavailable"); + }, + failed, + }); + observe(announced("call-1"), true); + expect(() => observe(completed("call-1"), true)).not.toThrow(); + expect(failed).toHaveBeenCalledExactlyOnceWith("Error: ipc unavailable"); +}); diff --git a/src/bundled/goose/forward-children.ts b/src/bundled/goose/forward-children.ts index 710ef4fcdd..939967a6c6 100644 --- a/src/bundled/goose/forward-children.ts +++ b/src/bundled/goose/forward-children.ts @@ -36,13 +36,19 @@ export function forwardChildren({ register, failed }: ForwardChildren) { !delegates.has(call) ) return; - register(call, child).then( - (registered) => { - if (registered.type === "registered") return; - if (registered.reason === "generation_revoked") revoked = true; - failed(registered.reason); - }, - (error: unknown) => failed(String(error)), - ); + // Nothing escapes the update: a throw, rejection or bad result fails + // the turn visibly instead. + const fail = (error: unknown) => failed(String(error)); + try { + register(call, child) + .then((registered) => { + if (registered.type === "registered") return; + if (registered.reason === "generation_revoked") revoked = true; + failed(registered.reason); + }) + .catch(fail); + } catch (error) { + fail(error); + } }; } From fe667b8f1eb8e6d532a017473a4dc07fed95682d Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:16:22 -0600 Subject: [PATCH 03/21] test(host): make child rendezvous timing tests deterministic The deadline case asserts only the refusal and elapsed bound, never the brief parked state; tests that observe parking use a deadline they never reach. A new case parks two requests for the same child, cancels one and shows registration admits only the survivor. Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Philip Azar --- src-tauri/src/host_forward/tests.rs | 43 +++++++++++++++++++++++++---- 1 file changed, 37 insertions(+), 6 deletions(-) diff --git a/src-tauri/src/host_forward/tests.rs b/src-tauri/src/host_forward/tests.rs index 63406c6d7f..14b24446c7 100644 --- a/src-tauri/src/host_forward/tests.rs +++ b/src-tauri/src/host_forward/tests.rs @@ -1435,6 +1435,13 @@ fn pending(forward: &HostForward) -> usize { let registry = forward.registry(); registry.generations.values().map(|g| g.pending.len()).sum() } +/// Ready, with a deadline no test reaches, so a parked request stays parked +/// until the test releases it. +fn patient(origin: &str) -> HostForward { + let mut forward = ready(origin); + forward.rendezvous = Duration::from_secs(3600); + forward +} /// Waits until `count` requests are parked: an observed state, not a delay. async fn parked(forward: &HostForward, count: usize) { tokio::time::timeout(Duration::from_secs(5), async { @@ -1476,7 +1483,7 @@ async fn a_registered_child_is_admitted_under_its_own_session() { #[tokio::test] async fn a_request_first_child_waits_dispatching_nothing_until_registered() { let mut gateway = fake_gateway().await; - let forward = ready(gateway.origin); + let forward = patient(gateway.origin); let generation = goose(&forward); let early = send(invoke(&generation.url, CHILD)); parked(&forward, 1).await; @@ -1500,10 +1507,10 @@ async fn an_unregistered_child_is_refused_at_the_deadline() { let mut forward = ready(gateway.origin); forward.rendezvous = Duration::from_millis(300); let mut generation = goose(&forward); + // Only the outcome is asserted, never the brief parked state; the + // request-first test proves parking. let started = std::time::Instant::now(); - let early = send(invoke(&generation.url, CHILD)); - parked(&forward, 1).await; - let response = early.await.unwrap(); + let response = invoke(&generation.url, CHILD).send().await.unwrap(); assert!(started.elapsed() >= forward.rendezvous); assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT); assert!(response @@ -1532,7 +1539,7 @@ async fn an_unregistered_child_is_refused_at_the_deadline() { #[tokio::test] async fn parked_requests_are_capped_and_free_only_their_own_slot() { let gateway = fake_gateway().await; - let forward = ready(gateway.origin); + let forward = patient(gateway.origin); let mut generation = goose(&forward); let waiting: Vec<_> = (0..PENDING_CAP) .map(|n| send(invoke(&generation.url, &format!("child-{n}")))) @@ -1561,10 +1568,34 @@ async fn parked_requests_are_capped_and_free_only_their_own_slot() { assert_eq!(pending(&forward), PENDING_CAP - 2); } +#[tokio::test] +async fn a_cancelled_request_leaves_its_same_child_sibling_parked() { + let mut gateway = fake_gateway().await; + let forward = patient(gateway.origin); + let generation = goose(&forward); + let cancelled = send(invoke(&generation.url, CHILD)); + parked(&forward, 1).await; + let survivor = send(invoke(&generation.url, CHILD)); + parked(&forward, 2).await; + cancelled.abort(); + parked(&forward, 1).await; + assert_eq!( + register(&forward, generation.id, "call-1", CHILD), + Registered::Registered + ); + assert_eq!(survivor.await.unwrap().status(), StatusCode::OK); + assert_eq!( + recv(&mut gateway.seen).await.headers["agent-session-id"], + CHILD + ); + assert_eq!(pending(&forward), 0); + assert!(gateway.seen.try_recv().is_err()); +} + #[tokio::test] async fn seal_revocation_and_close_release_every_parked_request() { let gateway = fake_gateway().await; - let forward = ready(gateway.origin); + let forward = patient(gateway.origin); let generation = goose(&forward); let early = send(invoke(&generation.url, CHILD)); parked(&forward, 1).await; From 67e726c0d0baf27de40d6fbd52ddee553b734526 Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:24:49 -0600 Subject: [PATCH 04/21] fix(goose): contain a throwing child failure reporter; bound parked test waits The observer's failure reporter is now called at most once per failure and never from its own error. A reporter that throws stops registration, so a later child's inference is refused as child_unregistered instead of running unseen. Covered on the synchronous, refusal and rejection paths. Parked rendezvous tests keep their 3600s deadline but bound each response wait at 5s, so a lost wake fails fast. Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Philip Azar --- src-tauri/src/host_forward/tests.rs | 34 ++++++------ src/bundled/goose/forward-children.test.ts | 61 ++++++++++++++++++++++ src/bundled/goose/forward-children.ts | 40 ++++++++++---- 3 files changed, 109 insertions(+), 26 deletions(-) diff --git a/src-tauri/src/host_forward/tests.rs b/src-tauri/src/host_forward/tests.rs index 14b24446c7..90947da317 100644 --- a/src-tauri/src/host_forward/tests.rs +++ b/src-tauri/src/host_forward/tests.rs @@ -1455,6 +1455,14 @@ async fn parked(forward: &HostForward, count: usize) { fn send(request: reqwest::RequestBuilder) -> tokio::task::JoinHandle { tokio::spawn(async move { request.send().await.unwrap() }) } +/// A parked request's response, bounded well inside `patient`'s deadline so +/// a lost wake fails the test in seconds instead of holding it for an hour. +async fn answered(response: tokio::task::JoinHandle) -> reqwest::Response { + tokio::time::timeout(Duration::from_secs(5), response) + .await + .expect("answered in time") + .unwrap() +} #[tokio::test] async fn a_registered_child_is_admitted_under_its_own_session() { @@ -1492,7 +1500,7 @@ async fn a_request_first_child_waits_dispatching_nothing_until_registered() { register(&forward, generation.id, "call-1", CHILD), Registered::Registered ); - assert_eq!(early.await.unwrap().status(), StatusCode::OK); + assert_eq!(answered(early).await.status(), StatusCode::OK); assert_eq!( recv(&mut gateway.seen).await.headers["agent-session-id"], CHILD @@ -1510,7 +1518,7 @@ async fn an_unregistered_child_is_refused_at_the_deadline() { // Only the outcome is asserted, never the brief parked state; the // request-first test proves parking. let started = std::time::Instant::now(); - let response = invoke(&generation.url, CHILD).send().await.unwrap(); + let response = answered(send(invoke(&generation.url, CHILD))).await; assert!(started.elapsed() >= forward.rendezvous); assert_eq!(response.status(), StatusCode::GATEWAY_TIMEOUT); assert!(response @@ -1527,10 +1535,8 @@ async fn an_unregistered_child_is_refused_at_the_deadline() { // A registration after expiry still admits a retry; expiry never did. register(&forward, generation.id, "call-1", CHILD); assert_eq!( - invoke(&generation.url, CHILD) - .send() + answered(send(invoke(&generation.url, CHILD))) .await - .unwrap() .status(), StatusCode::OK ); @@ -1545,7 +1551,7 @@ async fn parked_requests_are_capped_and_free_only_their_own_slot() { .map(|n| send(invoke(&generation.url, &format!("child-{n}")))) .collect(); parked(&forward, PENDING_CAP).await; - let full = invoke(&generation.url, "child-x").send().await.unwrap(); + let full = answered(send(invoke(&generation.url, "child-x"))).await; assert_eq!(full.status(), StatusCode::TOO_MANY_REQUESTS); assert_eq!( recv(&mut generation.events).await, @@ -1560,9 +1566,9 @@ async fn parked_requests_are_capped_and_free_only_their_own_slot() { parked(&forward, PENDING_CAP).await; // Both requests for child-1 are woken by one registration. register(&forward, generation.id, "call-1", "child-1"); - assert_eq!(again.await.unwrap().status(), StatusCode::OK); + assert_eq!(answered(again).await.status(), StatusCode::OK); assert_eq!( - waiting.next().unwrap().await.unwrap().status(), + answered(waiting.next().unwrap()).await.status(), StatusCode::OK ); assert_eq!(pending(&forward), PENDING_CAP - 2); @@ -1583,7 +1589,7 @@ async fn a_cancelled_request_leaves_its_same_child_sibling_parked() { register(&forward, generation.id, "call-1", CHILD), Registered::Registered ); - assert_eq!(survivor.await.unwrap().status(), StatusCode::OK); + assert_eq!(answered(survivor).await.status(), StatusCode::OK); assert_eq!( recv(&mut gateway.seen).await.headers["agent-session-id"], CHILD @@ -1600,16 +1606,14 @@ async fn seal_revocation_and_close_release_every_parked_request() { let early = send(invoke(&generation.url, CHILD)); parked(&forward, 1).await; forward.seal(PLUGIN, generation.id).unwrap(); - assert_eq!(early.await.unwrap().status(), StatusCode::GONE); + assert_eq!(answered(early).await.status(), StatusCode::GONE); assert_eq!( register(&forward, generation.id, "call-1", CHILD), refused("generation_ended") ); assert_eq!( - invoke(&generation.url, CHILD) - .send() + answered(send(invoke(&generation.url, CHILD))) .await - .unwrap() .status(), StatusCode::GONE ); @@ -1620,7 +1624,7 @@ async fn seal_revocation_and_close_release_every_parked_request() { forward .push(forward.page(), PLUGIN, "route-a", 2, true) .unwrap(); - let response = early.await.unwrap(); + let response = answered(early).await; assert_eq!(response.status(), StatusCode::GONE); assert!(response .text() @@ -1637,7 +1641,7 @@ async fn seal_revocation_and_close_release_every_parked_request() { let early = send(invoke(&closed.url, CHILD)); parked(&forward, 1).await; forward.close(PLUGIN, closed.id).unwrap(); - assert_eq!(early.await.unwrap().status(), StatusCode::GONE); + assert_eq!(answered(early).await.status(), StatusCode::GONE); assert_eq!( register(&forward, closed.id, "call-1", CHILD), refused("generation_ended") diff --git a/src/bundled/goose/forward-children.test.ts b/src/bundled/goose/forward-children.test.ts index 6c4c91e3f6..7abc4af76c 100644 --- a/src/bundled/goose/forward-children.test.ts +++ b/src/bundled/goose/forward-children.test.ts @@ -106,3 +106,64 @@ it("surfaces a registration that throws instead of returning a promise", () => { expect(() => observe(completed("call-1"), true)).not.toThrow(); expect(failed).toHaveBeenCalledExactlyOnceWith("Error: ipc unavailable"); }); + +describe("a failure reporter that throws", () => { + const throwing = () => + vi.fn((_reason: string) => { + throw new Error("turn already ended"); + }); + + it("is called once on a synchronous failure, and stops registration", () => { + const logged = vi.spyOn(console, "error").mockImplementation(() => {}); + try { + const failed = throwing(); + const register = vi.fn((_call: string, _child: string) => { + throw new Error("ipc unavailable"); + }); + const observe = forwardChildren({ register, failed }); + observe(announced("call-1"), true); + expect(() => observe(completed("call-1"), true)).not.toThrow(); + observe(announced("call-2"), true); + observe(completed("call-2"), true); + expect(failed).toHaveBeenCalledExactlyOnceWith("Error: ipc unavailable"); + expect(register).toHaveBeenCalledOnce(); + expect(logged).toHaveBeenCalledOnce(); + } finally { + logged.mockRestore(); + } + }); + + it.each([ + [ + "a refusal", + async () => ({ type: "refused", reason: "child_cap" }) as const, + "child_cap", + ], + ["a rejection", () => Promise.reject("not associated"), "not associated"], + ] as const)( + "is called once on %s, leaving no promise unhandled", + async (_, outcome, reason) => { + const unhandled = vi.fn(); + const logged = vi.spyOn(console, "error").mockImplementation(() => {}); + process.on("unhandledRejection", unhandled); + try { + const failed = throwing(); + const register = vi.fn((_call: string, _child: string) => outcome()); + const observe = forwardChildren({ register, failed }); + observe(announced("call-1"), true); + observe(completed("call-1"), true); + await new Promise((resolve) => setTimeout(resolve, 0)); + observe(announced("call-2"), true); + observe(completed("call-2"), true); + await new Promise((resolve) => setTimeout(resolve, 0)); + expect(failed).toHaveBeenCalledExactlyOnceWith(reason); + expect(register).toHaveBeenCalledOnce(); + expect(logged).toHaveBeenCalledOnce(); + expect(unhandled).not.toHaveBeenCalled(); + } finally { + process.off("unhandledRejection", unhandled); + logged.mockRestore(); + } + }, + ); +}); diff --git a/src/bundled/goose/forward-children.ts b/src/bundled/goose/forward-children.ts index 939967a6c6..64ba72d886 100644 --- a/src/bundled/goose/forward-children.ts +++ b/src/bundled/goose/forward-children.ts @@ -18,9 +18,24 @@ export type ForwardChildren = Readonly<{ * `lease.rs`), and replayed or out-of-prompt updates are ignored. */ export function forwardChildren({ register, failed }: ForwardChildren) { const delegates = new Set(); - let revoked = false; + let stopped = false; + // The turn's failure signal, called once per failure and never again from + // its own error. A reporter that throws stops registration too, so every + // later child's inference is refused (`child_unregistered`) instead of + // running unseen. + const report = (reason: string) => { + try { + failed(reason); + } catch (error) { + stopped = true; + console.error("A goose child registration failure went unreported", { + reason, + error, + }); + } + }; return (update: Record, prompting: boolean) => { - if (!prompting || revoked) return; + if (!prompting || stopped) return; const call = update.toolCallId; if (typeof call !== "string") return; if (update.sessionUpdate === "tool_call") { @@ -38,17 +53,20 @@ export function forwardChildren({ register, failed }: ForwardChildren) { return; // Nothing escapes the update: a throw, rejection or bad result fails // the turn visibly instead. - const fail = (error: unknown) => failed(String(error)); + let registering: Promise; try { - register(call, child) - .then((registered) => { - if (registered.type === "registered") return; - if (registered.reason === "generation_revoked") revoked = true; - failed(registered.reason); - }) - .catch(fail); + registering = register(call, child); } catch (error) { - fail(error); + report(String(error)); + return; } + registering.then( + (registered) => { + if (registered.type === "registered") return; + if (registered.reason === "generation_revoked") stopped = true; + report(registered.reason); + }, + (error) => report(String(error)), + ); }; } From ddda098369f6de0aad9bcea2e86fe5a58d4de0b3 Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:30:13 -0600 Subject: [PATCH 05/21] feat(acp): opt-in session/load in AcpProcess.open `open({ load })` resumes a saved session or fails visibly; it never opens a new session in its place. It needs `loadSession`, refuses a system prompt that only `session/new` could carry, and lets one open own a saved id at a time, including one whose close the harness has not acknowledged. The session is registered only after the load answers, so replayed history never reaches it. Naming, title and new-session meta are skipped; the Buzz MCP server is declared afresh with a new token; a hook system prompt is set again under its key, and a first-prompt one is not repeated. Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Philip Azar --- src/features/agent-harness/acp/client.test.ts | 183 +++++++++++++++++- src/features/agent-harness/acp/process.ts | 125 ++++++++---- 2 files changed, 267 insertions(+), 41 deletions(-) diff --git a/src/features/agent-harness/acp/client.test.ts b/src/features/agent-harness/acp/client.test.ts index bcf9cc7cd5..3eb8e878b3 100644 --- a/src/features/agent-harness/acp/client.test.ts +++ b/src/features/agent-harness/acp/client.test.ts @@ -12,7 +12,8 @@ import type { HostProcessOptions, } from "../../host/service"; import { AgentSessions, ServerPool, SHARED_POOL } from "./agent-sessions"; -import { AcpProcess, REQUEST_TIMEOUT_MS } from "./process"; +import { AcpProcess, REQUEST_TIMEOUT_MS, TOOL_SERVER } from "./process"; +import { AcpSession } from "./session"; import type { AcpHooks, ToolServer } from "./types"; type Message = { @@ -53,6 +54,8 @@ function fakeHarness( promptError?: { code: number; message: string }; /** Why a turn that ran ends. */ stopReason?: string; + /** Every `session/load` is answered with this error. */ + loadError?: { code: number; message: string }; }> = {}, ) { const started: { @@ -135,7 +138,20 @@ function fakeHarness( else if (method === "session/new") reply(id, { sessionId: `session-${++sessions}` }); else if (method === "session/close") reply(id, {}); - else if (method === "session/prompt" && rejectPrompts > 0) { + else if (method === "session/load" && options.loadError) + send({ id, error: options.loadError }); + else if (method === "session/load") { + // As goose does: the saved conversation replays before the answer. + for (const sessionUpdate of ["user_message_chunk", "tool_call"]) + send({ + method: "session/update", + params: { + sessionId: params.sessionId, + update: { sessionUpdate, toolCallId: "call-1" }, + }, + }); + reply(id, {}); + } else if (method === "session/prompt" && rejectPrompts > 0) { rejectPrompts--; send({ id, error: { code: -32000, message: "Sign in first" } }); } else if (method === "session/prompt" && options.promptError) @@ -665,3 +681,166 @@ describe("the shared server pool", () => { genericSessions.dispose(); }); }); + +describe("resuming a saved session", () => { + const LOADS = { + mcpCapabilities: { http: true }, + loadSession: true, + sessionCapabilities: { close: {} }, + }; + + it("loads it with fresh tools, never seeing its replay, naming or new-session meta", async () => { + const names = vi.fn(async () => true); + const newSessionMeta = vi.fn(() => ({})); + const systemPrompt = vi.fn(async () => undefined); + const hooks: AcpHooks = { ...HOOKS, names, newSessionMeta, systemPrompt }; + const harness = fakeHarness({ capabilities: LOADS }); + const acp = await start(harness, {}, hooks); + const first = await acp.open({ + cwd: "/work", + systemPrompt: "Be Buzz.", + tools, + }); + await acp.close(first); + names.mockClear(); + newSessionMeta.mockClear(); + systemPrompt.mockClear(); + const updates = vi.spyOn(AcpSession.prototype, "update"); + const seen: unknown[] = []; + const named = vi.fn(); + try { + const session = await acp.open({ + cwd: "/work", + systemPrompt: "Be Buzz.", + title: "A new title", + named, + tools, + load: first.id, + onSession: (opened) => { + opened.observe = (update) => seen.push(update); + }, + }); + await flush(); + expect(session.id).toBe(first.id); + expect(updates).not.toHaveBeenCalled(); + expect(seen).toEqual([]); + expect(harness.sent("session/new")).toHaveLength(1); + // The first session's server was token-1; its token is never reused. + expect(harness.sent("session/load")[0]?.params).toEqual({ + sessionId: first.id, + cwd: "/work", + mcpServers: [ + { + type: "http", + name: TOOL_SERVER, + url: "http://127.0.0.1:40002/mcp", + headers: [{ name: "Authorization", value: "Bearer token-2" }], + }, + ], + }); + expect(names).not.toHaveBeenCalled(); + expect(newSessionMeta).not.toHaveBeenCalled(); + expect(named).not.toHaveBeenCalled(); + // Set under its key again, which replaces it rather than adding one. + expect(systemPrompt).toHaveBeenCalledExactlyOnceWith( + acp, + first.id, + "Be Buzz.", + ); + } finally { + updates.mockRestore(); + } + }); + + it("leads no prompt with the instructions its conversation already holds", async () => { + const harness = fakeHarness({ capabilities: LOADS }); + const acp = await start(harness); + const session = await acp.open({ + cwd: "/work", + systemPrompt: "Be Buzz.", + load: "saved-1", + }); + await session.send("next"); + expect(harness.turns.map((turn) => turn.text)).toEqual(["next"]); + }); + + it("refuses, never opening a new session, without loadSession or when the load fails", async () => { + const without = fakeHarness(); + const plain = await start(without); + await expect( + plain.open({ cwd: "/work", systemPrompt: "", load: "saved-1" }), + ).rejects.toThrow("This Harness cannot resume a session"); + expect(without.sent("session/load")).toEqual([]); + expect(without.sent("session/new")).toEqual([]); + + const failing = fakeHarness({ + capabilities: LOADS, + loadError: { code: -32002, message: "Session not found: saved-1" }, + }); + const acp = await start(failing); + await expect( + acp.open({ cwd: "/work", systemPrompt: "", tools, load: "saved-1" }), + ).rejects.toThrow("Session not found: saved-1"); + expect(failing.sent("session/new")).toEqual([]); + // Nothing holds the id: a later load may try again. + failing.sent("session/load").length = 0; + await expect( + acp.open({ cwd: "/work", systemPrompt: "", load: "saved-1" }), + ).rejects.toThrow("Session not found: saved-1"); + }); + + it("refuses a system prompt it would have to send with session/new", async () => { + const harness = fakeHarness({ + capabilities: { + ...LOADS, + sessionCapabilities: { close: {}, systemPrompt: {} }, + }, + }); + const acp = await start(harness); + await expect( + acp.open({ cwd: "/work", systemPrompt: "Be Buzz.", load: "saved-1" }), + ).rejects.toThrow( + "This Harness cannot resume a session with its system prompt", + ); + expect(harness.sent("session/load")).toEqual([]); + }); + + it("lets one open own a saved session at a time", async () => { + const harness = fakeHarness({ capabilities: LOADS }); + const acp = await start(harness); + const loading = acp.open({ + cwd: "/work", + systemPrompt: "", + load: "saved-1", + }); + await expect( + acp.open({ cwd: "/work", systemPrompt: "", load: "saved-1" }), + ).rejects.toThrow("This Harness session is already open"); + const session = await loading; + await expect( + acp.open({ cwd: "/work", systemPrompt: "", load: "saved-1" }), + ).rejects.toThrow("This Harness session is already open"); + expect(harness.sent("session/load")).toHaveLength(1); + await acp.close(session); + await expect( + acp.open({ cwd: "/work", systemPrompt: "", load: "saved-1" }), + ).resolves.toMatchObject({ id: "saved-1" }); + }); + + it("waits for the harness to acknowledge a close before loading it again", async () => { + const harness = fakeHarness({ + capabilities: { ...LOADS, sessionCapabilities: {} }, + }); + const acp = await start(harness); + const session = await acp.open({ + cwd: "/work", + systemPrompt: "", + load: "saved-1", + }); + await acp.close(session); + await expect( + acp.open({ cwd: "/work", systemPrompt: "", load: "saved-1" }), + ).rejects.toThrow("This Harness session has not finished closing"); + expect(harness.sent("session/load")).toHaveLength(1); + }); +}); diff --git a/src/features/agent-harness/acp/process.ts b/src/features/agent-harness/acp/process.ts index 4ce6fcddf4..925f115232 100644 --- a/src/features/agent-harness/acp/process.ts +++ b/src/features/agent-harness/acp/process.ts @@ -61,11 +61,28 @@ type Pending = { reject: (error: Error) => void; }; +type AcpLaunch = Readonly<{ + cwd: string; + systemPrompt: string; + title?: string; + named?: (title: string) => void; + tools?: ToolServer; + choice?: () => Choice; + /** The id of a saved session to resume with `session/load`. It keeps + * its name, so `title` and `named` are ignored. */ + load?: string; + /** Given the session as soon as the harness has opened it, before it + * is set up, so a caller can close one whose open then fails. */ + onSession?: (session: AcpSession) => void; +}>; + export class AcpProcess { readonly exited: Promise; private readonly sessions = new Map(); /** Sessions whose close is in flight or failed, by id. */ private readonly closing = new Map>(); + /** Saved sessions being resumed, each by one `open` at a time. */ + private readonly loading = new Set(); private readonly calls = new Map(); private requests = 0; private buffer = ""; @@ -221,28 +238,47 @@ export class AcpProcess { * `systemPrompt` appended. With `named`, the harness names it instead when * it names sessions, and `named` takes that name. `tools` answers its Buzz * tools; without them it has no tools at all. Each turn runs on the - * provider and model `choice` gives as it starts. */ - async open( - launch: Readonly<{ - cwd: string; - systemPrompt: string; - title?: string; - named?: (title: string) => void; - tools?: ToolServer; - choice?: () => Choice; - /** Given the session as soon as the harness has opened it, before it - * is set up, so a caller can close one whose open then fails. */ - onSession?: (session: AcpSession) => void; - }>, - ): Promise { + * provider and model `choice` gives as it starts. With `load`, it resumes + * that saved session instead, or fails: never a new one in its place. */ + async open(launch: AcpLaunch): Promise { + const { load } = launch; + if (load === undefined) return this.opening(launch); + const { label } = this.hooks; + if (this.capabilities.loadSession !== true) + throw new Error(`This ${label} cannot resume a session`); + // Whether a harness keeps a `session/new` system prompt is unknown. + if ( + launch.systemPrompt && + !this.hooks.systemPrompt && + this.advertises("systemPrompt") + ) + throw new Error( + `This ${label} cannot resume a session with its system prompt`, + ); + if (this.sessions.has(load) || this.loading.has(load)) + throw new Error(`This ${label} session is already open`); + // Including a close the harness never acknowledged. + if (this.closing.has(load)) + throw new Error(`This ${label} session has not finished closing`); + this.loading.add(load); + try { + return await this.opening(launch); + } finally { + this.loading.delete(load); + } + } + + private async opening(launch: AcpLaunch): Promise { const { hooks } = this; - const named = - launch.named && (await this.names()) ? launch.named : undefined; - const title = named ? undefined : launch.title; - const provider = launch.choice?.().provider ?? ""; - const { systemPrompt } = launch; + const { load, systemPrompt } = launch; const prompting = !!systemPrompt && !hooks.systemPrompt; const standard = prompting && this.advertises("systemPrompt"); + const named = + load === undefined && launch.named && (await this.names()) + ? launch.named + : undefined; + const title = named || load !== undefined ? undefined : launch.title; + const provider = launch.choice?.().provider ?? ""; let server: HostMcpServer | undefined; // The harness connects while it opens the session, before it is known here. let session: AcpSession | undefined; @@ -272,26 +308,36 @@ export class AcpProcess { let sessionId: string; let opened: ConfigOptions; try { - ({ sessionId, ...opened } = (await this.request( - "session/new", - { - cwd: launch.cwd, - mcpServers: buzz ? [buzz] : [], - ...(standard && { - systemPrompt: { - mode: "append", - content: [{ type: "text", text: systemPrompt }], - }, - }), - ...(hooks.newSessionMeta && { - _meta: hooks.newSessionMeta(buzz, { - ...(title ? { title } : {}), - provider, + // Replayed history arrives before the answer, while the session is + // not yet in `sessions`, so none of it reaches the session. + if (load !== undefined) { + sessionId = load; + opened = (await this.request( + "session/load", + { sessionId, cwd: launch.cwd, mcpServers: buzz ? [buzz] : [] }, + REQUEST_TIMEOUT_MS, + )) as ConfigOptions; + } else + ({ sessionId, ...opened } = (await this.request( + "session/new", + { + cwd: launch.cwd, + mcpServers: buzz ? [buzz] : [], + ...(standard && { + systemPrompt: { + mode: "append", + content: [{ type: "text", text: systemPrompt }], + }, }), - }), - }, - REQUEST_TIMEOUT_MS, - )) as ConfigOptions & { sessionId: string }); + ...(hooks.newSessionMeta && { + _meta: hooks.newSessionMeta(buzz, { + ...(title ? { title } : {}), + provider, + }), + }), + }, + REQUEST_TIMEOUT_MS, + )) as ConfigOptions & { sessionId: string }); } catch (error) { void server?.close(); throw error; @@ -302,7 +348,8 @@ export class AcpProcess { server, launch.choice, { provider, model: "" }, - prompting && !standard ? systemPrompt : "", + // A resumed conversation already opens with it. + prompting && !standard && load === undefined ? systemPrompt : "", ); session.offer(opened); session.named = named; From e52c64ed2862121a387116fcb96bb5260dc50295 Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:33:48 -0600 Subject: [PATCH 06/21] feat(goose): keep a generation's child registration failure for its turn `forwardChildren` now returns `{ observe, settled }`. `settled()` resolves, once every registration the generation started has been answered, to its first failure, kept before the notifier runs and whatever the notifier does. The caller awaits it after the prompt and before reporting success. Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Philip Azar --- src/bundled/goose/forward-children.test.ts | 90 ++++++++++++++++++++-- src/bundled/goose/forward-children.ts | 28 +++++-- 2 files changed, 104 insertions(+), 14 deletions(-) diff --git a/src/bundled/goose/forward-children.test.ts b/src/bundled/goose/forward-children.test.ts index 7abc4af76c..b33f1f4eea 100644 --- a/src/bundled/goose/forward-children.test.ts +++ b/src/bundled/goose/forward-children.test.ts @@ -23,12 +23,13 @@ function tracked( ) { const register = vi.fn((_call: string, _child: string) => outcome()); const failed = vi.fn(); - return { observe: forwardChildren({ register, failed }), register, failed }; + const { observe, settled } = forwardChildren({ register, failed }); + return { observe, settled, register, failed }; } describe("forwardChildren", () => { it("registers a live delegate's child, and its resume under a new call", async () => { - const { observe, register, failed } = tracked(); + const { observe, settled, register, failed } = tracked(); observe(announced("call-1"), true); observe({ ...completed("call-1"), status: "in_progress" }, true); observe(completed("call-1"), true); @@ -40,6 +41,7 @@ describe("forwardChildren", () => { ["call-2", "20261104_2"], ]); expect(failed).not.toHaveBeenCalled(); + expect(await settled()).toBeUndefined(); }); it("ignores replay, other tools and calls it never saw announced", async () => { @@ -62,7 +64,7 @@ describe("forwardChildren", () => { { type: "refused", reason: "child_cap" }, { type: "refused", reason: "generation_revoked" }, ]; - const { observe, register, failed } = tracked( + const { observe, settled, register, failed } = tracked( async () => outcomes.shift() ?? { type: "registered" }, ); for (const call of ["call-1", "call-2", "call-3"]) { @@ -72,6 +74,8 @@ describe("forwardChildren", () => { } expect(register).toHaveBeenCalledTimes(2); expect(failed.mock.calls).toEqual([["child_cap"], ["generation_revoked"]]); + // The turn fails for the first. + expect(await settled()).toBe("child_cap"); }); it("surfaces a rejected registration, leaving no promise unhandled", async () => { @@ -96,7 +100,7 @@ describe("forwardChildren", () => { it("surfaces a registration that throws instead of returning a promise", () => { const failed = vi.fn(); - const observe = forwardChildren({ + const { observe } = forwardChildren({ register: () => { throw new Error("ipc unavailable"); }, @@ -113,19 +117,20 @@ describe("a failure reporter that throws", () => { throw new Error("turn already ended"); }); - it("is called once on a synchronous failure, and stops registration", () => { + it("is called once on a synchronous failure, and stops registration", async () => { const logged = vi.spyOn(console, "error").mockImplementation(() => {}); try { const failed = throwing(); const register = vi.fn((_call: string, _child: string) => { throw new Error("ipc unavailable"); }); - const observe = forwardChildren({ register, failed }); + const { observe, settled } = forwardChildren({ register, failed }); observe(announced("call-1"), true); expect(() => observe(completed("call-1"), true)).not.toThrow(); observe(announced("call-2"), true); observe(completed("call-2"), true); expect(failed).toHaveBeenCalledExactlyOnceWith("Error: ipc unavailable"); + expect(await settled()).toBe("Error: ipc unavailable"); expect(register).toHaveBeenCalledOnce(); expect(logged).toHaveBeenCalledOnce(); } finally { @@ -149,7 +154,7 @@ describe("a failure reporter that throws", () => { try { const failed = throwing(); const register = vi.fn((_call: string, _child: string) => outcome()); - const observe = forwardChildren({ register, failed }); + const { observe, settled } = forwardChildren({ register, failed }); observe(announced("call-1"), true); observe(completed("call-1"), true); await new Promise((resolve) => setTimeout(resolve, 0)); @@ -157,6 +162,7 @@ describe("a failure reporter that throws", () => { observe(completed("call-2"), true); await new Promise((resolve) => setTimeout(resolve, 0)); expect(failed).toHaveBeenCalledExactlyOnceWith(reason); + expect(await settled()).toBe(reason); expect(register).toHaveBeenCalledOnce(); expect(logged).toHaveBeenCalledOnce(); expect(unhandled).not.toHaveBeenCalled(); @@ -167,3 +173,73 @@ describe("a failure reporter that throws", () => { }, ); }); + +it("keeps a failure for the turn when the only child's report throws", async () => { + const logged = vi.spyOn(console, "error").mockImplementation(() => {}); + try { + const { observe, settled } = forwardChildren({ + register: async () => ({ + type: "refused", + reason: "child_call_conflict", + }), + failed: () => { + throw new Error("turn already ended"); + }, + }); + observe(announced("call-1"), true); + observe(completed("call-1"), true); + await flush(); + // No later child request exists to be refused: only this signal remains. + expect(await settled()).toBe("child_call_conflict"); + } finally { + logged.mockRestore(); + } +}); + +it("fails the turn for a registration still unanswered when its prompt ends", async () => { + let answer!: (registered: ForwardChildRegistered) => void; + const { observe, settled, failed } = tracked( + () => new Promise((resolve) => (answer = resolve)), + ); + observe(announced("call-1"), true); + observe(completed("call-1"), true); + // The prompt has ended; the registration has not been answered. + let outcome: string | undefined | null = null; + const checking = settled().then((failure) => (outcome = failure)); + await flush(); + expect(outcome).toBeNull(); + answer({ type: "refused", reason: "generation_ended" }); + await checking; + expect(outcome).toBe("generation_ended"); + expect(failed).toHaveBeenCalledExactlyOnceWith("generation_ended"); +}); + +it("fails the turn for a held registration that rejects into a throwing notifier", async () => { + const logged = vi.spyOn(console, "error").mockImplementation(() => {}); + try { + let reject!: (error: unknown) => void; + const failed = vi.fn(() => { + throw new Error("turn already ended"); + }); + const { observe, settled } = forwardChildren({ + register: () => new Promise((_, rejected) => (reject = rejected)), + failed, + }); + observe(announced("call-1"), true); + observe(completed("call-1"), true); + // The prompt settles while the IPC answer is held: success stays blocked. + let checked = false; + const checking = settled().then((failure) => { + checked = true; + return failure; + }); + await flush(); + expect(checked).toBe(false); + // No later child follows. + reject("ipc closed"); + await expect(checking).resolves.toBe("ipc closed"); + expect(failed).toHaveBeenCalledOnce(); + } finally { + logged.mockRestore(); + } +}); diff --git a/src/bundled/goose/forward-children.ts b/src/bundled/goose/forward-children.ts index 64ba72d886..df41a197bf 100644 --- a/src/bundled/goose/forward-children.ts +++ b/src/bundled/goose/forward-children.ts @@ -11,7 +11,7 @@ export type ForwardChildren = Readonly<{ }>; /** Registers the children goose delegates during one forwarded prompt, as - * an `AcpSession.observe` of the generation's parent session. One per + * its `observe` as the generation's parent session's. One per * generation, so the delegate calls it trusts are this prompt's. Only a live * `delegate` call announced earlier in the prompt registers: goose strips * `subagentSessionId` from every other tool's result (goose 1.57.0 @@ -19,11 +19,14 @@ export type ForwardChildren = Readonly<{ export function forwardChildren({ register, failed }: ForwardChildren) { const delegates = new Set(); let stopped = false; - // The turn's failure signal, called once per failure and never again from - // its own error. A reporter that throws stops registration too, so every - // later child's inference is refused (`child_unregistered`) instead of - // running unseen. + /** The first failure, kept whatever `failed` does with it. */ + let failure: string | undefined; + /** Registrations not yet answered, none of which ever rejects. */ + const pending = new Set>(); + // Kept first, then shown once and never again from its own error. A + // `failed` that throws stops registration too: nothing more is trusted. const report = (reason: string) => { + failure ??= reason; try { failed(reason); } catch (error) { @@ -34,7 +37,7 @@ export function forwardChildren({ register, failed }: ForwardChildren) { }); } }; - return (update: Record, prompting: boolean) => { + const observe = (update: Record, prompting: boolean) => { if (!prompting || stopped) return; const call = update.toolCallId; if (typeof call !== "string") return; @@ -60,7 +63,7 @@ export function forwardChildren({ register, failed }: ForwardChildren) { report(String(error)); return; } - registering.then( + const answered = registering.then( (registered) => { if (registered.type === "registered") return; if (registered.reason === "generation_revoked") stopped = true; @@ -68,5 +71,16 @@ export function forwardChildren({ register, failed }: ForwardChildren) { }, (error) => report(String(error)), ); + pending.add(answered); + void answered.finally(() => pending.delete(answered)); }; + /** The turn's registration failure, once every registration started has + * been answered. The caller awaits it after the prompt and before + * reporting success, so a failed registration fails the turn even when + * nothing else would notice. */ + const settled = async () => { + while (pending.size) await Promise.all(pending); + return failure; + }; + return { observe, settled }; } From 7631141d7185e61db097926338cf365d37d6b0e2 Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:41:19 -0600 Subject: [PATCH 07/21] feat(acp): hold a conversation for awaited handoff work `AgentSessions.exclusive(key, work)` gives `work` the conversation to itself. Messages wait; its live session finishes and is closed, and work starts only once that close and every earlier one of the conversation's sessions is acknowledged or their process has exited. Never after one that failed, nor once interrupted or stopped. Saved sessions are never deleted. Callers waiting on an opening session or on another hold re-check and reserve with nothing awaited between. `interrupt` and `dispose` abort the work's signal without waiting for it; `shutdown` waits for it. A session the work leaves running (`outlives`) blocks the next hold until it ends. `keep(id)` reorders the saved sessions only while the work holds the key. Co-authored-by: coder2 <4acc49cb93576448ba82e631688c51a40bf03d6a889d78fddf57917d198de10f@buzz.block.builderlab.xyz> Co-Authored-By: Claude Opus 5.5 Signed-off-by: Philip Azar --- .../agent-harness/acp/agent-sessions.ts | 140 +++++++++-- src/features/agent-harness/acp/client.test.ts | 235 +++++++++++++++++- 2 files changed, 359 insertions(+), 16 deletions(-) diff --git a/src/features/agent-harness/acp/agent-sessions.ts b/src/features/agent-harness/acp/agent-sessions.ts index ba32eb1dbd..05d88dc595 100644 --- a/src/features/agent-harness/acp/agent-sessions.ts +++ b/src/features/agent-harness/acp/agent-sessions.ts @@ -156,6 +156,22 @@ export type SessionsOptions = Readonly<{ onChange?(): void; }>; +/** A conversation `exclusive` holds. */ +type Held = { controller: AbortController; done: Promise }; + +/** What `exclusive` gives the work it runs. */ +export type Exclusive = Readonly<{ + /** The conversation's newest saved session, if any. */ + saved: string | undefined; + /** Makes `id` its newest saved session, while the work holds it. */ + keep(id: string): void; + /** Aborts on `interrupt` or `dispose`. */ + signal: AbortSignal; + /** A session the work opened outlives it until `done`: until then the + * conversation cannot be held again. */ + outlives(done: Promise): void; +}>; + type Live = { session: AcpSession; process: AcpProcess; @@ -182,6 +198,10 @@ export class AgentSessions { string, Promise<{ settled?: Promise }> >(); + /** Conversations `exclusive` holds, which messages wait for. */ + private readonly held = new Map(); + /** Conversations whose last held work left a session not yet retired. */ + private readonly outliving = new Map>(); /** created_at of the newest event each conversation's session was shown. */ private readonly shown = new Map(); /** Conversations that have had a session, whose next new one may have been @@ -206,7 +226,11 @@ export class AgentSessions { return this.shown.get(key) ?? 0; } active(key: string) { - return this.opening.has(key) || !!this.live.get(key)?.session.working; + return ( + this.opening.has(key) || + this.held.has(key) || + !!this.live.get(key)?.session.working + ); } /** The harness session serving a conversation or one-off turn, and the * process it lives in. */ @@ -275,6 +299,16 @@ export class AgentSessions { for (;;) { if (current && !current()) return { ok: true, undelivered: true }; if (this.disposed) return { ...this.stopped, undelivered: true }; + // One that arrives while `exclusive` holds the conversation waits. + const held = this.held.get(key); + if (held) { + try { + await abortable(held.done, signal); + } catch { + return this.late; + } + continue; + } // A message that arrives while its conversation's session opens uses it. const opening = this.opening.get(key); if (opening) { @@ -378,14 +412,86 @@ export class AgentSessions { /** Stops the turn running in conversation `key`, if any, once a session * opening for it has taken its message. */ async interrupt(key: string) { + // Held work is told at once, never waited for. + const held = this.held.get(key); + if (held) { + held.controller.abort(); + this.live.get(key)?.session.interrupt(); + return; + } await this.opening.get(key)?.catch(() => undefined); this.live.get(key)?.session.interrupt(); } /** Includes the model call after a final send has answered the turn. */ async idle(key: string) { + await this.held.get(key)?.done; await this.opening.get(key)?.catch(() => undefined); await this.live.get(key)?.session.idle(); } + + /** Runs `work` with conversation `key` to itself: messages wait for it. + * Its session finishes its work and is closed first, and `work` starts + * only once the harness has acknowledged that close and every earlier + * close of the conversation's sessions, or their process has exited; it + * never starts after a close that was not, nor once interrupted or + * stopped. Saved sessions are never deleted. `work` retires whatever + * session it opens before it settles, or says what it `outlives`. */ + async exclusive( + key: string, + work: (exclusive: Exclusive) => Promise, + ): Promise { + const { label } = this.options; + for (;;) { + const busy = this.opening.get(key) ?? this.held.get(key)?.done; + if (!busy) break; + await busy.catch(() => undefined); + } + // Rechecked and reserved with nothing awaited between. + if (this.disposed) throw new Error(`${label} was stopped`); + const controller = new AbortController(); + let release!: () => void; + const mine: Held = { + controller, + done: new Promise((resolve) => (release = resolve)), + }; + this.held.set(key, mine); + this.options.onChange?.(); + try { + if (this.outliving.has(key)) + throw new Error(`${label} has not retired this conversation's session`); + const live = this.live.get(key); + if (live) { + await live.session.idle(); + if (this.live.get(key) === live) this.drop(key, live); + } + await Promise.all( + [...this.closing] + .filter((closing) => closing.session.conversation === key) + .map((closing) => this.strict(closing)), + ); + if (controller.signal.aborted || this.disposed) + throw new Error(`${label} was stopped`); + const saved = savedConversation(this.options.store?.get(key)); + return await work({ + saved: saved?.sessions?.[0], + keep: (id) => { + if (this.held.get(key) === mine) this.save(key, id, Date.now()); + }, + signal: controller.signal, + outlives: (done) => { + this.outliving.set(key, done); + const retired = () => { + if (this.outliving.get(key) === done) this.outliving.delete(key); + }; + done.then(retired, retired); + }, + }); + } finally { + if (this.held.get(key) === mine) this.held.delete(key); + release(); + this.options.onChange?.(); + } + } /** Waits for the conversation's session to finish its work, then compacts * it or closes it, so the message opens a new one. Without an open session * there is nothing to do. */ @@ -490,7 +596,10 @@ export class AgentSessions { * opening, is closed or its process has exited; rejects if one cannot be. */ async shutdown() { this.dispose(); - await Promise.allSettled(this.opens); + await Promise.allSettled([ + ...this.opens, + ...[...this.held.values()].map((held) => held.done), + ]); const closed = await Promise.allSettled( [...this.closing].map((live) => this.strict(live)), ); @@ -502,6 +611,7 @@ export class AgentSessions { * agents. */ dispose() { this.disposed = true; + for (const held of this.held.values()) held.controller.abort(); for (const [key, live] of this.live) this.drop(key, live); for (const live of this.once.values()) if (live) void this.close(live); } @@ -678,22 +788,24 @@ export class AgentSessions { * has been shown, and restarts its idle countdown. */ private touch(key: string, live: Live, seen?: number) { live.at = Date.now(); - const { store } = this.options; - if (store) { - const { id } = live.session; - const sessions = savedConversation(store.get(key))?.sessions ?? []; - store.set(key, { - at: live.at, - sessions: [id, ...sessions.filter((kept) => kept !== id)].slice( - 0, - SAVED_SESSIONS, - ), - }); - } + this.save(key, live.session.id, live.at); this.countdown(key, live); if (seen !== undefined) this.shown.set(key, Math.max(seen, this.shown.get(key) ?? 0)); } + /** Saves `id` as the conversation's newest session, used `at`. */ + private save(key: string, id: string, at: number) { + const { store } = this.options; + if (!store) return; + const sessions = savedConversation(store.get(key))?.sessions ?? []; + store.set(key, { + at, + sessions: [id, ...sessions.filter((kept) => kept !== id)].slice( + 0, + SAVED_SESSIONS, + ), + }); + } /** Closes the session after IDLE_MS, or waits again while it is busy: a * long turn is not new use. */ private countdown(key: string, live: Live) { diff --git a/src/features/agent-harness/acp/client.test.ts b/src/features/agent-harness/acp/client.test.ts index 3eb8e878b3..532f257577 100644 --- a/src/features/agent-harness/acp/client.test.ts +++ b/src/features/agent-harness/acp/client.test.ts @@ -11,7 +11,12 @@ import type { HostProcess, HostProcessOptions, } from "../../host/service"; -import { AgentSessions, ServerPool, SHARED_POOL } from "./agent-sessions"; +import { + AgentSessions, + type SavedConversation, + ServerPool, + SHARED_POOL, +} from "./agent-sessions"; import { AcpProcess, REQUEST_TIMEOUT_MS, TOOL_SERVER } from "./process"; import { AcpSession } from "./session"; import type { AcpHooks, ToolServer } from "./types"; @@ -56,6 +61,10 @@ function fakeHarness( stopReason?: string; /** Every `session/load` is answered with this error. */ loadError?: { code: number; message: string }; + /** Each `session/close` waits in `closes` to be answered. */ + holdClose?: boolean; + /** Every `session/close` is answered with this error. */ + closeError?: { code: number; message: string }; }> = {}, ) { const started: { @@ -70,6 +79,7 @@ function fakeHarness( steered: string[]; finish(): void; }[] = []; + const closes: (() => void)[] = []; let servers = 0; let rejectPrompts = options.rejectPrompts ?? 0; const serve = async (): Promise => { @@ -137,6 +147,10 @@ function fakeHarness( }); else if (method === "session/new") reply(id, { sessionId: `session-${++sessions}` }); + else if (method === "session/close" && options.closeError) + send({ id, error: options.closeError }); + else if (method === "session/close" && options.holdClose) + closes.push(() => reply(id, {})); else if (method === "session/close") reply(id, {}); else if (method === "session/load" && options.loadError) send({ id, error: options.loadError }); @@ -195,7 +209,7 @@ function fakeHarness( }; const sent = (method: string) => received.filter((message) => message.method === method); - return { spawn, serve, started, received, turns, sent }; + return { spawn, serve, started, received, turns, sent, closes }; } const start = ( @@ -844,3 +858,220 @@ describe("resuming a saved session", () => { expect(harness.sent("session/load")).toHaveLength(1); }); }); + +describe("holding a conversation", () => { + const CLOSES = { + mcpCapabilities: { http: true }, + sessionCapabilities: { close: {} }, + }; + function held( + harness: ReturnType, + acp: AcpProcess, + launch = async () => ({ workspace: "/work", systemPrompt: "" }), + ) { + const saved = new Map(); + const sessions = new AgentSessions({ + label: "Harness", + resolveDirectory: async (path) => path, + tools, + process: { get: async () => acp }, + launch, + choice: () => ({ provider: "", model: "" }), + fingerprint: () => "", + store: { + get: (key) => saved.get(key), + set: (key, value) => saved.set(key, value), + }, + }); + return { sessions, saved, harness }; + } + const deferred = () => { + let resolve!: (value: T) => void; + let reject!: (error: unknown) => void; + const promise = new Promise((yes, no) => { + resolve = yes; + reject = no; + }); + return { promise, resolve, reject }; + }; + + it("starts its work only once the old session's close is acknowledged, and messages wait for it", async () => { + const harness = fakeHarness({ capabilities: CLOSES, holdClose: true }); + const { sessions } = held(harness, await start(harness)); + await sessions.deliver("chat", "hello", 1); + const order: string[] = []; + const holding = sessions.exclusive("chat", async ({ saved }) => { + order.push(`work ${saved}`); + return "done"; + }); + await flush(); + const later = sessions + .deliver("chat", "later", 2) + .then(() => order.push("later")); + await flush(); + expect(harness.sent("session/close")).toHaveLength(1); + expect(order).toEqual([]); + expect(sessions.active("chat")).toBe(true); + harness.closes.shift()?.(); + await expect(holding).resolves.toBe("done"); + await later; + expect(order).toEqual(["work session-1", "later"]); + expect(harness.turns.map((turn) => turn.text)).toEqual(["hello", "later"]); + expect(harness.sent("session/delete")).toEqual([]); + }); + + it("never starts its work after a close the harness failed, deleting nothing", async () => { + const harness = fakeHarness({ + capabilities: CLOSES, + closeError: { code: -32000, message: "close failed" }, + }); + const { sessions, saved } = held(harness, await start(harness)); + await sessions.deliver("chat", "hello", 1); + const before = saved.get("chat"); + const work = vi.fn(async () => undefined); + await expect(sessions.exclusive("chat", work)).rejects.toThrow( + "close failed", + ); + // The failed close stays failed: the next hold is refused too. + await expect(sessions.exclusive("chat", work)).rejects.toThrow( + "close failed", + ); + expect(work).not.toHaveBeenCalled(); + expect(sessions.active("chat")).toBe(false); + expect(saved.get("chat")).toEqual(before); + expect(harness.sent("session/delete")).toEqual([]); + }); + + it("leaves other conversations alone", async () => { + const harness = fakeHarness({ capabilities: CLOSES }); + const { sessions } = held(harness, await start(harness)); + await sessions.deliver("other", "hello", 1); + const working = deferred(); + const holding = sessions.exclusive("chat", () => working.promise); + await flush(); + await expect(sessions.deliver("other", "again", 2)).resolves.toMatchObject({ + ok: true, + }); + expect(harness.sent("session/close")).toEqual([]); + working.resolve(); + await holding; + }); + + it("lets one caller in at a time once an opening session settles", async () => { + const harness = fakeHarness({ capabilities: CLOSES }); + const launching = deferred(); + const { sessions } = held(harness, await start(harness), async () => { + await launching.promise; + return { workspace: "/work", systemPrompt: "" }; + }); + const opening = sessions.deliver("chat", "hello", 1); + await flush(); + const first = deferred(); + const entered: string[] = []; + const a = sessions.exclusive("chat", () => { + entered.push("a"); + return first.promise; + }); + const b = sessions.exclusive("chat", async () => { + entered.push("b"); + }); + launching.resolve(); + await opening; + await flush(); + expect(entered).toEqual(["a"]); + first.resolve(); + await Promise.all([a, b]); + expect(entered).toEqual(["a", "b"]); + }); + + it("tells its work of an interrupt at once, without waiting for it", async () => { + const harness = fakeHarness({ capabilities: CLOSES }); + const { sessions } = held(harness, await start(harness)); + const working = deferred(); + let signal: AbortSignal | undefined; + const holding = sessions.exclusive("chat", (held) => { + signal = held.signal; + return working.promise; + }); + await flush(); + await sessions.interrupt("chat"); + expect(signal?.aborted).toBe(true); + expect(sessions.active("chat")).toBe(true); + working.resolve(); + await holding; + expect(sessions.active("chat")).toBe(false); + }); + + it("never starts work interrupted while it waits for a close", async () => { + const harness = fakeHarness({ capabilities: CLOSES, holdClose: true }); + const { sessions } = held(harness, await start(harness)); + await sessions.deliver("chat", "hello", 1); + const work = vi.fn(async () => undefined); + const holding = sessions.exclusive("chat", work); + await flush(); + await sessions.interrupt("chat"); + harness.closes.shift()?.(); + await expect(holding).rejects.toThrow("Harness was stopped"); + expect(work).not.toHaveBeenCalled(); + expect(sessions.active("chat")).toBe(false); + }); + + it("is aborted by dispose, and shutdown waits for its work", async () => { + const harness = fakeHarness({ capabilities: CLOSES }); + const { sessions } = held(harness, await start(harness)); + const working = deferred(); + let signal: AbortSignal | undefined; + const holding = sessions.exclusive("chat", (held) => { + signal = held.signal; + return working.promise; + }); + await flush(); + let shut = false; + const shutting = sessions.shutdown().then(() => (shut = true)); + await flush(); + expect(signal?.aborted).toBe(true); + expect(shut).toBe(false); + working.resolve(); + await Promise.all([holding, shutting]); + await expect( + sessions.exclusive("chat", async () => undefined), + ).rejects.toThrow("Harness was stopped"); + }); + + it("refuses the conversation while a session its work opened outlives it", async () => { + const harness = fakeHarness({ capabilities: CLOSES }); + const { sessions } = held(harness, await start(harness)); + const exited = deferred(); + await sessions.exclusive("chat", async ({ outlives }) => { + outlives(exited.promise); + }); + const work = vi.fn(async () => undefined); + await expect(sessions.exclusive("chat", work)).rejects.toThrow( + "Harness has not retired this conversation's session", + ); + expect(work).not.toHaveBeenCalled(); + exited.resolve(); + await flush(); + await sessions.exclusive("chat", work); + expect(work).toHaveBeenCalledOnce(); + }); + + it("keeps a session only while it holds the conversation, and releases it on failure", async () => { + const harness = fakeHarness({ capabilities: CLOSES }); + const { sessions, saved } = held(harness, await start(harness)); + let late: ((id: string) => void) | undefined; + await sessions.exclusive("chat", async ({ keep }) => { + keep("saved-1"); + late = keep; + }); + expect(saved.get("chat")?.sessions).toEqual(["saved-1"]); + late?.("saved-2"); + expect(saved.get("chat")?.sessions).toEqual(["saved-1"]); + await expect( + sessions.exclusive("chat", async ({ saved: newest }) => { + throw new Error(`could not load ${newest}`); + }), + ).rejects.toThrow("could not load saved-1"); + expect(sessions.active("chat")).toBe(false); + }); +}); From 532df993896b83ad09c3a7b237e134c595ff9103 Mon Sep 17 00:00:00 2001 From: Philip Azar Date: Fri, 6 Nov 2026 14:43:29 -0600 Subject: [PATCH 08/21] fix(acp): a held conversation has had a session; document hold bounds The next ordinary message after `exclusive` rebuilds with `fresh` like any later session. The doc comment states that an interrupt during the close wait lands within REQUEST_TIMEOUT_MS, that refusals reject with "