path: root/net-misc/airouter/files/airouter-session-service.patch

--- 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<String>,
+}
+
+#[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<Event>,
+}
+
+#[derive(Clone)]
+pub struct Client {
+    http: reqwest::Client,
+}
+impl Client {
+    pub fn unix(socket: &std::path::Path) -> Result<Self, String> {
+        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<T: serde::de::DeserializeOwned>(
+        &self,
+        request: reqwest::RequestBuilder,
+    ) -> Result<T, String> {
+        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<Status, String> {
+        self.send(self.http.post("http://session/v1/sessions").json(request))
+            .await
+    }
+    pub async fn status(&self, id: Uuid) -> Result<Status, String> {
+        self.send(self.http.get(format!("http://session/v1/sessions/{id}")))
+            .await
+    }
+    pub async fn events(&self, id: Uuid, after: u64) -> Result<Events, String> {
+        self.send(self.http.get(format!(
+            "http://session/v1/sessions/{id}/events?after={after}"
+        )))
+        .await
+    }
+    pub async fn cancel(&self, id: Uuid) -> Result<Status, String> {
+        self.send(
+            self.http
+                .post(format!("http://session/v1/sessions/{id}/cancel")),
+        )
+        .await
+    }
+    pub async fn reconcile(&self, id: Uuid) -> Result<Status, String> {
+        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<Arc<AtomicUsize>>,
+}
+#[derive(Clone)]
+struct Adapter(Arc<Effect>);
+impl Backend for Adapter {
+    type Host = Self;
+    fn prepare(&self, _: &wire::Submit, _: &Run) -> Result<Self, String> {
+        Ok(self.clone())
+    }
+}
+impl managed_session::Host for Adapter {
+    async fn execute(
+        &self,
+        _: &Run,
+        _: tokio::sync::mpsc::Sender<Frame>,
+    ) -> Result<session::Outcome, String> {
+        panic!("cleanup-aware host required")
+    }
+    async fn execute_cancellable(
+        &self,
+        _: &Run,
+        frames: tokio::sync::mpsc::Sender<Frame>,
+        cancellation: CancellationToken,
+    ) -> Result<session::Outcome, String> {
+        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<Arc<AtomicUsize>>) -> 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<Service<Adapter>>,
+    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<String, Registration>,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn std::error::Error>> {
+    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.<app>. 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<B: Backend>(service: Arc<Service<B>>) -> Router {
+    Router::new()
+        .route("/healthz", get(|| async { "ok" }))
+        .route("/v1/sessions", post(submit::<B>))
+        .route("/v1/sessions/{id}", get(status::<B>))
+        .route("/v1/sessions/{id}/events", get(events::<B>))
+        .route("/v1/sessions/{id}/cancel", post(cancel::<B>))
+        .route("/v1/sessions/{id}/reconcile", post(reconcile::<B>))
+        .layer(DefaultBodyLimit::max(8 * 1024 * 1024))
+        .with_state(service)
+}
+async fn submit<B: Backend>(
+    State(service): State<Arc<Service<B>>>,
+    Json(request): Json<Submit>,
+) -> Result<Json<Status>, Error> {
+    service.submit(request).await.map(Json).map_err(rejected)
+}
+async fn status<B: Backend>(
+    State(service): State<Arc<Service<B>>>,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Status>, Error> {
+    service.status(id).map(Json).map_err(rejected)
+}
+#[derive(Deserialize)]
+struct Cursor {
+    #[serde(default)]
+    after: u64,
+}
+async fn events<B: Backend>(
+    State(service): State<Arc<Service<B>>>,
+    Path(id): Path<Uuid>,
+    Query(cursor): Query<Cursor>,
+) -> Result<Json<Events>, Error> {
+    service.events(id, cursor.after).map(Json).map_err(rejected)
+}
+async fn cancel<B: Backend>(
+    State(service): State<Arc<Service<B>>>,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Status>, Error> {
+    service.cancel(id).await.map(Json).map_err(rejected)
+}
+async fn reconcile<B: Backend>(
+    State(service): State<Arc<Service<B>>>,
+    Path(id): Path<Uuid>,
+) -> Result<Json<Status>, 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<B: Backend> Service<B> {
+    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(&current.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<Status, String> {
+        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<Self::Host, String>;
+}
+
+#[derive(Clone)]
+pub struct Options {
+    pub application: String,
+    pub host: String,
+    pub max_active: usize,
+    pub lease_seconds: u32,
+}
+
+pub struct Service<B: Backend> {
+    client: Client,
+    store: Store,
+    backend: B,
+    options: Options,
+    admission: AsyncMutex<()>,
+    active: Mutex<HashMap<Uuid, CancellationToken>>,
+    changed: Notify,
+    stopping: AtomicBool,
+}
+
+impl<B: Backend> Service<B> {
+    pub fn open(
+        client: Client,
+        root: &Path,
+        backend: B,
+        options: Options,
+    ) -> Result<Arc<Self>, 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<Self>, request: Submit) -> Result<Status, String> {
+        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<Self>, 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<Status, String> {
+        Ok(self.store.read(id)?.status)
+    }
+    pub fn events(
+        &self,
+        id: Uuid,
+        after: u64,
+    ) -> Result<airouter_agent_client::session_service::Events, String> {
+        self.store.events(id, after)
+    }
+
+    pub async fn cancel(&self, id: Uuid) -> Result<Status, String> {
+        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<Status, String> {
+        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<Self>) -> 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<String, Registration>,
+    executor: ExternalExecutor,
+}
+impl NativeBackend {
+    pub fn new(
+        registrations: BTreeMap<String, Registration>,
+        executor: ExternalExecutor,
+    ) -> Result<Self, String> {
+        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<NativeHost, String> {
+        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(&registration.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<Frame>) -> Result<Outcome, String> {
+        self.execute_cancellable(run, frames, CancellationToken::new())
+            .await
+    }
+    async fn execute_cancellable(
+        &self,
+        _: &Run,
+        frames: mpsc::Sender<Frame>,
+        cancellation: CancellationToken,
+    ) -> Result<Outcome, String> {
+        // 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<File, String> {
+    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<Self, String> {
+        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<Record, String> {
+        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<Record, String> {
+        let _guard = self.io.lock().unwrap();
+        self.read_inner(id)
+    }
+    pub fn exists(&self, id: Uuid) -> Result<bool, String> {
+        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<Record, String> {
+        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<Vec<Uuid>, 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<Events, String> {
+        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)
+    }
+}