path: root/net-misc/airouter/files/airouter-managed-task-outcomes.patch

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