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

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