Persist supervised SSH receipts before import and cleanup

docs/qualification/supervised-ssh.md

@@ -1,8 +1,10 @@
 # Supervised SSH qualification
 
 Status: **not qualified**. The attempt adapter has deterministic injected-transport
-coverage. Normal supervisor dispatch still refuses SSH before launch. There is no
-shipped lifecycle SSH client or operator trust-config loader for that dispatch.
+coverage. Supervisor dispatch supports explicitly installed trusted embedding
+configuration, with authenticated lifecycle receipts and collection recovery.
+Normal CLI dispatch still refuses SSH before launch because there is no shipped
+lifecycle SSH client or operator trust-config loader for that dispatch.
 The existing atomic command SSH client speaks a different protocol. Do not use it
 to manufacture lifecycle receipts or treat a direct remote command as supervised
 execution.
@@ -27,7 +29,9 @@ An uncertain launch is queried using the submitted identity and never relaunched
 execution identity, collection key, and policy. Stage uncertainty remains an
 operator reconciliation gate. Cancellation uses the existing lifecycle
 TERM/grace/KILL/wait sequence, and collection requires a terminal observation.
-The adapter never calls workspace cleanup, including on drop or error.
+The attempt adapter never calls workspace cleanup on drop or error. The supervisor
+requests it separately after authenticated collection persistence, idempotent local
+import, successful local recapture, and durable import/recapture receipts.
 
 ## Deterministic checks
 
@@ -39,18 +43,52 @@ cargo test --locked -p multi-agent-coding-orchestrator --lib external_agent::exe
 cargo test --locked -p multi-agent-coding-orchestrator --lib external_agent::executor::supervised_tests::lost_launch_reconciles_before_cancellation_without_relaunch -- --exact
 cargo test --locked -p multi-agent-coding-orchestrator --lib external_agent::executor::supervised_tests::cancellation_keeps_custody_until_terminal_proof -- --exact
 cargo test --locked -p multi-agent-coding-orchestrator --lib external_agent::executor::supervised_tests::lost_collection_reuses_exact_key_and_never_cleans_up -- --exact
+cargo test --locked -p multi-agent-coding-orchestrator --lib supervise::remote_execution::tests::lost_reply_restart_never_duplicates_launch -- --exact
+cargo test --locked -p multi-agent-coding-orchestrator --lib supervise::remote_execution::tests::cleanup_requires_persisted_imported_collection -- --exact
+cargo test --locked -p multi-agent-coding-orchestrator --lib supervise::remote_execution::tests::selected_supervisor_dispatch_requires_explicit_trust_and_keeps_local_default -- --exact
 ```
 
 ## Opt-in two-host gate (do not run as part of deterministic checks)
 
-This gate remains blocked until production supervisor wiring and durable recovery
-ship. That follow-up must persist intent/receipts before the next external effect,
-persist the bounded collection bundle before local import, recapture the imported
-candidate using coordinator-owned inspection, and authorize cleanup only after
-both persistence and import/recapture succeed. It must also provide the real
-lifecycle client and reviewed operator trust configuration. The in-memory adapter
-alone does not provide crash recovery or duplicate-launch protection after a
-coordinator restart.
+The live gate remains unproven. An embedding caller can configure
+`supervise::remote_execution::SupervisedSshHost::configure` with matching
+`SshConfig`, a lifecycle transport factory, and a trusted
+`SupervisedSshCoordinator`. `bind_supervised_ssh_hosts` captures that configuration
+when loading the plan, including for scheduler threads. A plan cannot install
+these capabilities. Missing, changed, or revoked trust refuses without Local
+fallback. Attached parent messaging and hosted pre-action review require an
+equivalent remote capability and currently refuse SSH.
+
+The coordinator integration supplies bounded frozen inputs, imports the validated
+candidate idempotently using its exact collection key, and recaptures the local
+candidate and held native runtime evidence. Remote JSON cannot manufacture that
+native evidence. Existing claim, budget, containment, acceptance, and publication
+gates still apply to the returned run. No default integration or real wire client
+is installed by this API.
+After native recapture returns, receipt or cleanup failures retain that run's
+observations for normal supervisor usage settlement and mark it nonpublishable.
+
+Receipts live in the repository-authenticated `supervised-ssh-v1` journal namespace.
+The stable run/assignment/optional Worker/attempt tuple selects one exclusively
+locked journal. Its first intent retains the nonce; checkpoints bind the exact
+stage key, session, workspace, launch key, PID/PGID/start token, and collection
+policy. Launch intent is written before launch. A restart queries that submitted
+identity, including when the reply was lost; `NotFound` does not authorize replay.
+Collections use bounded chunks with an authenticated final digest/commit before
+import. An interrupted collection retries its original key. Failed persistence,
+import, or recapture preserves remote output. Cleanup is recorded before its
+effect and reconciled using the exact cleanup key only after the import barrier.
+Stage uncertainty remains an operator gate.
+
+An embedding caller resumes with `SupervisedSshHost::dispatch` using the identical
+identity, command, scopes, and frozen inputs. A completed collection is restored
+and revalidated locally, without fetching remote output again; native runtime
+evidence must be recaptured, never restored from serialized flags. The general
+CLI `supervise resume` still refuses uncertain assignments: reconciliation of
+remote custody alone does not settle the interrupted scheduler's accounting or
+accept its candidate. A future operator integration must document its actual
+trust-loader, recovery/accounting handoff, and fault-control commands before a
+live qualification can begin.
 
 Use an existing authenticated, explicitly trusted second host and a disposable
 qualification repository. Do not create credentials, guess endpoints, weaken
@@ -60,7 +98,7 @@ to the accepted installed binary. `SSH_PLAN` is a reviewed plan containing one
 substantive bounded source edit with an explicit SSH executor. `EVIDENCE` is an
 existing private directory. Choose a fresh `RUN_ID` and use it throughout.
 
-After the production follow-up makes this route available, the exact supervisor
+After an operator integration makes the CLI route available, the exact supervisor
 commands are:
 
 ```sh

src/external_agent/executor/mod.rs

@@ -10,24 +10,25 @@
 //! [`SupervisedSshExecutor`] binds an explicitly selected, locally validated trust
 //! configuration to a transport factory and an attempt-scoped lifecycle. It
 //! translates declared local paths to manifest names and observes the existing
-//! cancellation token. It does not install a production SSH client or enable
-//! supervisor dispatch, durable resume, candidate import, or cleanup.
+//! cancellation token. It does not install a lifecycle SSH wire client.
+//! The supervisor owns durable receipts, recovery,
+//! candidate import/recapture, and cleanup through explicit embedding configuration.
 //!
 //! Collection validates and returns manifest-bound candidate evidence without
-//! deleting its remote workspace. Cleanup is a separate typed effect that a future
-//! coordinator may request only after durable persistence and local import/recapture;
-//! that coordinator wiring is intentionally not claimed here.
+//! deleting its remote workspace. Cleanup is a separate typed effect requested by
+//! the coordinator only after durable persistence and local import/recapture.
 //!
 //! `LocalExecutor` is compatibility forwarding, not remote-lifecycle trait parity. The real
 //! crate's existing concrete higher-ranked reviewed runner forwards through it without
-//! changing the callback surface. Production supervisor wiring remains outside
-//! this executor-owned foundation.
+//! changing the callback surface. Supervisor authority remains outside this
+//! executor-owned foundation.
 
 mod checksum;
 mod local;
 mod selection;
 mod ssh;
 mod supervised;
+mod supervised_checkpoint;
 mod types;
 
 pub use local::LocalExecutor;
@@ -37,6 +38,7 @@ pub use supervised::{
     SupervisedAttemptId, SupervisedEvent, SupervisedPaths, SupervisedSshAttempt,
     SupervisedSshExecutor,
 };
+pub(crate) use supervised_checkpoint::{DurableCollection, SupervisedCheckpoint};
 pub use types::*;
 
 #[cfg(test)]

src/external_agent/executor/ssh.rs

@@ -359,25 +359,10 @@ impl<T: SshTransport> AgentExecutor for SshExecutor<T> {
                     })));
                 }
             };
-        let manifest_digest = receipt.manifest_digest.clone();
-        let validated = validate_collection_contents(&policy, receipt).map_err(|error| {
-            ExecutorError::CollectionRejected {
-                reason: error.to_string(),
-            }
-        })?;
-        Ok(Effect::Confirmed(CollectedResult {
-            identity: request.identity,
-            patch: validated.patch,
-            patch_digest: validated.patch_digest,
-            changed_paths: validated.changed_paths,
-            outputs: validated.outputs,
-            binding: CollectionEvidenceBinding {
-                manifest_digest,
-                policy_digest,
-                collection_key: key,
-            },
-            candidate_evidence_only: true,
-        }))
+        Ok(Effect::Confirmed(restore_collection_receipt(
+            &request.identity,
+            receipt,
+        )?))
     }
 
     fn cleanup(&self, request: CleanupRequest) -> ExecutorResult<Effect<CleanupReceipt>> {
@@ -744,6 +729,41 @@ fn validate_execution_collection_authority(identity: &ExecutionIdentity) -> Exec
     )
 }
 
+/// Revalidate authenticated local collection bytes through the same boundary as
+/// a fresh receipt. Restoring a checkpoint never creates native run evidence.
+pub(super) fn restore_collection_receipt(
+    identity: &ExecutionIdentity,
+    receipt: CollectionReceipt,
+) -> ExecutorResult<CollectedResult> {
+    validate_execution_collection_authority(identity)?;
+    let authority = identity.collection_authority();
+    let receipt = validate_collection_envelope(
+        identity,
+        authority.policy_digest(),
+        authority.collection_key(),
+        receipt,
+    )?;
+    let manifest_digest = receipt.manifest_digest.clone();
+    let validated = validate_collection_contents(authority.policy(), receipt).map_err(|error| {
+        ExecutorError::CollectionRejected {
+            reason: error.to_string(),
+        }
+    })?;
+    Ok(CollectedResult {
+        identity: identity.clone(),
+        patch: validated.patch,
+        patch_digest: validated.patch_digest,
+        changed_paths: validated.changed_paths,
+        outputs: validated.outputs,
+        binding: CollectionEvidenceBinding {
+            manifest_digest,
+            policy_digest: authority.policy_digest().clone(),
+            collection_key: authority.collection_key().clone(),
+        },
+        candidate_evidence_only: true,
+    })
+}
+
 struct ValidatedCollection {
     patch: Vec<u8>,
     patch_digest: Digest,

src/external_agent/executor/supervised.rs

@@ -3,7 +3,7 @@
 //! This seam does not create local run evidence from remote bytes. Its events
 //! must be persisted by the coordinator before advancing across effects. Durable
 //! resume, candidate import/recapture, and cleanup authorization belong to that
-//! coordinator; cleanup is deliberately absent from this adapter.
+//! coordinator. Advancing an attempt never authorizes cleanup.
 
 use super::*;
 use crate::executor::{SshConfig, SshConfigInput};
@@ -14,7 +14,8 @@ use std::{
 };
 
 /// Coordinator-owned identity. Reuse the nonce only when resuming this attempt.
-#[derive(Debug, Clone, PartialEq, Eq)]
+#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
 pub struct SupervisedAttemptId {
     pub run_id: RunId,
     pub assignment_id: AssignmentId,
@@ -102,6 +103,13 @@ pub struct SupervisedSshExecutor<T> {
 }
 
 impl<T: SshTransport> SupervisedSshExecutor<T> {
+    pub(crate) fn cancellation_grace_millis(&self) -> u64 {
+        self.trust.limits().cancellation_grace_millis
+    }
+    pub(crate) fn validate_trust(&self) -> ExecutorResult<()> {
+        revalidate_trust(&self.trust)
+    }
+
     pub fn configure(
         selection: &ExecutorSelection,
         trust: Option<SshConfig>,
@@ -198,7 +206,7 @@ pub enum SupervisedEvent {
 }
 
 #[allow(clippy::large_enum_variant)]
-enum State {
+pub(super) enum State {
     Prepared,
     Staged(StagedAssignment),
     StageUncertain(Box<UncertainEffect>),
@@ -214,15 +222,33 @@ enum State {
 /// Dropping it does not establish remote quiescence. The coordinator must retain
 /// custody on errors/uncertain effects and persist bindings before dispatch.
 pub struct SupervisedSshAttempt<'a, T> {
-    owner: &'a SupervisedSshExecutor<T>,
-    binding: SupervisedAttemptId,
-    request: StageRequest,
+    pub(super) owner: &'a SupervisedSshExecutor<T>,
+    pub(super) binding: SupervisedAttemptId,
+    pub(super) request: StageRequest,
     cancellation: ProcessCancellation,
     termination: TerminationPolicy,
-    state: State,
+    pub(super) state: State,
 }
 
 impl<T: SshTransport> SupervisedSshAttempt<'_, T> {
+    pub(crate) fn observe_prelaunch_cancellation(&mut self) {
+        if matches!(self.state, State::Prepared | State::Staged(_))
+            && self.cancellation.is_cancelled()
+        {
+            self.state = State::CancelledBeforeLaunch;
+        }
+    }
+    /// Only the coordinator may authorize this after its durable import barrier.
+    pub(crate) fn cleanup_collected(&self) -> ExecutorResult<Effect<CleanupReceipt>> {
+        let State::Collected(result) = &self.state else {
+            return Err(invalid("cleanup requires a collected attempt"));
+        };
+        revalidate_trust(&self.owner.trust)?;
+        self.owner
+            .executor
+            .cleanup(CleanupRequest::new(result.identity().clone()))
+    }
+
     pub fn binding(&self) -> &SupervisedAttemptId {
         &self.binding
     }
@@ -252,11 +278,7 @@ impl<T: SshTransport> SupervisedSshAttempt<'_, T> {
     /// inherited cancellation is observed between bounded transport operations.
     /// Transport implementations must also bound their own connection/read time.
     pub fn advance(&mut self) -> ExecutorResult<SupervisedEvent> {
-        if matches!(self.state, State::Prepared | State::Staged(_))
-            && self.cancellation.is_cancelled()
-        {
-            self.state = State::CancelledBeforeLaunch;
-        }
+        self.observe_prelaunch_cancellation();
         if matches!(self.state, State::CancelledBeforeLaunch) {
             return Ok(SupervisedEvent::CancelledBeforeLaunch);
         }