--- a/crates/airouter-agentd/src/controller_store.rs +++ b/crates/airouter-agentd/src/controller_store.rs @@ -1,6 +1,30 @@ //! Ownership and revision are checked under the same locks as checkpoint writes. use crate::{store::{Error, Result}, tasks}; -use airouter_agent_protocol::{controller::{self, Binding}, projection, task::{State, Task}, Run}; +use airouter_agent_protocol::{controller::{self, Binding}, projection, task::{Action, State, Task}, Phase, Run}; + +/// Delivery acknowledgements retain their application receipt and lease checks, +/// but only the bound execution can determine a managed task's outcome. Call +/// under the same store lock/transaction used for controller checkpoint writes. +pub fn settlement(run: &Run, action: &Action) -> Result { + if !run.phase.terminal() { + return Err(Error::Conflict); + } + let mut action = action.clone(); + if let Action::Finish { succeeded, error, .. } | Action::Resolve { succeeded, error, .. } = &mut action { + *succeeded = run.phase == Phase::Complete; + *error = if *succeeded { + None + } else { + let mut reason = run.failure.clone().unwrap_or_else(|| match run.phase { + Phase::Cancelled => "managed execution cancelled".into(), + _ => "managed execution failed".into(), + }); + reason.truncate(reason.floor_char_boundary(4096)); + Some(reason) + }; + } + Ok(action) +} #[derive(Clone, Debug)] pub enum Authority { --- a/crates/airouter-agentd/src/task_store.rs +++ b/crates/airouter-agentd/src/task_store.rs @@ -1,10 +1,11 @@ //! Task persistence uses one short transaction per application; models/tools never //! run while a database lock is held. Session keys also fence uncertain outcomes. use crate::{ + controller_store, store::{Error, MemoryStore, PgStore, Result}, tasks, }; -use airouter_agent_protocol::{Event, task::*}; +use airouter_agent_protocol::{Event, Run, task::*}; use async_trait::async_trait; use sqlx::{Postgres, Transaction, types::Json}; use uuid::Uuid; @@ -217,14 +218,16 @@ .iter_mut() .find(|(t, _)| t.application == app && t.request.id == id) .ok_or(Error::NotFound)?; + let mut settlement = None; if matches!(action,Action::Finish{..}|Action::Resolve{..}) { let bindings=self.bindings.lock().await; if let Some(binding)=bindings.values().find(|b|b.application==app && b.task_id==id) { let runs=self.inner.lock().await; - if !runs.get(&binding.run_id).is_some_and(|(run,_,_)|run.phase.terminal()) {return Err(Error::Conflict);} - } - } - if let Some(event) = transition(task, action, now)? { + let (run, _, _) = runs.get(&binding.run_id).ok_or(Error::Conflict)?; + settlement = Some(controller_store::settlement(run, action)?); + } + } + if let Some(event) = transition(task, settlement.as_ref().unwrap_or(action), now)? { events.push(event); } Ok(task.clone()) @@ -435,12 +438,15 @@ tx.commit().await?; return Err(Error::NotFound); }; + let mut settlement = None; if matches!(action,Action::Finish{..}|Action::Resolve{..}) { - let phase:Option=sqlx::query_scalar("SELECT r.snapshot->>'phase' FROM agent_controller_bindings b JOIN agent_runs r ON r.id=b.run_id WHERE b.application=$1 AND b.task_id=$2") + let run:Option>=sqlx::query_scalar("SELECT r.snapshot FROM agent_controller_bindings b JOIN agent_runs r ON r.id=b.run_id WHERE b.application=$1 AND b.task_id=$2") .bind(app).bind(id).fetch_optional(&mut *tx).await?; - if phase.as_deref().is_some_and(|p|!matches!(p,"complete"|"failed"|"cancelled")) {return Err(Error::Conflict);} - } - let result = transition(&mut task, action, now); + if let Some(Json(run)) = run { + settlement = Some(controller_store::settlement(&run, action)?); + } + } + let result = transition(&mut task, settlement.as_ref().unwrap_or(action), now); if let Ok(Some(ref event)) = result { save(&mut tx, &task, event).await?; } --- a/crates/airouter-agentd/tests/controller_fencing.rs +++ b/crates/airouter-agentd/tests/controller_fencing.rs @@ -5,6 +5,9 @@ use serde_json::json; use sqlx::{Executor,postgres::PgPoolOptions}; use uuid::Uuid; + +#[path = "controller_outcomes/mod.rs"] +mod controller_outcomes; async fn verify(store:&dyn Store) { let now=chrono::Utc::now().timestamp(); @@ -42,6 +45,7 @@ let mut changed=binding; changed.lease_id=Uuid::new_v4(); assert!(store.save_owned(interrupted.revision,&cancelled,&event,&Authority::Observation(changed)).await.is_err(),"wrong lease cannot report evidence"); assert_eq!(store.events("fixture",run.id,0).await.unwrap().len(),2); + controller_outcomes::verify(store).await; } #[tokio::test] --- /dev/null +++ b/crates/airouter-agentd/tests/controller_outcomes/mod.rs @@ -0,0 +1,102 @@ +//! A delivery acknowledgement cannot override the bound execution's outcome. +use super::*; + +pub async fn verify(store: &dyn Store) { + for (phase, expired) in [ + (Phase::Failed, false), + (Phase::Failed, true), + (Phase::Complete, false), + (Phase::Cancelled, false), + ] { + let now = chrono::Utc::now().timestamp(); + let tasks = store.tasks().unwrap(); + let task = tasks.submit_task("outcomes", task::Submit { + version: task::VERSION.into(), id: Uuid::new_v4(), queue: "work".into(), + partition: "a".into(), session: Uuid::new_v4().to_string(), kind: "answer".into(), + reference: json!({"controller": controller::VERSION, "host": "fixture"}), + }, now).await.unwrap(); + let lease = tasks.claim_task("outcomes", &task::Claim { + claim_id: Uuid::new_v4(), worker: "host-one".into(), queue: "work".into(), + partition: None, lease_seconds: 60, task_id: Some(task.request.id), + }, 8, now).await.unwrap().unwrap().lease.unwrap(); + tasks.step_task("outcomes", task.request.id, &task::Action::Start { + lease_id: lease.id, worker: lease.worker.clone(), + }, now).await.unwrap(); + let run = core::session::create("outcomes".into(), CreateRun { + api_version: API_VERSION.into(), request_id: Uuid::new_v4(), subject: "fixture".into(), + profile: "fixture".into(), instructions: String::new(), + messages: vec![InputMessage { role: InputRole::User, content: "test".into() }], + servers: vec![], limits: Limits::default(), + }, session::Profile { + version: session::PROFILE_VERSION.into(), executor: session::Executor::NativeCodex, + provider: "openai".into(), model: "fixture".into(), modalities: vec!["text".into()], + }, session::Input { modalities: vec!["text".into()], attachments: vec![] }, now).unwrap(); + let binding = Binding { + version: controller::VERSION.into(), application: "outcomes".into(), + task_id: task.request.id, run_id: run.id, lease_id: lease.id, + host: "fixture".into(), worker: lease.worker.clone(), + profile_digest: run.model_policy_digest.clone(), projection_version: projection::VERSION.into(), + }; + store.create_managed(run.clone(), None, binding.clone(), 8).await.unwrap(); + let (mut active, event) = core::session::start(&run, now).unwrap(); + store.save_owned(run.revision, &active, &event, &Authority::Controller(binding.clone())).await.unwrap(); + let authority = Authority::Observation(binding); + let result = json!({"state": "completed", "command_row_id": 3356}); + // Deliberately invert the execution outcome, as the old Workbench path did. + let mut action = task::Action::Finish { + lease_id: lease.id, worker: lease.worker.clone(), succeeded: phase != Phase::Complete, + result: result.clone(), error: Some("application delivery classification".into()), + }; + let mut finish_at = now; + if expired { + finish_at += 61; + assert_eq!(tasks.read_tasks("outcomes", &[task.request.id], finish_at).await.unwrap()[0].state, task::State::Uncertain); + let (uncertain, event) = core::recover(&active, finish_at).unwrap().unwrap(); + store.save_owned(active.revision, &uncertain, &event, &authority).await.unwrap(); + active = uncertain; + action = task::Action::Resolve { + lease_id: lease.id, succeeded: true, result: result.clone(), + evidence: "application delivery completed; execution receipt saved separately".into(), error: None, + }; + } + assert!(tasks.step_task("outcomes", task.request.id, &action, finish_at).await.is_err(), + "in-flight or uncertain execution cannot be resolved by a caller's success flag"); + if phase == Phase::Cancelled { + let (cancelled, event) = core::apply(&active, Command::Cancel, now).unwrap(); + store.save_owned(active.revision, &cancelled, &event, &authority).await.unwrap(); + active = cancelled; + } + let outcome = session::Outcome { + final_response_id: None, + usage: Usage { usage_unknown: true, coverage: Some(Default::default()), ..Default::default() }, + error: (phase != Phase::Complete).then(|| "executor startup failed".into()), + }; + let (finished, event) = core::session::observe(&active, session::Observation::Finished { outcome: outcome.clone() }, finish_at).unwrap().unwrap(); + store.save_owned(active.revision, &finished, &event, &authority).await.unwrap(); + assert_eq!(finished.phase, phase); + let mut wrong_lease = action.clone(); + match &mut wrong_lease { + task::Action::Finish { lease_id, .. } | task::Action::Resolve { lease_id, .. } => *lease_id = Uuid::new_v4(), + _ => unreachable!(), + } + assert!(tasks.step_task("outcomes", task.request.id, &wrong_lease, finish_at).await.is_err()); + let settled = tasks.step_task("outcomes", task.request.id, &action, finish_at).await.unwrap(); + assert_eq!(settled.state, if phase == Phase::Complete { task::State::Complete } else { task::State::Failed }, + "managed outcome must come from AIRouter's saved run, not the application success flag"); + assert_eq!(settled.error, finished.failure); + assert_eq!(settled.result, Some(result), "application receipt remains opaque and intact"); + let events = tasks.task_events("outcomes", task.request.id, None).await.unwrap(); + assert_eq!(events.last().unwrap().data["state"], json!(settled.state)); + // Late accounting enriches the run revision but cannot change settlement or + // turn a lost acknowledgement into a conflicting second task transition. + let mut accounted = outcome; + accounted.usage.input_tokens = 12; + accounted.usage.coverage.as_mut().unwrap().input_tokens = true; + let (late, event) = core::session::observe(&finished, session::Observation::Finished { outcome: accounted }, finish_at).unwrap().unwrap(); + store.save_owned(finished.revision, &late, &event, &authority).await.unwrap(); + let retry = tasks.step_task("outcomes", task.request.id, &action, finish_at).await.unwrap(); + assert_eq!(retry, settled); + assert_eq!(json!(tasks.task_events("outcomes", task.request.id, None).await.unwrap()), json!(events)); + assert_eq!(store.get("outcomes", run.id).await.unwrap(), late); + } +} --- a/crates/airouter-agentd/TASKS.md +++ b/crates/airouter-agentd/TASKS.md @@ -88,6 +88,22 @@ `failed`; running work expires to `uncertain` and keeps its session occupied. Neither state automatically re-executes the task. Resolve records application observations/evidence and releases an uncertain session. + +For tasks bound to a managed run, AIRouter derives Finish/Resolve success and +error from the saved run under the same store lock as checkpoint writes. A +completed application command is a delivery acknowledgement and cannot turn a +failed execution into success, or a complete execution into failure. Only +`complete` succeeds; `failed` and `cancelled` settle the task as failed, while +the bound run retains its exact phase. In-flight, interrupted and uncertain +runs cannot be settled. Lease/worker identity and Resolve evidence remain +required. `result` remains the opaque application delivery receipt, including +any application-specific `state` field; task `state` and `error` describe the +execution. Unbound application tasks keep their existing outcome contract. + +Late accounting changes do not alter the execution outcome or produce duplicate +task events on acknowledgement retry. This boundary does not automatically +settle tasks when a client disappears; application projection/recovery still +acknowledges delivery. Existing historical classifications are not rewritten. ## Workbench adapter --- a/crates/airouter-agentd/tests/host_session.rs +++ b/crates/airouter-agentd/tests/host_session.rs @@ -22,6 +22,8 @@ #[path = "native_session/mod.rs"] mod native_session; +#[path = "managed_session_recovery/mod.rs"] +mod managed_session_recovery; struct Model; #[async_trait] @@ -69,6 +71,7 @@ client: Client, server: tokio::task::JoinHandle<()>, controller: tokio::task::JoinHandle<()>, + unavailable: Arc, } impl Drop for Fixture { fn drop(&mut self) { @@ -162,14 +165,31 @@ runtime.session_profiles = session_profiles; let runtime = Arc::new(runtime); let controller = tokio::spawn(airouter_agentd::controller::supervise(runtime.clone())); + let unavailable = Arc::new(AtomicUsize::new(0)); + let outage = unavailable.clone(); let server = tokio::spawn(async move { - axum::serve(listener, router(runtime)).await.unwrap(); + let service = router(runtime).layer(axum::middleware::from_fn(move |request: axum::extract::Request, next: axum::middleware::Next| { + let outage = outage.clone(); + async move { + let fail = match outage.load(Ordering::SeqCst) { + 1 => true, + 2 => request.uri().path().ends_with("/observations"), + _ => false, + }; + if fail { + use axum::response::IntoResponse; + axum::http::StatusCode::SERVICE_UNAVAILABLE.into_response() + } else { next.run(request).await } + } + })); + axum::serve(listener, service).await.unwrap(); }); Self { _directory: directory, client: Client::unix(socket.to_str().unwrap(), token.into()).unwrap(), server, controller, + unavailable, } } async fn managed_request(&self) -> controller::Prepare { --- /dev/null +++ b/crates/airouter-agentd/tests/managed_session_recovery/mod.rs @@ -0,0 +1,80 @@ +//! A controller outage must not discard the original host's cleanup receipt. +use super::*; +use airouter_agent_client::{CancellationToken, managed_session, session_stream::Frame}; +use std::time::Duration; + +struct CleanupHost { + unavailable: Arc, + mode: usize, + calls: AtomicUsize, + cleaned: AtomicUsize, + started: Notify, +} + +impl managed_session::Host for CleanupHost { + async fn execute(&self, _: &Run, _: tokio::sync::mpsc::Sender) -> Result { + panic!("managed sessions must use cooperative cleanup") + } + + async fn execute_cancellable(&self, _: &Run, frames: tokio::sync::mpsc::Sender, cancellation: CancellationToken) -> Result { + self.calls.fetch_add(1, Ordering::SeqCst); + self.unavailable.store(self.mode, Ordering::SeqCst); + self.started.notify_one(); + if self.mode == 2 { + frames.send(Frame::ResponseStarted { id: "progress".into() }).await.unwrap(); + } + cancellation.cancelled().await; + // Require the caller to poll the original future after cancellation. + tokio::task::yield_now().await; + self.cleaned.fetch_add(1, Ordering::SeqCst); + if self.mode == 2 { + frames.send(Frame::ResponseCompleted { id: "progress".into(), text: "saved during cleanup".into() }).await.unwrap(); + } + Ok(session::Outcome { + final_response_id: None, + error: Some("original execution stopped after controller outage".into()), + usage: Usage { usage_unknown: true, coverage: Some(Default::default()), ..Default::default() }, + }) + } +} + +async fn outage(mode: usize, user_cancel: bool) { + let fixture = Fixture::new().await; + let prepared = fixture.managed_session().await; + let id = prepared.run.id; + let directory = tempfile::tempdir().unwrap(); + let journal = Journal::new(directory.path()).unwrap(); + let host = CleanupHost { + unavailable: fixture.unavailable.clone(), mode, calls: 0.into(), cleaned: 0.into(), started: Notify::new(), + }; + let cancellation = CancellationToken::new(); + let action = managed_session::serve_cancellable(&fixture.client, prepared, &host, &journal, cancellation.clone(), |_| {}); + let cancel = async { + host.started.notified().await; + if user_cancel { cancellation.cancel(); } + }; + let (result, ()) = tokio::time::timeout(Duration::from_secs(5), async { tokio::join!(action, cancel) }).await.expect("outage did not settle original execution"); + assert!(result.error.is_some()); + assert_eq!(host.cleaned.load(Ordering::SeqCst), 1, "controller error discarded cleanup future"); + let saved = journal.session_observations(&result.run, &result.run.session.as_ref().unwrap().operation).unwrap(); + assert!(matches!(saved.last(), Some(session::Observation::Finished { outcome }) if outcome.error.is_some()), "cleanup outcome was not durable"); + fixture.unavailable.store(0, Ordering::SeqCst); + let prepared = fixture.client.read_host(id).await.unwrap(); + let recovered = managed_session::reconcile_saved(&fixture.client, &prepared, &journal).await.unwrap(); + assert_eq!(recovered.phase, Phase::Failed); + if mode == 2 { + assert_eq!(fixture.client.projection(id).await.unwrap().messages[0].text, "saved during cleanup"); + } + let prepared = fixture.client.read_host(id).await.unwrap(); + managed_session::serve(&fixture.client, prepared, &host, &journal, |_| panic!("recovery emitted preview")).await; + assert_eq!(host.calls.load(Ordering::SeqCst), 1, "recovery repeated execution"); +} + +#[tokio::test] +async fn managed_session_read_outage_preserves_cleanup_receipt() { outage(1, false).await; } + +#[tokio::test] +async fn managed_session_observation_outage_preserves_cleanup_receipt() { outage(2, false).await; } + +#[tokio::test] +async fn managed_session_cancel_ack_outage_preserves_cleanup_receipt() { outage(1, true).await; } --- a/crates/airouter-agent-client/src/process.rs +++ b/crates/airouter-agent-client/src/process.rs @@ -148,6 +148,9 @@ #[derive(Debug)] pub enum ProcessRunError { + /// The owned executor cgroup is empty. External effects and billing can + /// still be unknown, but no local execution remains to race recovery. + CleanupConfirmed(Box), Executor(String), EmptyCommand, MissingPipe(&'static str), @@ -166,6 +169,7 @@ impl fmt::Display for ProcessRunError { fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { match self { + Self::CleanupConfirmed(error) => write!(f, "executor stopped after verified cleanup: {error}"), Self::Executor(message) => f.write_str(message), Self::EmptyCommand => write!(f, "empty provider command"), Self::MissingPipe(name) => write!(f, "missing subprocess {name} pipe"), --- a/crates/airouter-agent-client/src/process/external.rs +++ b/crates/airouter-agent-client/src/process/external.rs @@ -108,7 +108,12 @@ F: FnMut(&str) -> Result<(), ProcessRunError>, { run_cancellable(input, executor, on_line, crate::CancellationToken::new()) - .await? + .await + .map_err(|error| match error { + // Preserve the original error contract for legacy process callers. + ProcessRunError::CleanupConfirmed(error) => *error, + error => error, + })? .ok_or_else(|| invalid("unexpected executor cancellation")) } @@ -196,11 +201,14 @@ let Some(result) = result else { return Ok(None); }; - let outer = result?; + let stopped = |error| ProcessRunError::CleanupConfirmed(Box::new(error)); + let outer = result.map_err(stopped)?; if outer.exit_code != Some(0) { - return Err(invalid("executor exited unsuccessfully")); - } - let (stderr, exit_code) = terminal.ok_or_else(|| invalid("missing terminal result"))??; + return Err(stopped(invalid("executor exited unsuccessfully"))); + } + let (stderr, exit_code) = terminal + .ok_or_else(|| stopped(invalid("missing terminal result")))? + .map_err(stopped)?; Ok(Some(JsonlProcessOutput { stdout_lines, stderr, --- a/crates/airouter-agent-client/src/native/session.rs +++ b/crates/airouter-agent-client/src/native/session.rs @@ -255,15 +255,24 @@ }, cancellation, ) - .await - .map_err(|e| e.to_string())?; + .await; match output { - Some(output) => parser.finish(output.exit_code, &output.stderr), - None => { + Ok(Some(output)) => parser.finish(output.exit_code, &output.stderr), + Ok(None) => { // The process layer has verified local cleanup. Preserve every // observed usage field; absent terminal counters remain unknown. parser.error = Some("native session cancelled after verified local cleanup".into()); parser.finish(None, "") } - } -} + Err(ProcessRunError::CleanupConfirmed(error)) => { + // A rejected stream is an observed failed attempt once the process + // owner has drained every descendant. Journal that outcome through + // the normal session path; never turn it into another execution. + parser.error = Some(error.to_string().chars().take(3000).collect()); + let mut outcome = parser.finish(None, "")?; + outcome.final_response_id = None; + Ok(outcome) + } + Err(error) => Err(error.to_string()), + } +} --- a/crates/airouter-agent-client/src/managed_session.rs +++ b/crates/airouter-agent-client/src/managed_session.rs @@ -271,6 +271,7 @@ let mut open=true; let mut user_cancelled=false; let mut stopping=false; + let mut delivery_failed=false; let stop_deadline=tokio::time::sleep(Duration::from_secs(10)); tokio::pin!(stop_deadline); let outcome=loop { @@ -282,17 +283,23 @@ // stop local work. The Drop guards preserve uncertainty. let cancelled=request_cancel(client,&binding,run.id).await; execution_cancel.cancel(); - run=cancelled?; + match cancelled { + Ok(current)=>run=current, + Err(_)=>delivery_failed=true, + } stopping=true; stop_deadline.as_mut().reset(tokio::time::Instant::now()+Duration::from_secs(10)); }, result=&mut action=>break result?, _=&mut stop_deadline,if stopping=>return Err("session cleanup acknowledgement timed out; reconcile original receipt".into()), - _=tick.tick()=>{ - let current=client.read_host(run.id).await?; + _=tick.tick(),if !stopping=>{ + let current=client.read_host(run.id).await; let now=std::time::SystemTime::now().duration_since(std::time::UNIX_EPOCH).map_err(|_|"invalid clock")?.as_secs() as i64; - if current.binding!=binding || !current.host_action_allowed - || current.run.phase!=Phase::SessionInFlight || now>=run.deadline { + let allowed=current.as_ref().is_ok_and(|current| + current.binding==binding && current.host_action_allowed + && current.run.phase==Phase::SessionInFlight && nowmatch event { - Some(event)=>frame(client,&binding,&mut run,journal,&mut sequences,event,&mut on_event).await?, + Some(event)=>{ + // A broken observation connection must not drop the + // cleanup future and lose its terminal receipt. Keep + // journaling boundaries while the original host stops. + let result=if delivery_failed { + match event { + Frame::ResponseStarted{id}=>journal.record_session_observation(&run,&op,session::Observation::ResponseStarted{id}), + Frame::ResponseCompleted{id,text}=>journal.record_session_observation(&run,&op,session::Observation::ResponseCompleted{id,text}), + _=>Ok(()), + } + } else {frame(client,&binding,&mut run,journal,&mut sequences,event,&mut on_event).await}; + if result.is_err() { + delivery_failed=true; + execution_cancel.cancel(); + if !stopping { + stopping=true; + stop_deadline.as_mut().reset(tokio::time::Instant::now()+Duration::from_secs(10)); + } + } + }, None=>open=false, }, }