--- a/crates/airouter-agentd/Cargo.toml +++ b/crates/airouter-agentd/Cargo.toml @@ -3,9 +3,10 @@ version = "0.2.0" edition = "2024" license = "MIT" +default-run = "airouter-agentd" [dependencies] -airouter-agent-client = { path = "../airouter-agent-client" } +airouter-agent-client = { path = "../airouter-agent-client", features = ["host", "native"] } base64 = "0.22" prost.workspace = true airouter-agent-core = { path = "../airouter-agent-core" } --- a/crates/airouter-agentd/src/lib.rs +++ b/crates/airouter-agentd/src/lib.rs @@ -11,3 +11,5 @@ pub mod communication; pub mod lifecycle; +pub mod session_service; +pub mod socket_runtime; --- a/crates/airouter-agent-client/src/lib.rs +++ b/crates/airouter-agent-client/src/lib.rs @@ -12,6 +12,8 @@ pub mod managed_host; #[cfg(all(feature = "native", feature = "host"))] pub mod managed_session; +#[cfg(all(feature = "native", feature = "host"))] +pub mod session_service; #[cfg(feature = "native")] pub mod native; #[cfg(feature = "process")] --- a/crates/airouter-agent-client/src/managed_session.rs +++ b/crates/airouter-agent-client/src/managed_session.rs @@ -60,7 +60,7 @@ Err("session cancellation acknowledgement unavailable; reconcile original run".into()) } -#[derive(Debug, Clone, PartialEq)] +#[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)] pub enum Event { Preview { source_message_id: String, --- /dev/null +++ b/crates/airouter-agent-client/src/session_service.rs @@ -0,0 +1,108 @@ +//! Same-Unix-account observation IPC for AIRouter's independent native driver. +//! Submission retries reuse the exact body; disconnecting never cancels a job. +use crate::{ + managed_session::Event, + process::JsonlProcessInput, + protocol::{Phase, controller::Binding}, +}; +use serde::{Deserialize, Serialize}; +use uuid::Uuid; + +pub const VERSION: &str = "airouter.session-service.v1"; + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Submit { + pub version: String, + pub binding: Binding, + /// Transitional trusted adapter input. The daemon pins the executable and + /// executor to its registration; model-facing requests cannot supply this. + pub input: JsonlProcessInput, +} + +#[derive(Clone, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum State { + Accepted, + Driving, + Finished, + ReconciliationRequired, +} + +#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)] +pub struct Status { + pub run_id: Uuid, + pub state: State, + pub phase: Phase, + pub cancel_requested: bool, + /// False after an interrupted driver: canonical response/usage recovery can + /// succeed even when application activity receipts need reconciliation. + pub activity_complete: bool, + pub error: Option, +} + +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct Events { + /// Opaque byte cursor. Pass it back unchanged to obtain the next page. + pub next: u64, + pub more: bool, + pub events: Vec, +} + +#[derive(Clone)] +pub struct Client { + http: reqwest::Client, +} +impl Client { + pub fn unix(socket: &std::path::Path) -> Result { + Ok(Self { + http: reqwest::Client::builder() + .unix_socket(socket) + .timeout(std::time::Duration::from_secs(10)) + .build() + .map_err(|e| e.to_string())?, + }) + } + async fn send( + &self, + request: reqwest::RequestBuilder, + ) -> Result { + let response = request.send().await.map_err(|e| e.to_string())?; + if !response.status().is_success() { + return Err(format!( + "session service {}: {}", + response.status(), + response.text().await.unwrap_or_default() + )); + } + response.json().await.map_err(|e| e.to_string()) + } + pub async fn submit(&self, request: &Submit) -> Result { + self.send(self.http.post("http://session/v1/sessions").json(request)) + .await + } + pub async fn status(&self, id: Uuid) -> Result { + self.send(self.http.get(format!("http://session/v1/sessions/{id}"))) + .await + } + pub async fn events(&self, id: Uuid, after: u64) -> Result { + self.send(self.http.get(format!( + "http://session/v1/sessions/{id}/events?after={after}" + ))) + .await + } + pub async fn cancel(&self, id: Uuid) -> Result { + self.send( + self.http + .post(format!("http://session/v1/sessions/{id}/cancel")), + ) + .await + } + pub async fn reconcile(&self, id: Uuid) -> Result { + self.send( + self.http + .post(format!("http://session/v1/sessions/{id}/reconcile")), + ) + .await + } +} --- a/crates/airouter-agentd/tests/host_session.rs +++ b/crates/airouter-agentd/tests/host_session.rs @@ -24,6 +24,8 @@ mod native_session; #[path = "managed_session_recovery/mod.rs"] mod managed_session_recovery; +#[path = "session_service/mod.rs"] +mod session_service; struct Model; #[async_trait] --- /dev/null +++ b/crates/airouter-agentd/tests/session_service/mod.rs @@ -0,0 +1,396 @@ +//! Controller + independent driver + Unix observer. No model calls or app DB. +use super::*; +use airouter_agent_client::{ + CancellationToken, managed_session, + native::events::ProviderStreamEvent, + process::{JsonlProcessInput, ProcessLaunchPlan}, + session_service::{self as wire, State}, + session_stream::Frame, +}; +use airouter_agentd::session_service::{Backend, Options, Service, api}; +use std::{os::unix::fs::PermissionsExt, path::Path, time::Duration}; + +struct Effect { + calls: AtomicUsize, + cleaned: AtomicUsize, + started: Notify, + release: Notify, + outage: Option>, +} +#[derive(Clone)] +struct Adapter(Arc); +impl Backend for Adapter { + type Host = Self; + fn prepare(&self, _: &wire::Submit, _: &Run) -> Result { + Ok(self.clone()) + } +} +impl managed_session::Host for Adapter { + async fn execute( + &self, + _: &Run, + _: tokio::sync::mpsc::Sender, + ) -> Result { + panic!("cleanup-aware host required") + } + async fn execute_cancellable( + &self, + _: &Run, + frames: tokio::sync::mpsc::Sender, + cancellation: CancellationToken, + ) -> Result { + self.0.calls.fetch_add(1, Ordering::SeqCst); + frames + .send(Frame::ResponseStarted { + id: "answer".into(), + }) + .await + .unwrap(); + if let Some(outage) = &self.0.outage { + outage.store(1, Ordering::SeqCst); + } + self.0.started.notify_one(); + let cancelled = tokio::select! { _ = cancellation.cancelled() => true, _ = self.0.release.notified() => false }; + tokio::task::yield_now().await; + self.0.cleaned.fetch_add(1, Ordering::SeqCst); + if !cancelled { + frames.send(Frame::Activity { event: ProviderStreamEvent::ToolResult { tool_use_id: "fixture".into(), tool: "fixture-effect".into(), output: format!("{}{{\"workbench_handoff_receipt\":{{\"cmd_id\":1,\"state\":\"completed\"}}}}", "x".repeat(9000)), exit_code: 0 } }).await.unwrap(); + frames + .send(Frame::ResponseCompleted { + id: "answer".into(), + text: "one durable answer".into(), + }) + .await + .unwrap(); + } + Ok(session::Outcome { + final_response_id: (!cancelled).then(|| "answer".into()), + error: cancelled.then(|| "original host stopped after cleanup".into()), + usage: Usage { + usage_unknown: true, + coverage: Some(Default::default()), + ..Default::default() + }, + }) + } +} +fn adapter(outage: Option>) -> Adapter { + Adapter(Arc::new(Effect { + calls: 0.into(), + cleaned: 0.into(), + started: Notify::new(), + release: Notify::new(), + outage, + })) +} +fn options() -> Options { + Options { + application: "inventory".into(), + host: "stockroom".into(), + max_active: 2, + lease_seconds: 15, + } +} +fn submit(prepared: &controller::Prepared, root: &Path) -> wire::Submit { + wire::Submit { + version: wire::VERSION.into(), + binding: prepared.binding.clone(), + input: JsonlProcessInput::new( + ProcessLaunchPlan::new( + vec!["/fixture/registered-program".into()], + root.to_str().unwrap(), + ), + "admitted input", + ), + } +} +async fn wait_for(mut condition: impl AsyncFnMut() -> bool) { + tokio::time::timeout(Duration::from_secs(8), async { + while !condition().await { + tokio::time::sleep(Duration::from_millis(10)).await; + } + }) + .await + .expect("bounded service condition"); +} +async fn socket( + service: Arc>, + root: &Path, +) -> ( + tokio::task::JoinHandle<()>, + airouter_agentd::socket_runtime::SocketGuard, + wire::Client, +) { + let path = root.join("observer.sock"); + let (listener, guard) = airouter_agentd::socket_runtime::bind(&path, None) + .await + .unwrap(); + let server = tokio::spawn(async move { + axum::serve(listener, api::router(service)).await.unwrap(); + }); + (server, guard, wire::Client::unix(&path).unwrap()) +} + +#[tokio::test] +async fn observer_disconnect_preserves_execution_renewal_and_exact_retry() { + 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(); + assert!( + Service::open(f.client.clone(), dir.path(), host.clone(), options()).is_err(), + "second socket cannot bypass spool ownership" + ); + let request = submit(&prepared, dir.path()); + let (server, guard, observer) = socket(service.clone(), dir.path()).await; + observer.submit(&request).await.unwrap(); + host.0.started.notified().await; + let first_expiry = f + .client + .lookup_tasks(&[request.binding.task_id]) + .await + .unwrap()[0] + .lease + .as_ref() + .unwrap() + .expires_at; + drop(observer); + server.abort(); + let _ = server.await; + drop(guard); + service.recover().await.unwrap(); // A live driver is never adopted twice. + wait_for(async || { + f.client + .lookup_tasks(&[request.binding.task_id]) + .await + .unwrap()[0] + .lease + .as_ref() + .unwrap() + .expires_at + > first_expiry + }) + .await; + assert_eq!(host.0.calls.load(Ordering::SeqCst), 1); + assert_eq!(host.0.cleaned.load(Ordering::SeqCst), 0); + let (server, _guard, observer) = socket(service.clone(), dir.path()).await; + observer.submit(&request).await.unwrap(); + let mut changed = request.clone(); + changed.input.stdin.push_str("changed"); + assert!(observer.submit(&changed).await.is_err()); + host.0.release.notify_one(); + wait_for(async || { + observer.status(request.binding.run_id).await.unwrap().state == State::Finished + }) + .await; + let status = observer.status(request.binding.run_id).await.unwrap(); + assert_eq!(status.phase, Phase::Complete); + assert_eq!(status.error, None); + let events = observer.events(status.run_id, 0).await.unwrap(); + assert!(events.events.iter().any(|e| matches!(e, managed_session::Event::Activity(ProviderStreamEvent::ToolResult { output, .. }) if output.ends_with("\"completed\"}}") && output.len() > 9000))); + assert!( + observer + .events(status.run_id, events.next) + .await + .unwrap() + .events + .is_empty() + ); + assert!(observer.events(status.run_id, 1).await.is_err()); + assert_eq!( + f.client + .projection(status.run_id) + .await + .unwrap() + .messages + .len(), + 1 + ); + assert_eq!( + f.client + .lookup_tasks(&[request.binding.task_id]) + .await + .unwrap()[0] + .state, + task::State::Running, + "execution must not unlock application projection ordering" + ); + observer.submit(&request).await.unwrap(); + observer.reconcile(status.run_id).await.unwrap(); + assert_eq!(host.0.calls.load(Ordering::SeqCst), 1); + service.shutdown().await; + server.abort(); + let _ = server.await; +} + +#[tokio::test] +async fn explicit_cancel_waits_for_original_cleanup_and_fences_retry() { + 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.submit(&foreign).await.is_err()); + assert!(observer.status(request.binding.run_id).await.is_err()); + observer.submit(&request).await.unwrap(); + host.0.started.notified().await; + assert!( + observer + .cancel(request.binding.run_id) + .await + .unwrap() + .cancel_requested + ); + wait_for(async || { + observer.status(request.binding.run_id).await.unwrap().state == State::Finished + }) + .await; + assert_eq!(host.0.cleaned.load(Ordering::SeqCst), 1); + assert_eq!( + observer.status(request.binding.run_id).await.unwrap().phase, + Phase::Cancelled + ); + observer.submit(&request).await.unwrap(); + service.shutdown().await; + assert_eq!(host.0.calls.load(Ordering::SeqCst), 1); + server.abort(); + let _ = server.await; +} + +#[tokio::test] +async fn restart_delivers_saved_cleanup_without_another_execution() { + 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(Some(f.unavailable.clone())); + let service = Service::open(f.client.clone(), dir.path(), host.clone(), options()).unwrap(); + let request = submit(&prepared, dir.path()); + service.submit(request.clone()).await.unwrap(); + host.0.started.notified().await; + wait_for(async || { + service.status(request.binding.run_id).unwrap().state == State::ReconciliationRequired + }) + .await; + assert_eq!(host.0.cleaned.load(Ordering::SeqCst), 1); + service.shutdown().await; + drop(service); + f.unavailable.store(0, Ordering::SeqCst); + let reopened = Service::open(f.client.clone(), dir.path(), host.clone(), options()).unwrap(); + reopened.recover().await.unwrap(); + let status = reopened.status(request.binding.run_id).unwrap(); + assert_eq!(status.state, State::Finished); + assert_eq!(status.phase, Phase::Failed); + reopened.submit(request).await.unwrap(); + assert_eq!(host.0.calls.load(Ordering::SeqCst), 1); + reopened.shutdown().await; +} + +#[tokio::test] +async fn restart_with_unfinished_claim_preserves_uncertainty_and_never_executes() { + 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 journal = Journal::new(dir.path()).unwrap(); + journal + .claim( + &prepared.run, + &prepared.run.session.as_ref().unwrap().operation, + ) + .unwrap(); + let host = adapter(None); + let service = Service::open(f.client.clone(), dir.path(), host.clone(), options()).unwrap(); + let request = submit(&prepared, dir.path()); + service.submit(request.clone()).await.unwrap(); + wait_for(async || { + service.status(request.binding.run_id).unwrap().state == State::ReconciliationRequired + }) + .await; + service.shutdown().await; + drop(service); + let reopened = Service::open(f.client.clone(), dir.path(), host.clone(), options()).unwrap(); + reopened.recover().await.unwrap(); + assert_eq!( + reopened.status(request.binding.run_id).unwrap().state, + State::ReconciliationRequired + ); + assert!( + reopened + .status(request.binding.run_id) + .unwrap() + .error + .is_some() + ); + assert_eq!(host.0.calls.load(Ordering::SeqCst), 0); + reopened.shutdown().await; +} + +#[tokio::test] +async fn native_registration_rejects_changed_program_and_unstaged_attachments() { + use airouter_agent_client::process::external::ExternalExecutor; + use airouter_agentd::session_service::native::{NativeBackend, Registration}; + let f = Fixture::new().await; + let mut prepared = f.managed_session().await; + let dir = tempfile::Builder::new().permissions(std::fs::Permissions::from_mode(0o700)).tempdir().unwrap(); + let mut request = submit(&prepared, dir.path()); + let backend = NativeBackend::new( + [( + prepared.run.request.profile.clone(), + Registration { + program: "/fixture/registered-program".into(), + executor: session::Executor::NativeCodex, + }, + )] + .into(), + ExternalExecutor { + executable: "/fixture/executor".into(), + cgroup_root: "/fixture/cgroup".into(), + }, + ) + .unwrap(); + assert!(backend.prepare(&request, &prepared.run).is_ok()); + request.input.plan.args[0] = "/unregistered/program".into(); + assert!(backend.prepare(&request, &prepared.run).is_err()); + request.input.plan.args[0] = "/fixture/registered-program".into(); + prepared + .run + .session + .as_mut() + .unwrap() + .input + .attachments + .push(session::Attachment { + reference: "file".into(), + media_type: "image/png".into(), + sha256: "a".repeat(64), + bytes: 1, + }); + assert!(backend.prepare(&request, &prepared.run).is_err()); + let alias = dir.path().join("alias"); + std::os::unix::fs::symlink(dir.path(), &alias).unwrap(); + assert!(Service::open(f.client.clone(), &alias, adapter(None), options()).is_err()); + std::fs::set_permissions(dir.path(), std::fs::Permissions::from_mode(0o750)).unwrap(); + assert!(Service::open(f.client.clone(), dir.path(), adapter(None), options()).is_err()); +} + +#[tokio::test] +#[ignore = "explicit private Unix fixture for the standalone sessiond binary"] +async fn serve_native_service_fixture() { + let root = std::path::PathBuf::from(std::env::var_os("AIROUTER_SESSION_FIXTURE_DIR").expect("private fixture output directory")); + assert!(root.is_absolute() && root.is_dir()); + let fixture = Fixture::new().await; + let prepared = fixture.managed_session().await; + let metadata = json!({ + "socket": fixture._directory.path().join("agent.sock"), + "token": "host-fixture-only-application-credential-123456", + "binding": prepared.binding, + "profile": prepared.run.request.profile, + }); + std::fs::write(root.join("controller.json"), serde_json::to_vec(&metadata).unwrap()).unwrap(); + std::fs::set_permissions(root.join("controller.json"), std::fs::Permissions::from_mode(0o600)).unwrap(); + tokio::signal::ctrl_c().await.unwrap(); +} --- /dev/null +++ b/crates/airouter-agentd/src/bin/airouter-sessiond.rs @@ -0,0 +1,97 @@ +//! One AIRouter-owned process per registered application Unix account. +use airouter_agent_client::{ + Client, + process::{containment, external::ExternalExecutor}, +}; +use airouter_agentd::{ + session_service::{ + Options, Service, api, + native::{NativeBackend, Registration}, + }, + socket_runtime, +}; +use clap::Parser; +use serde::Deserialize; +use std::{collections::BTreeMap, os::unix::fs::MetadataExt, path::PathBuf}; + +#[derive(Parser)] +#[command(version, about = "Independent AIRouter native session driver")] +struct Args { + #[arg(long)] + config: PathBuf, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct Config { + application: String, + host: String, + socket: PathBuf, + spool: PathBuf, + framework_socket: PathBuf, + framework_token_file: PathBuf, + executor: PathBuf, + cgroup_root: PathBuf, + max_active: usize, + lease_seconds: u32, + profiles: BTreeMap, +} + +#[tokio::main] +async fn main() -> Result<(), Box> { + tracing_subscriber::fmt() + .with_env_filter(tracing_subscriber::EnvFilter::from_default_env()) + .init(); + let args = Args::parse(); + if nix::unistd::geteuid().is_root() { + return Err("run sessiond under its registered application Unix account".into()); + } + let config: Config = toml::from_str(&std::fs::read_to_string(args.config)?)?; + containment::delegation(&config.cgroup_root)?; + let token_meta = std::fs::symlink_metadata(&config.framework_token_file)?; + if !token_meta.is_file() + || ![0, nix::unistd::geteuid().as_raw()].contains(&token_meta.uid()) + || token_meta.mode() & 0o027 != 0 + { + return Err("framework token must be a protected regular file".into()); + } + let token = std::fs::read_to_string(&config.framework_token_file)?; + let client = Client::unix( + config + .framework_socket + .to_str() + .ok_or("invalid framework socket")?, + token.trim().to_owned(), + )?; + let backend = NativeBackend::new( + config.profiles, + ExternalExecutor { + executable: config.executor, + cgroup_root: config.cgroup_root, + }, + )?; + let service = Service::open( + client, + &config.spool, + backend, + Options { + application: config.application, + host: config.host, + max_active: config.max_active, + lease_seconds: config.lease_seconds, + }, + )?; + let (listener, _socket_guard) = socket_runtime::bind(&config.socket, None).await?; + service.recover().await?; + let stopping = service.clone(); + axum::serve(listener, api::router(service)) + .with_graceful_shutdown(async move { + let mut term = + tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate()) + .expect("SIGTERM registration"); + tokio::select! { _ = tokio::signal::ctrl_c() => {}, _ = term.recv() => {} } + stopping.shutdown().await; + }) + .await?; + Ok(()) +} --- /dev/null +++ b/crates/airouter-agentd/SESSION_SERVICE.md @@ -0,0 +1,109 @@ +# Independent native session service + +`airouter-sessiond --config /etc/airouter/airouter-sessiond.APP.toml` runs an +AIRouter-owned native driver under an application's existing Unix account. +The private socket's directory and socket must be owned by that account, with +modes 0700 and 0600. There is no TCP listener or shared-group admission. The +framework application credential stays in the daemon's protected token file. + +This draft supports registered text Codex and Claude adapters. The registered +executable, native executor and cgroup placement come from daemon configuration. +Prompt/argument/environment assembly remains a trusted application adapter +input during migration. This is not a model-facing arbitrary command API. + +## Ownership and persistence + +The application first obtains the existing framework controller binding. It +persists that binding and the exact service submission before sending it to +`POST /v1/sessions`. The daemon verifies the run/task/lease/profile/application/ +host binding against the framework, saves the request, starts an independent +driver, and acknowledges admission. A client timeout is resolved by repeating +that identical submission. Changed input under the same run ID is rejected. + +The daemon renews the original task lease and owns the native process future. +Dropping a socket observer, request future or frontend does not cancel it. +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. + +The spool is a private canonical directory outside the model workspace. A +process lock prevents another socket/daemon from owning it simultaneously. +Atomic fsync'd records mark an attempt before polling its execution future; +the existing SDK journal also claims the exact session operation. After a +restart, an attempted job can only reconcile saved observations. An unfinished +claim stays `reconciliation_required`; it never launches the model again. +An accepted request that was never attempted can start only with its original +still-valid lease and registration. Do not remove claims to clear a queue. + +`POST /v1/sessions/RUN/reconcile` retries receipt delivery only. Startup attempts +reconciliation; if the framework remains unavailable, restore it and invoke +this endpoint. The daemon does not continuously retry uncertain jobs. + +## Observation API + +The SDK exposes `airouter_agent_client::session_service::Client::unix(path)` +with submit, status, events, cancel and reconcile methods (features `host,native`). +JSON submission schema: `version = airouter.session-service.v1`, the exact +controller `binding`, and shared `JsonlProcessInput` as `input`. Execution +placement is never accepted from that wire input. New requests are limited to +8 MiB and the configured active-job capacity (1–64). + +- `GET /v1/sessions/RUN` returns state, canonical phase, cancellation request, + `activity_complete` and an optional error. It never returns the stored prompt, + environment, credential or launch descriptor. +- `GET /v1/sessions/RUN/events?after=CURSOR` returns a bounded activity page and + opaque next cursor. Start with zero. Persist and reuse the returned cursor; + clients deduplicate using their persisted cursor. This includes previews and + original tool-result envelopes for app-owned receipt projection. +- Completed answers and usage remain in the framework projection. The event + spool is a delivery journal, not another conversation store. +- A complete framework answer can coexist with `activity_complete = false` + after a driver interruption. Reconcile application receipts before declaring + activity delivery complete. Canonical response recovery does not invent + missing tool events. + +Activity records are capped at 1 MiB each and 64 MiB per job. Overflow stops +the original execution and reports an error rather than silently truncating +application receipts. Spool retention/deletion belongs to an explicit delivery +acknowledgement policy; this draft does not automatically delete evidence. + +Task settlement intentionally follows the application's committed projection. +The execution service does not finish the task early and permit a later turn +to race projection of the previous answer. A disconnected app reconnects, +projects saved evidence, then acknowledges or resolves the original task. +Do not claim the full terminal-task ownership migration is finished yet. + +## Installation + +Build `airouter-agentd` (including `airouter-sessiond`) and the shared native +executor with `cargo build --offline --release -p airouter-agent-client +--features executor --bin airouter-native-exec`. The package installs the +daemon in `/usr/sbin` and the executor in `/usr/libexec/airouter`. + +Use `config/session-service.example.toml` with actual registered profile keys, +host, framework socket and token file. Provision the private spool parent and +instance configuration first. The supplied OpenRC template runs an instance +such as `airouter-sessiond.workbench`; set `AIROUTER_SESSION_USER=workbench` +and matching state/run/config paths in its conf.d file. The TOML cgroup root +must be `/sys/fs/cgroup/openrc.INSTANCE`. Its separate service cgroup uses the +existing qualified native containment implementation, so stopping Workbench +does not kill the AIRouter session. Run this service as the app account, never +root. Keep the existing AIRouter Codex UID/home registration and login. + +## Remaining production integration + +Before Workbench cutover, implement durable attachment staging and preserve +exact admitted attachment hashes, move its permission/tool bridge into an +independently available application resource service, and replace its direct +native driver with submit/observe/project calls. Transfer heartbeat and explicit +cancellation together. Do not leave observer-drop cancellation enabled there. +Grok and HTTP adapters are not registered by this draft. Their existing paths +continue until separately migrated. Prompt/persona/model/context policy still +follows the main AI ownership completion sequence. + +Focused qualification uses the existing private Unix framework fixture for +observer loss/renewal, duplicate binding checks, cancellation/cleanup, saved +outcome recovery and unfinished claims. Standalone qualification uses a private +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. --- /dev/null +++ b/config/session-service.example.toml @@ -0,0 +1,18 @@ +# One AIRouter service instance under the application's existing Unix account. +# Copy to a private operator-managed configuration and use matching OpenRC paths. +application = "workbench" +host = "workbench-native" +socket = "/run/airouter-session-workbench/session.sock" +spool = "/var/lib/airouter-native/sessions/workbench" +framework_socket = "/run/airouter-framework/agent.sock" +framework_token_file = "/etc/workbench/airouter-framework.token" +executor = "/usr/libexec/airouter/airouter-native-exec" +cgroup_root = "/sys/fs/cgroup/openrc.airouter-sessiond.workbench" +max_active = 4 +lease_seconds = 120 + +# These keys must be exact profiles registered in the framework. Keep the +# program chosen by the existing trusted adapter until its policy migration. +[profiles.example_native_codex] +program = "/usr/libexec/airouter/airouter-codex" +executor = "native_codex" --- /dev/null +++ b/packaging/openrc/airouter-sessiond @@ -0,0 +1,60 @@ +#!/sbin/openrc-run +# Instance: airouter-sessiond.. Configure the app account and TOML first. +description="AIRouter independent native session owner" +: ${AIROUTER_SESSION_USER:=} +: ${AIROUTER_SESSION_GROUP:=${AIROUTER_SESSION_USER}} +: ${AIROUTER_SESSION_CONFIG:=/etc/airouter/${RC_SVCNAME}.toml} +: ${AIROUTER_SESSION_STATE:=/var/lib/airouter-native/sessions/${AIROUTER_SESSION_USER}} +: ${AIROUTER_SESSION_RUN:=/run/airouter-session-${AIROUTER_SESSION_USER}} + +command="/usr/sbin/airouter-sessiond" +command_args="--config ${AIROUTER_SESSION_CONFIG}" +command_user="${AIROUTER_SESSION_USER}:${AIROUTER_SESSION_GROUP}" +command_background="yes" +pidfile="/run/${RC_SVCNAME}.pid" +output_log="/var/log/${RC_SVCNAME}.log" +error_log="${output_log}" +retry="TERM/40/KILL/3" + +depend() { need airouter-frameworkd; } + +# Reuse the qualified Workbench delegation shape in this independent service's +# cgroup. Workbench stop/restart cannot kill these sibling execution groups. +cgroup2_set_limits() { + local scope="/sys/fs/cgroup/openrc.${RC_SVCNAME}" + mkdir -p "${scope}/service" || return 1 + printf '%s' 0 > "${scope}/service/cgroup.procs" || return 1 +} + +start_pre() { + [ -n "${AIROUTER_SESSION_USER}" ] && [ "${AIROUTER_SESSION_USER}" != root ] || { + eerror "Set AIROUTER_SESSION_USER to the registered app Unix account" + return 1 + } + [ -f "${AIROUTER_SESSION_CONFIG}" ] || return 1 + local scope="/sys/fs/cgroup/openrc.${RC_SVCNAME}" + printf '+cpu +memory +pids' > "${scope}/cgroup.subtree_control" || return 1 + chown "${AIROUTER_SESSION_USER}:${AIROUTER_SESSION_GROUP}" "${scope}" "${scope}/cgroup.procs" || return 1 + chmod 0700 "${scope}" || return 1 + chgrp "${AIROUTER_SESSION_GROUP}" "${scope}/service" "${scope}/service/cgroup.procs" || return 1 + chmod 0750 "${scope}/service" || return 1 + chmod 0640 "${scope}/service/cgroup.procs" || return 1 + checkpath -d -m 0700 -o "${AIROUTER_SESSION_USER}:${AIROUTER_SESSION_GROUP}" "${AIROUTER_SESSION_STATE}" "${AIROUTER_SESSION_RUN}" || return 1 + umask 077 +} + +cgroup2_remove() { + local scope="/sys/fs/cgroup/openrc.${RC_SVCNAME}" child + [ -d "${scope}" ] || return 0 + printf '%s' 0 > /sys/fs/cgroup/cgroup.procs || return 1 + for child in "${scope}"/exec-*/payload "${scope}"/exec-* "${scope}/service"; do + [ ! -d "${child}" ] || rmdir "${child}" 2>/dev/null || : + done + rmdir "${scope}" 2>/dev/null || : +} + +stop_post() { + local scope="/sys/fs/cgroup/openrc.${RC_SVCNAME}" + printf '%s' 0 > /sys/fs/cgroup/cgroup.procs || return 1 + [ ! -f "${scope}/cgroup.kill" ] || printf '%s' 1 > "${scope}/cgroup.kill" +} --- /dev/null +++ b/scripts/qualify-native-session-service.py @@ -0,0 +1,203 @@ +#!/usr/bin/env python3 +"""Run the real session daemon with a private controller and synthetic JSONL child. + +Root provisions only a disposable delegation, then all execution runs as --user. +Uses no production service, database, token, login, workspace or model provider. +""" +import argparse +import hashlib +import http.client +import json +import os +from pathlib import Path +import pwd +import shutil +import signal +import socket +import subprocess +import tempfile +import time +import uuid + + +class UnixHTTP(http.client.HTTPConnection): + def __init__(self, path): + super().__init__('fixture', timeout=5) + self.path = str(path) + + def connect(self): + self.sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM) + self.sock.settimeout(self.timeout) + self.sock.connect(self.path) + + +def request(path, method, route, body=None): + connection = UnixHTTP(path) + try: + connection.request(method, route, None if body is None else json.dumps(body), + {'Content-Type': 'application/json'}) + response = connection.getresponse() + data = response.read() + assert response.status == 200, (response.status, data[:512]) + return json.loads(data) + finally: + connection.close() + + +def wait(condition, seconds=20): + until = time.monotonic() + seconds + while time.monotonic() < until: + if condition(): + return + time.sleep(.05) + raise RuntimeError('bounded fixture wait expired') + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument('--fixture-binary', type=Path, required=True) + parser.add_argument('--session-binary', type=Path, required=True) + parser.add_argument('--executor-binary', type=Path, required=True) + parser.add_argument('--output', type=Path, required=True) + parser.add_argument('--user', default='workbench') + args = parser.parse_args() + assert os.geteuid() == 0, 'root is required only to provision the private qualification delegation' + account = pwd.getpwnam(args.user) + assert account.pw_uid != 0 + args.output.mkdir(parents=True, exist_ok=True) + root = Path(tempfile.mkdtemp(prefix='airouter-session-qualification-', dir='/var/lib/beosrepo')) + scope = Path('/sys/fs/cgroup') / ('airouter-session-qualification-' + uuid.uuid4().hex) + processes, logs = [], [] + receipt = {'passed': False, 'model_calls': 0, 'production_mutations': 0, + 'uid': account.pw_uid, 'cgroup': str(scope), 'binary_sha256': {}} + + def owned(path, mode): + os.chown(path, account.pw_uid, account.pw_gid) + path.chmod(mode) + + def text_file(path, value): + path.write_text(value) + owned(path, 0o600) + + def identity(cgroup): + def setup(): + if cgroup: + (scope / 'service/cgroup.procs').write_text(str(os.getpid())) + os.setgroups(os.getgrouplist(account.pw_name, account.pw_gid)) + os.setgid(account.pw_gid) + os.setuid(account.pw_uid) + return setup + + def launch(argv, log_name, cgroup=False, extra_env=None): + log = (args.output / log_name).open('wb') + logs.append(log) + env = {'PATH': '/usr/bin:/bin', 'HOME': str(root), 'TMPDIR': str(root), + 'RUST_LOG': 'warn', **(extra_env or {})} + process = subprocess.Popen([str(a) for a in argv], cwd=root, env=env, + stdin=subprocess.DEVNULL, stdout=log, stderr=subprocess.STDOUT, + preexec_fn=identity(cgroup)) + processes.append(process) + return process + + def stop(process, sig=signal.SIGTERM): + if process.poll() is None: + process.send_signal(sig) + try: + process.wait(timeout=40) + except subprocess.TimeoutExpired: + process.kill() + process.wait(timeout=5) + raise + + try: + owned(root, 0o700) + for name, source in [('controller', args.fixture_binary), ('sessiond', args.session_binary), ('executor', args.executor_binary)]: + target = root / name + shutil.copyfile(source, target) + owned(target, 0o700) + receipt['binary_sha256'][name] = hashlib.sha256(target.read_bytes()).hexdigest() + for name in ['spool', 'runtime', 'workspace']: + (root / name).mkdir() + owned(root / name, 0o700) + scope.mkdir() + (scope / 'cgroup.subtree_control').write_text('+cpu +memory +pids') + (scope / 'service').mkdir() + owned(scope, 0o700) + owned(scope / 'cgroup.procs', 0o600) + controller = launch([root / 'controller', '--exact', 'session_service::serve_native_service_fixture', '--ignored', '--nocapture'], 'standalone-controller.log', extra_env={'AIROUTER_SESSION_FIXTURE_DIR': str(root)}) + wait(lambda: (root / 'controller.json').exists()) + fixture = json.loads((root / 'controller.json').read_text()) + text_file(root / 'token', fixture['token']) + values = { + 'application': 'inventory', 'host': 'stockroom', + 'socket': str(root / 'runtime/session.sock'), 'spool': str(root / 'spool'), + 'framework_socket': fixture['socket'], 'framework_token_file': str(root / 'token'), + 'executor': str(root / 'executor'), 'cgroup_root': str(scope), + 'max_active': 2, 'lease_seconds': 15, + } + config = '\n'.join(f'{key} = {json.dumps(value)}' for key, value in values.items()) + config += f'\n[profiles.{json.dumps(fixture["profile"])}]\nprogram = "/usr/bin/python3"\nexecutor = "native_codex"\n' + text_file(root / 'service.toml', config) + command = [root / 'sessiond', '--config', root / 'service.toml'] + daemon = launch(command, 'standalone-sessiond.log', cgroup=True) + endpoint = root / 'runtime/session.sock' + wait(endpoint.exists) + script = """ +import json, pathlib, time +def emit(value): print(json.dumps(value), flush=True) +with open('effects', 'a') as effect: effect.write('x') +emit({'type':'item.started','item':{'type':'agent_message','id':'answer'}}) +pathlib.Path('started').touch() +until = time.monotonic() + 30 +while not pathlib.Path('release').exists(): + if time.monotonic() > until: raise RuntimeError('fixture release unavailable') + time.sleep(.05) +emit({'type':'item.completed','item':{'type':'agent_message','id':'answer','text':'independent native answer'}}) +emit({'type':'turn.completed','usage':{'input_tokens':3,'output_tokens':2}}) +""" + body = {'version': 'airouter.session-service.v1', 'binding': fixture['binding'], + 'input': {'plan': {'args': ['/usr/bin/python3', '-I', '-c', script], + 'cwd': str(root / 'workspace'), 'env': {}}, + 'stdin': '', 'stdout_idle_timeout': {'secs': 30, 'nanos': 0}, + 'stdout_idle_retries': 0, 'watchdog': None}} + admitted = request(endpoint, 'POST', '/v1/sessions', body) + run_route = '/v1/sessions/' + admitted['run_id'] + wait(lambda: (root / 'workspace/started').exists()) + # Every request closes its socket. There is now no execution observer. + assert (root / 'workspace/effects').read_text() == 'x' + request(endpoint, 'POST', '/v1/sessions', body) + text_file(root / 'workspace/release', '') + wait(lambda: request(endpoint, 'GET', run_route)['state'] == 'finished') + result = request(endpoint, 'GET', run_route) + assert result['phase'] == 'complete' and result['error'] is None and result['activity_complete'], result + stop(daemon) + assert daemon.returncode == 0 + daemon = launch(command, 'standalone-sessiond-restarted.log', cgroup=True) + wait(endpoint.exists) + recovered = request(endpoint, 'POST', '/v1/sessions', body) + assert recovered == result + assert (root / 'workspace/effects').read_text() == 'x', 'native effect repeated' + stop(daemon) + stop(controller, signal.SIGINT) + assert daemon.returncode == 0 and controller.returncode == 0 + receipt.update(passed=True, status=result, executions=1, restart_preserved_result=True, + observer_disconnect_preserved_execution=True) + finally: + for process in reversed(processes): + stop(process) + for log in logs: + log.close() + if scope.exists(): + (scope / 'cgroup.kill').write_text('1') + wait(lambda: 'populated 0' in (scope / 'cgroup.events').read_text()) + for directory in sorted([p for p in scope.rglob('*') if p.is_dir()], key=lambda p: len(p.parts), reverse=True): + directory.rmdir() + scope.rmdir() + shutil.rmtree(root) + receipt['temporary_state_removed'] = not root.exists() and not scope.exists() + (args.output / 'standalone-qualification.json').write_text(json.dumps(receipt, indent=2) + '\n') + print(json.dumps(receipt, indent=2)) + + +if __name__ == '__main__': + main() --- /dev/null +++ b/crates/airouter-agentd/src/session_service/api.rs @@ -0,0 +1,66 @@ +use super::{Backend, Service}; +use airouter_agent_client::session_service::{Events, Status, Submit}; +use axum::{ + Json, Router, + extract::{DefaultBodyLimit, Path, Query, State}, + http::StatusCode, + routing::{get, post}, +}; +use serde::Deserialize; +use std::sync::Arc; +use uuid::Uuid; + +type Error = (StatusCode, String); +fn rejected(error: String) -> Error { + (StatusCode::CONFLICT, error) +} + +/// Bind only with socket_runtime::bind(path, None): filesystem ownership is the +/// app identity. This API is not an unauthenticated TCP or shared-group service. +pub fn router(service: Arc>) -> Router { + Router::new() + .route("/healthz", get(|| async { "ok" })) + .route("/v1/sessions", post(submit::)) + .route("/v1/sessions/{id}", get(status::)) + .route("/v1/sessions/{id}/events", get(events::)) + .route("/v1/sessions/{id}/cancel", post(cancel::)) + .route("/v1/sessions/{id}/reconcile", post(reconcile::)) + .layer(DefaultBodyLimit::max(8 * 1024 * 1024)) + .with_state(service) +} +async fn submit( + State(service): State>>, + Json(request): Json, +) -> Result, Error> { + service.submit(request).await.map(Json).map_err(rejected) +} +async fn status( + State(service): State>>, + Path(id): Path, +) -> Result, Error> { + service.status(id).map(Json).map_err(rejected) +} +#[derive(Deserialize)] +struct Cursor { + #[serde(default)] + after: u64, +} +async fn events( + State(service): State>>, + Path(id): Path, + Query(cursor): Query, +) -> Result, Error> { + service.events(id, cursor.after).map(Json).map_err(rejected) +} +async fn cancel( + State(service): State>>, + Path(id): Path, +) -> Result, Error> { + service.cancel(id).await.map(Json).map_err(rejected) +} +async fn reconcile( + State(service): State>>, + Path(id): Path, +) -> Result, Error> { + service.reconcile(id).await.map(Json).map_err(rejected) +} --- /dev/null +++ b/crates/airouter-agentd/src/session_service/driver.rs @@ -0,0 +1,147 @@ +use super::*; +use airouter_agent_client::journal::Journal; +use airouter_agent_protocol::{Command, task}; +use std::time::Duration; + +impl Service { + pub(super) async fn cancel_platform(&self, record: &Record) -> Result<(), String> { + for _ in 0..3 { + let current = self.client.read_host(record.status.run_id).await?; + if current.binding != record.request.binding { + return Err("session ownership changed".into()); + } + if current.run.phase.terminal() + || current.run.failure.as_deref() == Some("cancel_requested") + { + return Ok(()); + } + if self + .client + .step(¤t.run, Command::Cancel) + .await + .is_ok() + { + return Ok(()); + } + } + Err("session cancellation needs original-run reconciliation".into()) + } + pub(super) async fn reconcile_record(&self, id: Uuid) -> Result { + let saved = self.store.read(id)?; + if saved.status.cancel_requested { + self.cancel_platform(&saved).await?; + } + let prepared = self.client.read_host(id).await?; + if prepared.binding != saved.request.binding { + return Err("session ownership changed".into()); + } + let mut run = prepared.run.clone(); + let mut error = None; + if !run.phase.terminal() { + let journal = Journal::new(self.store.root())?; + match managed_session::reconcile_saved(&self.client, &prepared, &journal).await { + Ok(current) => run = current, + Err(_) => { + error = Some( + "original execution has no deliverable outcome; inspect saved evidence" + .into(), + ) + } + } + } + let record = self.store.update(id, |record| { + record.status.phase = run.phase.clone(); + record.status.state = if run.phase.terminal() { + State::Finished + } else { + State::ReconciliationRequired + }; + record.status.error = run.failure.or(error).or_else(|| { + (!run.phase.terminal()) + .then(|| "original execution remains uncertain; no automatic replay".into()) + }); + })?; + Ok(record.status) + } + pub(super) async fn drive( + &self, + id: Uuid, + cancellation: CancellationToken, + ) -> Result<(), String> { + let saved = self.store.read(id)?; + if saved.attempted { + self.reconcile_record(id).await?; + return Ok(()); + } + if saved.status.cancel_requested || cancellation.is_cancelled() { + self.cancel_platform(&saved).await?; + self.reconcile_record(id).await?; + return Ok(()); + } + let prepared = self.client.read_host(id).await?; + self.validate(&saved.request, &prepared)?; + let host = self.backend.prepare(&saved.request, &prepared.run)?; + let journal = Journal::new(self.store.root())?; + let binding = &saved.request.binding; + let renewal = task::Action::Renew { + lease_id: binding.lease_id, + worker: binding.worker.clone(), + lease_seconds: self.options.lease_seconds, + }; + self.client.step_task(binding.task_id, &renewal).await?; + self.store.update(id, |record| { + record.attempted = true; + record.status.state = State::Driving; + })?; + let mut activity_error = None; + let result = { + let action = managed_session::serve_cancellable( + &self.client, + prepared, + &host, + &journal, + cancellation.clone(), + |event| { + if activity_error.is_none() { + if let Err(error) = self.store.append(id, &event) { + activity_error = Some(error); + cancellation.cancel(); + } + } + }, + ); + tokio::pin!(action); + let period = Duration::from_secs(u64::from(self.options.lease_seconds / 3)); + let mut heartbeat = + tokio::time::interval_at(tokio::time::Instant::now() + period, period); + heartbeat.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + let mut renewal_failed = false; + loop { + tokio::select! { + result = &mut action => break result, + _ = heartbeat.tick(), if !renewal_failed => { + if self.client.step_task(binding.task_id, &renewal).await.is_err() { + renewal_failed = true; + cancellation.cancel(); + } + } + } + } + }; + self.store.update(id, |record| { + record.status.phase = result.run.phase.clone(); + record.status.state = if result.run.phase.terminal() { + State::Finished + } else { + State::ReconciliationRequired + }; + record.status.activity_complete = activity_error.is_none() && result.error.is_none(); + record.status.error = activity_error + .or(result.error) + .map(|e| e.chars().take(4096).collect()); + })?; + // Application projection still owns its delivery acknowledgement. Do not + // unlock task serialization before that durable projection has committed. + Ok(()) + } +} --- /dev/null +++ b/crates/airouter-agentd/src/session_service/mod.rs @@ -0,0 +1,223 @@ +//! Independent native session ownership. Socket observers never own job futures. +pub mod api; +mod driver; +pub mod native; +mod store; + +use airouter_agent_client::{ + CancellationToken, Client, managed_session, + session_service::{State, Status, Submit, VERSION}, +}; +use airouter_agent_protocol::{Run, controller::Prepared}; +use std::{ + collections::HashMap, + path::Path, + sync::{ + Arc, Mutex, + atomic::{AtomicBool, Ordering}, + }, +}; +use store::{Record, Store}; +use tokio::sync::{Mutex as AsyncMutex, Notify}; +use uuid::Uuid; + +/// Operator-selected executor. There is no callback into the submitting client. +pub trait Backend: Send + Sync + 'static { + type Host: managed_session::Host + Send + Sync + 'static; + fn prepare(&self, request: &Submit, run: &Run) -> Result; +} + +#[derive(Clone)] +pub struct Options { + pub application: String, + pub host: String, + pub max_active: usize, + pub lease_seconds: u32, +} + +pub struct Service { + client: Client, + store: Store, + backend: B, + options: Options, + admission: AsyncMutex<()>, + active: Mutex>, + changed: Notify, + stopping: AtomicBool, +} + +impl Service { + pub fn open( + client: Client, + root: &Path, + backend: B, + options: Options, + ) -> Result, String> { + if options.application.is_empty() + || options.host.is_empty() + || !(1..=64).contains(&options.max_active) + || !(15..=300).contains(&options.lease_seconds) + { + return Err("invalid session service registration or capacity".into()); + } + Ok(Arc::new(Self { + client, + store: Store::open(root)?, + backend, + options, + admission: AsyncMutex::new(()), + active: Mutex::new(HashMap::new()), + changed: Notify::new(), + stopping: AtomicBool::new(false), + })) + } + fn validate(&self, request: &Submit, prepared: &Prepared) -> Result<(), String> { + if request.version != VERSION + || request.binding != prepared.binding + || request.binding.application != self.options.application + || request.binding.host != self.options.host + { + return Err("session request does not match the registered controller binding".into()); + } + if !prepared.host_action_allowed || prepared.run.phase.terminal() { + return Err("session lease is not available for execution".into()); + } + self.backend.prepare(request, &prepared.run)?; + Ok(()) + } + pub async fn submit(self: &Arc, request: Submit) -> Result { + let _admission = self.admission.lock().await; + let id = request.binding.run_id; + if self.store.exists(id)? { + let saved = self.store.read(id)?; + if saved.request != request { + return Err("session submission changed; use the original body".into()); + } + return Ok(saved.status); + } + if self.stopping.load(Ordering::SeqCst) + || self.active.lock().unwrap().len() >= self.options.max_active + { + return Err("session service has no admission capacity".into()); + } + let prepared = self.client.read_host(id).await?; + self.validate(&request, &prepared)?; + let record = Record { + request, + attempted: false, + status: Status { + run_id: id, + state: State::Accepted, + phase: prepared.run.phase, + cancel_requested: false, + activity_complete: false, + error: None, + }, + }; + self.store.insert(&record)?; + self.start(id); + Ok(record.status) + } + fn start(self: &Arc, id: Uuid) { + let cancellation = CancellationToken::new(); + self.active.lock().unwrap().insert(id, cancellation.clone()); + let service = self.clone(); + tokio::spawn(async move { + let driver = service.clone(); + let result = tokio::spawn(async move { driver.drive(id, cancellation).await }) + .await + .unwrap_or_else(|_| { + Err("native session driver interrupted; reconcile original evidence".into()) + }); + if let Err(error) = result { + let result = service.store.update(id, |record| { + record.status.state = State::ReconciliationRequired; + record.status.error = Some(error.chars().take(4096).collect()); + }); + if result.is_err() { + tracing::error!(%id, "session status could not be saved; original record remains fenced"); + } + } + service.active.lock().unwrap().remove(&id); + service.changed.notify_waiters(); + }); + } + pub fn status(&self, id: Uuid) -> Result { + Ok(self.store.read(id)?.status) + } + pub fn events( + &self, + id: Uuid, + after: u64, + ) -> Result { + self.store.events(id, after) + } + + pub async fn cancel(&self, id: Uuid) -> Result { + let _admission = self.admission.lock().await; + let record = self + .store + .update(id, |r| r.status.cancel_requested = true)?; + let active = self.active.lock().unwrap().get(&id).cloned(); + if let Some(cancellation) = active { + cancellation.cancel(); + return Ok(record.status); + } + self.cancel_platform(&record).await?; + self.reconcile_record(id).await + } + pub async fn reconcile(&self, id: Uuid) -> Result { + let _admission = self.admission.lock().await; + if self.active.lock().unwrap().contains_key(&id) { + return self.status(id); + } + self.reconcile_record(id).await + } + /// Resume only durably accepted, never-attempted requests. Any possible + /// effect is restricted to receipt delivery, including an empty journal. + pub async fn recover(self: &Arc) -> Result<(), String> { + let _admission = self.admission.lock().await; + for id in self.store.ids()? { + if self.active.lock().unwrap().contains_key(&id) { + continue; + } + let record = self.store.read(id)?; + if record.status.state == State::Finished { + continue; + } + if record.attempted || record.status.cancel_requested { + if let Err(error) = self.reconcile_record(id).await { + self.store.update(id, |r| { + r.status.state = State::ReconciliationRequired; + r.status.error = Some(error.chars().take(4096).collect()); + })?; + } + } else if self.active.lock().unwrap().len() < self.options.max_active + && !self.stopping.load(Ordering::SeqCst) + { + self.start(id); + } + } + Ok(()) + } + /// Operator service stop, distinct from an observer disconnect. Wait for the + /// original host cleanup future and saved receipt before dropping the spool. + pub async fn shutdown(&self) { + { + let _admission = self.admission.lock().await; + self.stopping.store(true, Ordering::SeqCst); + for cancellation in self.active.lock().unwrap().values() { + cancellation.cancel(); + } + } + loop { + let changed = self.changed.notified(); + tokio::pin!(changed); + changed.as_mut().enable(); + if self.active.lock().unwrap().is_empty() { + break; + } + changed.await; + } + } +} --- /dev/null +++ b/crates/airouter-agentd/src/session_service/native.rs @@ -0,0 +1,134 @@ +//! Registered native adapters. Containment placement is configured by the daemon. +use super::Backend; +use airouter_agent_client::{ + CancellationToken, + managed_session::Host, + native::{self, ToolResultHooks}, + process::{JsonlProcessInput, ProcessWatchdogRegistration, external::ExternalExecutor}, + session_service::Submit, + session_stream::Frame, +}; +use airouter_agent_protocol::{ + Run, + session::{Executor, Outcome}, +}; +use serde::{Deserialize, Serialize}; +use std::{collections::BTreeMap, path::PathBuf}; +use tokio::sync::mpsc; + +#[derive(Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Registration { + pub program: PathBuf, + pub executor: Executor, +} + +pub struct NativeBackend { + registrations: BTreeMap, + executor: ExternalExecutor, +} +impl NativeBackend { + pub fn new( + registrations: BTreeMap, + executor: ExternalExecutor, + ) -> Result { + if registrations.is_empty() { + return Err("native session registrations are required".into()); + } + for registration in registrations.values() { + if !registration.program.is_absolute() + || !matches!( + registration.executor, + Executor::NativeCodex | Executor::NativeClaude + ) + { + return Err("register an absolute Codex/Claude program".into()); + } + } + Ok(Self { + registrations, + executor, + }) + } +} + +pub struct NativeHost { + input: JsonlProcessInput, + executor: Executor, +} +impl Backend for NativeBackend { + type Host = NativeHost; + fn prepare(&self, request: &Submit, run: &Run) -> Result { + let session = run + .session + .as_ref() + .ok_or("native session checkpoint missing")?; + let registration = self + .registrations + .get(&run.request.profile) + .ok_or("native profile is not registered with this service")?; + if registration.executor != session.profile.executor + || request.input.plan.args.first().map(PathBuf::from).as_ref() + != Some(®istration.program) + { + return Err("native executable or executor differs from its registration".into()); + } + if !session.input.attachments.is_empty() + || session.input.modalities.iter().any(|m| m != "text") + { + return Err("durable attachment staging is not yet supported by this service".into()); + } + if !std::path::Path::new(&request.input.plan.cwd).is_absolute() + || !std::path::Path::new(&request.input.plan.cwd).is_dir() + { + return Err("native working directory must be an existing absolute directory".into()); + } + if request.input.stdout_idle_retries != 0 || request.input.executor.is_some() { + return Err("native execution placement and retries belong to the service".into()); + } + let mut input = request.input.clone(); + input.executor = Some(self.executor.clone()); + input.watchdog = Some(ProcessWatchdogRegistration::new( + run.id.to_string(), + session.profile.provider.clone(), + session.profile.model.clone(), + )); + Ok(NativeHost { + input, + executor: registration.executor.clone(), + }) + } +} + +impl Host for NativeHost { + async fn execute(&self, run: &Run, frames: mpsc::Sender) -> Result { + self.execute_cancellable(run, frames, CancellationToken::new()) + .await + } + async fn execute_cancellable( + &self, + _: &Run, + frames: mpsc::Sender, + cancellation: CancellationToken, + ) -> Result { + // 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, + |frame| { + frames + .try_send(frame) + .map_err(|_| "native frame delivery unavailable".into()) + }, + cancellation, + ) + .await + } +} --- /dev/null +++ b/crates/airouter-agentd/src/session_service/store.rs @@ -0,0 +1,226 @@ +//! A single writer owns each private spool, even across different socket names. +use airouter_agent_client::{ + managed_session::Event, + session_service::{Events, Status, Submit}, +}; +use serde::{Deserialize, Serialize}; +use std::{ + fs::{self, File, OpenOptions}, + io::{Read, Seek, SeekFrom, Write}, + os::unix::fs::{MetadataExt, OpenOptionsExt}, + path::{Path, PathBuf}, + sync::Mutex, +}; +use uuid::Uuid; + +const MAX_RECORD: u64 = 16 * 1024 * 1024; +const MAX_EVENT: usize = 1024 * 1024; +const MAX_EVENTS: u64 = 64 * 1024 * 1024; + +#[derive(Clone, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub(super) struct Record { + pub request: Submit, + pub status: Status, + /// Committed before polling any execution future. Recovery cannot execute + /// again even if the host journal was not reached before the daemon died. + pub attempted: bool, +} + +pub(super) struct Store { + root: PathBuf, + _lock: File, + io: Mutex<()>, +} + +pub(super) fn private_file(path: &Path, write: bool, create: bool) -> Result { + let file = OpenOptions::new() + .read(true) + .write(write) + .create(create) + .truncate(false) + .mode(0o600) + .custom_flags(nix::libc::O_NOFOLLOW | nix::libc::O_NONBLOCK | nix::libc::O_CLOEXEC) + .open(path) + .map_err(|e| e.to_string())?; + let meta = file.metadata().map_err(|e| e.to_string())?; + if !meta.is_file() + || meta.uid() != nix::unistd::geteuid().as_raw() + || meta.nlink() != 1 + || meta.mode() & 0o7777 != 0o600 + { + return Err("session state must be a private regular file owned by this account".into()); + } + Ok(file) +} + +impl Store { + pub fn open(root: &Path) -> Result { + let meta = fs::symlink_metadata(root).map_err(|e| e.to_string())?; + if !root.is_absolute() + || !meta.is_dir() + || meta.uid() != nix::unistd::geteuid().as_raw() + || meta.mode() & 0o7777 != 0o700 + || fs::canonicalize(root).map_err(|e| e.to_string())? != root + { + return Err("session spool must be a canonical, private owned directory (0700)".into()); + } + let lock = private_file(&root.join("service.lock"), true, true)?; + lock.try_lock() + .map_err(|_| "session spool is already owned by another driver")?; + Ok(Self { + root: root.into(), + _lock: lock, + io: Mutex::new(()), + }) + } + pub fn root(&self) -> &Path { + &self.root + } + fn path(&self, id: Uuid) -> PathBuf { + self.root.join(format!("{id}.json")) + } + fn read_inner(&self, id: Uuid) -> Result { + let file = private_file(&self.path(id), false, false)?; + if file.metadata().map_err(|e| e.to_string())?.len() > MAX_RECORD { + return Err("session record exceeds limit".into()); + } + let record: Record = + serde_json::from_reader(file.take(MAX_RECORD + 1)).map_err(|e| e.to_string())?; + if record.request.binding.run_id != id || record.status.run_id != id { + return Err("session record identity mismatch".into()); + } + Ok(record) + } + pub fn read(&self, id: Uuid) -> Result { + let _guard = self.io.lock().unwrap(); + self.read_inner(id) + } + pub fn exists(&self, id: Uuid) -> Result { + match fs::symlink_metadata(self.path(id)) { + Ok(_) => Ok(true), + Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(false), + Err(e) => Err(e.to_string()), + } + } + fn save_inner(&self, record: &Record) -> Result<(), String> { + let bytes = serde_json::to_vec(record).map_err(|e| e.to_string())?; + if bytes.len() as u64 > MAX_RECORD { + return Err("session record exceeds limit".into()); + } + let temp = self.root.join(format!(".write-{}", Uuid::new_v4())); + let result = (|| { + let mut file = OpenOptions::new() + .write(true) + .create_new(true) + .mode(0o600) + .open(&temp) + .map_err(|e| e.to_string())?; + file.write_all(&bytes) + .and_then(|_| file.sync_all()) + .map_err(|e| e.to_string())?; + fs::rename(&temp, self.path(record.status.run_id)).map_err(|e| e.to_string())?; + File::open(&self.root) + .and_then(|f| f.sync_all()) + .map_err(|e| e.to_string()) + })(); + if result.is_err() { + let _ = fs::remove_file(temp); + } + result + } + pub fn insert(&self, record: &Record) -> Result<(), String> { + let _guard = self.io.lock().unwrap(); + if self.exists(record.status.run_id)? { + return Err("session record already exists".into()); + } + self.save_inner(record) + } + pub fn update(&self, id: Uuid, change: impl FnOnce(&mut Record)) -> Result { + let _guard = self.io.lock().unwrap(); + let mut record = self.read_inner(id)?; + change(&mut record); + self.save_inner(&record)?; + Ok(record) + } + pub fn ids(&self) -> Result, String> { + let mut ids = Vec::new(); + for entry in fs::read_dir(&self.root).map_err(|e| e.to_string())? { + let entry = entry.map_err(|e| e.to_string())?; + let name = entry.file_name(); + if let Some(id) = name.to_str().and_then(|s| s.strip_suffix(".json")) { + ids.push(Uuid::parse_str(id).map_err(|_| "unrecognized session record")?); + } + } + Ok(ids) + } + pub fn append(&self, id: Uuid, event: &Event) -> Result<(), String> { + let mut bytes = serde_json::to_vec(event).map_err(|e| e.to_string())?; + if bytes.len() > MAX_EVENT { + return Err("session activity exceeds durable event limit".into()); + } + bytes.push(b'\n'); + let _guard = self.io.lock().unwrap(); + let mut file = private_file(&self.root.join(format!("{id}.events")), true, true)?; + let end = file.seek(SeekFrom::End(0)).map_err(|e| e.to_string())?; + if end + bytes.len() as u64 > MAX_EVENTS { + return Err("session activity journal is full".into()); + } + file.write_all(&bytes) + .and_then(|_| file.sync_all()) + .map_err(|e| e.to_string())?; + File::open(&self.root) + .and_then(|f| f.sync_all()) + .map_err(|e| e.to_string()) + } + pub fn events(&self, id: Uuid, after: u64) -> Result { + let _guard = self.io.lock().unwrap(); + self.read_inner(id)?; + let path = self.root.join(format!("{id}.events")); + let mut file = match fs::symlink_metadata(&path) { + Err(e) if e.kind() == std::io::ErrorKind::NotFound && after == 0 => { + return Ok(Events { + next: 0, + more: false, + events: vec![], + }); + } + _ => private_file(&path, false, false)?, + }; + let length = file.metadata().map_err(|e| e.to_string())?.len(); + if after > length || length > MAX_EVENTS { + return Err("invalid session event cursor or size".into()); + } + if after > 0 { + file.seek(SeekFrom::Start(after - 1)) + .map_err(|e| e.to_string())?; + let mut previous = [0]; + file.read_exact(&mut previous).map_err(|e| e.to_string())?; + if previous != [b'\n'] { + return Err("invalid session event cursor".into()); + } + } + let mut bytes = Vec::new(); + file.take((2 * MAX_EVENT + 2) as u64) + .read_to_end(&mut bytes) + .map_err(|e| e.to_string())?; + let mut page = Events { + next: after, + more: false, + events: vec![], + }; + for line in bytes.split_inclusive(|b| *b == b'\n').take(128) { + if line.last() != Some(&b'\n') { + if page.next == after || after + bytes.len() as u64 == length { + return Err("incomplete session activity; inspect saved evidence".into()); + } + break; + } + page.events + .push(serde_json::from_slice(line).map_err(|e| e.to_string())?); + page.next += line.len() as u64; + } + page.more = page.next < length; + Ok(page) + } +}