--- a/Cargo.lock +++ b/Cargo.lock @@ -72,6 +72,7 @@ "airouter-agent-core", "airouter-agent-protocol", "airouter-client", + "airouter-learning", "airouter-mcp", "airouter-protocol", "async-trait", --- a/crates/airouter-agentd/Cargo.toml +++ b/crates/airouter-agentd/Cargo.toml @@ -11,6 +11,7 @@ prost.workspace = true airouter-agent-core = { path = "../airouter-agent-core" } airouter-agent-protocol = { path = "../airouter-agent-protocol" } +airouter-learning = { path = "../airouter-learning" } airouter-client.workspace = true airouter-protocol.workspace = true async-trait.workspace = true --- a/crates/airouter-agent-client/src/session_service.rs +++ b/crates/airouter-agent-client/src/session_service.rs @@ -92,6 +92,16 @@ self.send(self.http.post("http://session/v1/sessions").json(request)) .await } + /// Cancel a saved submission even when its original admission ACK was lost + /// or admission has not reached the service. This endpoint never executes. + pub async fn cancel_submission(&self, request: &Submit) -> Result { + self.send( + self.http + .post("http://session/v1/sessions/cancel") + .json(request), + ) + .await + } pub async fn status(&self, id: Uuid) -> Result { self.send(self.http.get(format!("http://session/v1/sessions/{id}"))) .await --- a/crates/airouter-agentd/src/session_service/mod.rs +++ b/crates/airouter-agentd/src/session_service/mod.rs @@ -85,8 +85,13 @@ } self.backend.prepare(request, &prepared.run)?; let workspace = Path::new(&request.input.plan.cwd) - .canonicalize().map_err(|_| "native working directory unavailable")?; - airouter_agent_client::journal::Journal::for_workspace(self.store.root(), &workspace, None)?; + .canonicalize() + .map_err(|_| "native working directory unavailable")?; + airouter_agent_client::journal::Journal::for_workspace( + self.store.root(), + &workspace, + None, + )?; Ok(()) } pub async fn submit(self: &Arc, request: Submit) -> Result { @@ -160,6 +165,48 @@ pub async fn cancel(&self, id: Uuid) -> Result { let _admission = self.admission.lock().await; + self.cancel_locked(id).await + } + + /// Serialize cancellation and admission on the same gate. A queued cancel + /// must never cause a previously unsubmitted process to start on recovery. + pub async fn cancel_submission(&self, request: Submit) -> Result { + let _admission = self.admission.lock().await; + let id = request.binding.run_id; + if self.store.exists(id)? { + if self.store.read(id)?.request != request { + return Err("session cancellation changed the original submission".into()); + } + } else { + let prepared = self.client.read_host(id).await?; + if request.version != VERSION + || request.binding != prepared.binding + || request.binding.application != self.options.application + || request.binding.host != self.options.host + || prepared.run.session.is_none() + || prepared.run.phase.terminal() + { + return Err( + "unsubmitted cancellation requires the original live session binding".into(), + ); + } + self.store.insert(&Record { + request, + attempted: false, + status: Status { + run_id: id, + state: State::Accepted, + phase: prepared.run.phase, + cancel_requested: true, + activity_complete: false, + error: None, + }, + })?; + } + self.cancel_locked(id).await + } + + async fn cancel_locked(&self, id: Uuid) -> Result { let record = self .store .update(id, |r| r.status.cancel_requested = true)?; --- a/crates/airouter-agentd/src/session_service/driver.rs +++ b/crates/airouter-agentd/src/session_service/driver.rs @@ -50,6 +50,11 @@ } } let record = self.store.update(id, |record| { + // A never-attempted terminal job has a provably empty activity + // journal. Attempted drivers retain uncertainty after interruption. + if !record.attempted && run.phase.terminal() { + record.status.activity_complete = true; + } record.status.phase = run.phase.clone(); record.status.state = if run.phase.terminal() { State::Finished @@ -82,6 +87,12 @@ self.validate(&saved.request, &prepared)?; let pinned = attachments::pinned(&saved.request, &prepared.run, self.store.root())?; let host = self.backend.prepare(&pinned, &prepared.run)?; + let host = ActivityHost { + inner: &host, + store: &self.store, + id, + complete: std::sync::atomic::AtomicBool::new(false), + }; let journal = Journal::new(self.store.root())?; let binding = &saved.request.binding; let renewal = task::Action::Renew { @@ -103,6 +114,11 @@ &journal, cancellation.clone(), |event| { + // Tool activity is captured at the host boundary, before + // controller transport or semantic terminal errors. + if matches!(event, managed_session::Event::Activity(_)) { + return; + } if activity_error.is_none() { if let Err(error) = self.store.append(id, &event) { activity_error = Some(error); @@ -136,7 +152,8 @@ } else { State::ReconciliationRequired }; - record.status.activity_complete = activity_error.is_none() && result.error.is_none(); + record.status.activity_complete = + activity_error.is_none() && host.complete.load(std::sync::atomic::Ordering::SeqCst); record.status.error = activity_error .or(result.error) .map(|e| e.chars().take(4096).collect()); @@ -146,3 +163,86 @@ Ok(()) } } + +/// Capture application facts before the controller observes response boundaries. +/// A cancelled/failed outcome can have complete activity; a lost cleanup or +/// journal write cannot. The original host future is always polled to cleanup. +struct ActivityHost<'a, H> { + inner: &'a H, + store: &'a super::store::Store, + id: Uuid, + complete: std::sync::atomic::AtomicBool, +} +impl managed_session::Host for ActivityHost<'_, H> { + async fn execute( + &self, + run: &Run, + frames: tokio::sync::mpsc::Sender, + ) -> Result { + self.execute_cancellable(run, frames, CancellationToken::new()) + .await + } + async fn execute_cancellable( + &self, + run: &Run, + frames: tokio::sync::mpsc::Sender, + cancellation: CancellationToken, + ) -> Result { + use airouter_agent_client::{native::events::ProviderStreamEvent, session_stream::Frame}; + let (tx, mut rx) = tokio::sync::mpsc::channel(256); + let execution = self + .inner + .execute_cancellable(run, tx, cancellation.clone()); + tokio::pin!(execution); + let mut error = None; + let relay = async |frame, error: &mut Option| { + let result = match frame { + Frame::Activity { event } => { + if matches!( + event, + ProviderStreamEvent::Text(_) | ProviderStreamEvent::ResponseDelta { .. } + ) { + Err("unidentified native activity text".into()) + } else if error.is_none() { + self.store + .append(self.id, &managed_session::Event::Activity(event)) + } else { + Ok(()) + } + } + frame => frames + .send(frame) + .await + .map_err(|_| "session observation consumer unavailable".to_string()), + }; + if let Err(reason) = result { + if error.is_none() { + *error = Some(reason); + } + cancellation.cancel(); + } + }; + let mut open = true; + let outcome = loop { + tokio::select! { + result=&mut execution=>break result, + frame=rx.recv(),if open=>match frame { + Some(frame)=>relay(frame,&mut error).await, + None=>open=false, + }, + } + }; + while let Ok(frame) = rx.try_recv() { + relay(frame, &mut error).await; + } + self.complete.store( + outcome.is_ok() && error.is_none(), + std::sync::atomic::Ordering::SeqCst, + ); + if let Some(error) = error { + Err(error) + } else { + outcome + } + } +} --- a/crates/airouter-agentd/src/session_service/native.rs +++ b/crates/airouter-agentd/src/session_service/native.rs @@ -56,6 +56,13 @@ input: JsonlProcessInput, executor: Executor, } + +fn result_hooks() -> ToolResultHooks { + ToolResultHooks { + retain: |text, _| text.to_owned(), + mcp_error: airouter_learning::outcome::mcp_error, + } +} impl Backend for NativeBackend { type Host = NativeHost; fn prepare(&self, request: &Submit, run: &Run) -> Result { @@ -110,14 +117,10 @@ // Keep the original tool envelope for application-side projection. The // bounded durable activity journal rejects overflow instead of silently // truncating away an application's handoff receipt or error metadata. - let hooks = ToolResultHooks { - retain: |text, _| text.to_owned(), - mcp_error: |value| value["isError"] == true || value["result"]["isError"] == true, - }; native::session::run_cancellable( self.executor.clone(), self.input.clone(), - hooks, + result_hooks(), |frame| { frames .try_send(frame) @@ -128,3 +131,33 @@ .await } } + +#[cfg(test)] +mod tests { + use super::*; + use native::{codex::CodexStreamParser, events::ProviderStreamEvent}; + use serde_json::json; + + #[test] + fn service_preserves_native_mcp_failures_and_full_receipts() { + let receipt = "saved-application-receipt".repeat(500); + for (body, expected) in [ + (json!({"isError":true,"content":receipt}), 1), + (json!({"is_error":true,"content":receipt}), 1), + (json!({"isError":false,"content":receipt}), 0), + (json!({"isError":"true","content":receipt}), 0), + (json!({"content":"example: {\"isError\":true}"}), 0), + ] { + let mut parser = CodexStreamParser::new(result_hooks()); + let events = parser.ingest_line(&json!({"type":"item.completed","item":{ + "type":"mcp_tool_call","id":"fixture","name":"resource", + "status":"completed","result":body, + }}).to_string()); + let [ProviderStreamEvent::ToolResult { exit_code, output, .. }] = events.as_slice() else { + panic!("expected one MCP tool result"); + }; + assert_eq!(*exit_code, expected); + assert_eq!(serde_json::from_str::(output).unwrap(), body); + } + } +} --- a/crates/airouter-agentd/src/session_service/api.rs +++ b/crates/airouter-agentd/src/session_service/api.rs @@ -21,6 +21,7 @@ Router::new() .route("/healthz", get(|| async { "ok" })) .route("/v1/sessions", post(submit::)) + .route("/v1/sessions/cancel", post(cancel_submission::)) .route("/v1/sessions/{id}", get(status::)) .route("/v1/sessions/{id}/events", get(events::)) .route("/v1/sessions/{id}/cancel", post(cancel::)) @@ -34,6 +35,17 @@ ) -> Result, Error> { service.submit(request).await.map(Json).map_err(rejected) } +async fn cancel_submission( + State(service): State>>, + Json(request): Json, +) -> Result, Error> { + service + .cancel_submission(request) + .await + .map(Json) + .map_err(rejected) +} + async fn status( State(service): State>>, Path(id): Path, --- a/crates/airouter-agentd/tests/session_service/mod.rs +++ b/crates/airouter-agentd/tests/session_service/mod.rs @@ -256,6 +256,7 @@ }) .await; assert_eq!(host.0.cleaned.load(Ordering::SeqCst), 1); + assert!(observer.status(request.binding.run_id).await.unwrap().activity_complete); assert_eq!( observer.status(request.binding.run_id).await.unwrap().phase, Phase::Cancelled @@ -399,3 +400,54 @@ std::fs::set_permissions(root.join("controller.json"), std::fs::Permissions::from_mode(0o600)).unwrap(); tokio::signal::ctrl_c().await.unwrap(); } + +#[tokio::test] +async fn recovered_cancel_before_attempt_has_complete_empty_activity() { + let f = Fixture::new().await; + let prepared = f.managed_session().await; + let dir = tempfile::Builder::new().permissions(std::fs::Permissions::from_mode(0o700)).tempdir().unwrap(); + let request = submit(&prepared,dir.path()); + // Durable accepted admission recovered before the first driver poll. + let path=dir.path().join(format!("{}.json",prepared.run.id)); + std::fs::write(&path,serde_json::to_vec(&json!({ + "request":request,"attempted":false,"status": { + "run_id":prepared.run.id,"state":"accepted","phase":prepared.run.phase, + "cancel_requested":true,"activity_complete":false,"error":null + } + })).unwrap()).unwrap(); + std::fs::set_permissions(&path,std::fs::Permissions::from_mode(0o600)).unwrap(); + let host=adapter(None); + let service=Service::open(f.client.clone(),dir.path(),host.clone(),options()).unwrap(); + service.recover().await.unwrap(); + wait_for(async || service.status(prepared.run.id).unwrap().state==State::Finished).await; + let status=service.status(prepared.run.id).unwrap(); + assert_eq!(status.phase,Phase::Cancelled); + assert!(status.activity_complete); + assert_eq!(host.0.calls.load(Ordering::SeqCst),0); + assert!(service.events(prepared.run.id,0).unwrap().events.is_empty()); + service.shutdown().await; +} + +#[tokio::test] +async fn cancellation_before_admission_never_starts_a_process() { + let f=Fixture::new().await; + let prepared=f.managed_session().await; + let dir=tempfile::Builder::new().permissions(std::fs::Permissions::from_mode(0o700)).tempdir().unwrap(); + let host=adapter(None); + let service=Service::open(f.client.clone(),dir.path(),host.clone(),options()).unwrap(); + let (server,_guard,observer)=socket(service.clone(),dir.path()).await; + let request=submit(&prepared,dir.path()); + let mut foreign=request.clone(); foreign.binding.lease_id=Uuid::new_v4(); + assert!(observer.cancel_submission(&foreign).await.is_err()); + assert!(observer.status(prepared.run.id).await.is_err()); + let cancelled=observer.cancel_submission(&request).await.unwrap(); + assert_eq!(cancelled.phase,Phase::Cancelled); + assert!(cancelled.activity_complete); + assert_eq!(observer.submit(&request).await.unwrap(),cancelled); + assert_eq!(observer.cancel_submission(&request).await.unwrap(),cancelled); + let mut changed=request;changed.input.stdin.push('x'); + assert!(observer.cancel_submission(&changed).await.is_err()); + assert_eq!(host.0.calls.load(Ordering::SeqCst),0); + assert!(observer.events(prepared.run.id,0).await.unwrap().events.is_empty()); + service.shutdown().await; server.abort(); let _=server.await; +} --- a/crates/airouter-agentd/SESSION_SERVICE.md +++ b/crates/airouter-agentd/SESSION_SERVICE.md @@ -26,6 +26,11 @@ Cancellation is explicit (`POST /v1/sessions/RUN/cancel`), and shutdown waits for the original host's cleanup and journal delivery. Cancellation acceptance is not itself proof of cleanup; observe the resulting status and framework run. +For an application outbox whose admission has not been acknowledged, send its +exact saved Submit body to `POST /v1/sessions/cancel`. Admission and cancellation +share the same lock. This endpoint saves a cancellation without starting a +process or reading attachment sources; repeating the original submission returns +the cancelled record. Changed bodies and foreign bindings are rejected. The spool is a private canonical directory outside the model workspace; the service rejects a spool beneath the submitted working directory. A @@ -124,3 +129,10 @@ non-root cgroup and synthetic JSONL child; it needs no production model login and makes no model calls. See project handoffs for exact source hashes and commands; do not repeat already passing checks absent changed behavior. + +Tool activity is captured and fsync'd at the native-host boundary. A cancelled or +failed run can have complete activity once its original host has returned its +cleanup outcome and every queued activity frame has been stored. Semantic run +errors do not imply missing activity. Interrupted cleanup or journal-write errors +still leave activity incomplete. Application consumers must drain the complete +journal and persist their tool receipts before publishing terminal delivery.