fix: retain exact standalone result recovery owners across capacity and job errors

Recover captured Job/attempt identities independently; wait for actual task return and cleanup before freeing execution capacity. Preserve per-job rejection and wait causes without dropping subsequent reservations. Add one-time additive v7 admission-rejection evidence, never infer execution drain from Failed or an absent map.

Validated: parent Jobs 20 tests and store/recovery/migration 13 tests on the current dependency snapshot; root check, format, names and diff pass. This is an unfinished, unapproved T-841 checkpoint; R1/R6 and final whole-contract validation/review remain outstanding.

crates/standalone/src/host.rs

@@ -495,7 +495,7 @@ impl StandaloneHost {
         if let Some(connection) = &subject {
             connection.host.recover_jobs();
         }
-        let reserved_result_jobs = jobs.reserved_result_job_ids();
+        let reserved_result_jobs = jobs.reserved_result_attempts();
         let result_jobs = jobs.clone();
         let result_recovery = tokio::spawn(async move {
             result_jobs
@@ -615,7 +615,7 @@ impl StandaloneHost {
     }
 
     pub async fn shutdown(mut self) -> Result<(), StandaloneShutdownError> {
-        self.stop_result_recovery().await;
+        let recovery_result = self.stop_result_recovery().await;
         let jobs_result = self.jobs.shutdown().await;
         let command = protocol::WorkerCommandEnvelope::new(u64::MAX);
         let _ = self.handle.send(Method::Shutdown { command }).await;
@@ -627,7 +627,7 @@ impl StandaloneHost {
             self.retain_lease();
             return Err(error);
         }
-        if jobs_result.is_err() {
+        if jobs_result.is_err() || recovery_result.is_err() {
             self.retain_lease();
             return Err(StandaloneShutdownError::JobCleanup);
         }
@@ -665,14 +665,15 @@ impl StandaloneHost {
         Ok(())
     }
 
-    async fn stop_result_recovery(&mut self) {
+    async fn stop_result_recovery(&mut self) -> Result<(), crate::jobs::JobServiceError> {
         self.jobs.close();
         if let Some(task) = self.result_recovery.take() {
-            // This joins only the dispatcher. Every admitted Worker/Host effect
-            // remains owned by jobs.shutdown; cancelling a waiter proves no cleanup.
-            task.abort();
-            let _ = task.await;
+            // Close wakes capacity/storage waiters. Join the dispatcher and retain
+            // its error; only jobs.shutdown can confirm admitted execution cleanup.
+            task.await
+                .map_err(|error| crate::jobs::JobServiceError::Cleanup(error.to_string()))??;
         }
+        Ok(())
     }
 
     fn retain_lease(&mut self) {

crates/standalone/src/job_store.rs

@@ -29,7 +29,11 @@ mod legacy_migrations;
 #[path = "job_store/recovery.rs"]
 mod recovery;
 
-const SCHEMA_VERSION: i64 = 6;
+#[cfg(test)]
+#[path = "job_store/start_rejection_tests.rs"]
+mod start_rejection_tests;
+
+const SCHEMA_VERSION: i64 = 7;
 
 /// Availability of the immutable input body, independent of execution outcome.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
@@ -120,6 +124,9 @@ pub struct JobSnapshot {
     pub acknowledged: bool,
     /// A version cutover is final even when input/checkpoint remains incomplete.
     pub version_stopped: bool,
+    /// The exact current reservation was rejected before dispatch, by this Host.
+    /// This is not a runner shutdown or cleanup receipt.
+    pub start_rejected_by: Option<String>,
     /// All historical execution owners, independent of Job outcome/checkpoint.
     pub executions: Vec<JobExecutionOwnership>,
 }
@@ -135,14 +142,16 @@ pub struct JobExecutionOwnership {
 
 impl JobSnapshot {
     /// Terminal failure/completion without the attempt's admission binding is not
-    /// an empty set of successfully joined executions. Keep that missing evidence
-    /// distinct from an unstarted reservation and from checkpoint coverage.
+    /// an empty set of successfully joined executions. Only an explicit rejection
+    /// saved while Reserved can preclude that binding; it proves no runner cleanup.
+    /// All historical admitted executions still require their own cleanup proof.
     pub fn executions_released(&self) -> bool {
         let needs_binding = matches!(
             self.attempt.state,
             JobAttemptState::Dispatched | JobAttemptState::Completed | JobAttemptState::Failed
         );
         (!needs_binding
+            || (self.attempt.state == JobAttemptState::Failed && self.start_rejected_by.is_some())
             || self
                 .executions
                 .iter()
@@ -559,6 +568,7 @@ impl JobStore {
              WHERE a.state IN ('dispatched','failed','completed')
                AND NOT EXISTS(SELECT 1 FROM job_execution_ownership e WHERE e.attempt_id=a.attempt_id)
                AND NOT EXISTS(SELECT 1 FROM job_version_stops s WHERE s.job_id=i.job_id)
+               AND NOT (a.state='failed' AND EXISTS(SELECT 1 FROM job_start_rejections r WHERE r.job_id=i.job_id AND r.attempt_id=a.attempt_id))
              LIMIT 1", [], |row|row.get(0)).optional()?;
         if let Some(attempt) = missing {
             return Err(JobStoreError::Corrupt(format!(
@@ -1040,6 +1050,7 @@ fn read_snapshot(conn: &Connection, job_id: &str) -> Result<Option<JobSnapshot>,
                 state,
                 acknowledged,
                 version_stopped: conn.query_row("SELECT EXISTS(SELECT 1 FROM job_version_stops WHERE job_id=?1)", [job_id], |r|r.get(0))?,
+                start_rejected_by: conn.query_row("SELECT owner_id FROM job_start_rejections WHERE job_id=?1 AND attempt_id=?2", params![job_id,attempt_id], |r|r.get(0)).optional()?,
                 executions: {
                     let mut stmt = conn.prepare("SELECT attempt_id,owner_id,run_id,writes_revoked,cleanup_receipt FROM job_execution_ownership WHERE job_id=?1 ORDER BY rowid")?;
                     stmt.query_map([job_id], |r| Ok(JobExecutionOwnership { attempt_id:r.get(0)?,owner_id:r.get(1)?,run_id:r.get(2)?,writes_revoked:r.get(3)?,cleanup_receipt:r.get(4)? }))?.collect::<Result<Vec<_>,_>>()?

crates/standalone/src/job_store/recovery.rs

@@ -3,6 +3,55 @@ use super::*;
 pub(super) const UPDATE_INTERRUPTION: &str = "バージョン更新による実行打切り";
 
 impl JobStore {
+    pub(crate) fn probe_reserved_result_write(
+        &mut self,
+        job_id: &str,
+        attempt_id: &str,
+    ) -> Result<(), JobStoreError> {
+        let tx = self
+            .conn
+            .transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let saved = require_snapshot(&tx, job_id)?;
+        check_attempt(&saved, attempt_id)?;
+        if saved.version_stopped || saved.domain_grant.is_some() {
+            return Err(JobStoreError::AttemptMismatch);
+        }
+        // A write probe, not an execution completion or resource release receipt.
+        tx.execute(
+            "UPDATE job_attempts SET failure=failure WHERE job_id=?1 AND attempt_id=?2",
+            params![job_id, attempt_id],
+        )?;
+        tx.commit()?;
+        Ok(())
+    }
+
+    pub(crate) fn fail_reserved_result(
+        &mut self,
+        job_id: &str,
+        attempt_id: &str,
+        cause: &str,
+    ) -> Result<(), JobStoreError> {
+        let tx = self
+            .conn
+            .transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let saved = require_snapshot(&tx, job_id)?;
+        check_attempt(&saved, attempt_id)?;
+        if saved.version_stopped
+            || saved.domain_grant.is_some()
+            || saved.state != JobState::Pending
+            || saved.attempt.state != JobAttemptState::Reserved
+        {
+            return Err(JobStoreError::AttemptMismatch);
+        }
+        tx.execute(
+            "INSERT INTO job_start_rejections(job_id,attempt_id,owner_id) VALUES(?1,?2,?3)",
+            params![job_id, attempt_id, self.owner_id],
+        )?;
+        set_terminal(&tx, &saved, JobAttemptState::Failed, Some(cause))?;
+        tx.commit()?;
+        Ok(())
+    }
+
     /// The Host supplies the identity from its revalidated exclusive Worker lease.
     /// This binds admission, results and restart reconciliation to the same owner,
     /// rather than a fresh connection UUID that could impersonate a restart.
@@ -156,9 +205,15 @@ pub(super) fn initialize(conn: &mut Connection) -> Result<(), JobStoreError> {
                 }
                 migrate(&tx)?;
             }
+            6 => {}
             SCHEMA_VERSION => {}
             other => return Err(JobStoreError::UnsupportedSchema(other)),
         }
+        if version <= 6 {
+            // Additive admission evidence only. Do not rerun the v6 cutover or
+            // reinterpret preexisting Failed rows as rejected-before-dispatch.
+            tx.execute_batch(START_REJECTION_SCHEMA)?;
+        }
         if tx.prepare("PRAGMA foreign_key_check")?.exists([])? {
             return Err(JobStoreError::Corrupt(
                 "Job migration foreign key check failed".into(),
@@ -242,6 +297,27 @@ fn migrate(conn: &Connection) -> Result<(), JobStoreError> {
     Ok(())
 }
 
+// The insertion guard binds the rejection to the still-Reserved exact attempt.
+// Neither an absent execution row nor a Failed state can create this evidence.
+const START_REJECTION_SCHEMA: &str = "
+CREATE TABLE job_start_rejections (
+    job_id TEXT NOT NULL REFERENCES job_intents(job_id),
+    attempt_id TEXT NOT NULL PRIMARY KEY REFERENCES job_attempts(attempt_id),
+    owner_id TEXT NOT NULL
+);
+CREATE TRIGGER job_start_rejection_guard BEFORE INSERT ON job_start_rejections
+WHEN NOT EXISTS(SELECT 1 FROM job_intents i JOIN job_attempts a ON a.job_id=i.job_id AND a.number=i.current_attempt
+    WHERE i.job_id=NEW.job_id AND a.attempt_id=NEW.attempt_id AND i.state='pending' AND a.state='reserved')
+ OR EXISTS(SELECT 1 FROM job_execution_ownership WHERE attempt_id=NEW.attempt_id)
+ OR EXISTS(SELECT 1 FROM job_domain_grants WHERE job_id=NEW.job_id)
+ OR EXISTS(SELECT 1 FROM job_version_stops WHERE job_id=NEW.job_id)
+BEGIN SELECT RAISE(ABORT,'start rejection requires unstarted result-only reservation'); END;
+CREATE TRIGGER job_start_rejection_immutable BEFORE UPDATE ON job_start_rejections
+BEGIN SELECT RAISE(ABORT,'immutable Job start rejection'); END;
+CREATE TRIGGER job_start_rejection_retained BEFORE DELETE ON job_start_rejections
+BEGIN SELECT RAISE(ABORT,'retained Job start rejection'); END;
+";
+
 pub(super) const SCHEMA: &str = "
 CREATE TABLE job_version_history (
     job_id TEXT NOT NULL,

crates/standalone/src/job_store/start_rejection_tests.rs

@@ -0,0 +1,195 @@
+//! Admission evidence tests use the production store/initializer and real SQLite.
+use super::*;
+
+fn request(id: &str) -> JobRequest {
+    JobRequest {
+        job_id: id.into(),
+        purpose: "rejection fixture".into(),
+        input_ref: "fixture:input".into(),
+        input: serde_json::json!({"body":"retained"}),
+        instruction: "Return result".into(),
+        profile: "builtin:job".into(),
+        serialization_key: None,
+        limits: job::JobLimits::default(),
+    }
+}
+
+#[test]
+fn start_rejection_is_atomic_and_failed_or_missing_execution_never_implies_it() {
+    let temp = tempfile::tempdir().unwrap();
+    let path = temp.path().join("jobs.sqlite3");
+    let mut store = JobStore::open(&path).unwrap();
+    let saved = store.reserve(request("reject")).unwrap();
+    let attempt = &saved.attempt.attempt_id;
+    store
+        .conn
+        .execute_batch(
+            "CREATE TRIGGER refuse_failure BEFORE UPDATE OF state ON job_attempts
+        WHEN NEW.state='failed' BEGIN SELECT RAISE(ABORT,'injected terminal write failure'); END;",
+        )
+        .unwrap();
+    assert!(
+        store
+            .fail_reserved_result("reject", attempt, "configuration unavailable")
+            .is_err()
+    );
+    assert_eq!(store.get("reject").unwrap(), saved);
+    assert_eq!(
+        store
+            .conn
+            .query_row("SELECT COUNT(*) FROM job_start_rejections", [], |r| r
+                .get::<_, i64>(0))
+            .unwrap(),
+        0
+    );
+    store
+        .conn
+        .execute_batch("DROP TRIGGER refuse_failure")
+        .unwrap();
+    store
+        .fail_reserved_result("reject", attempt, "configuration unavailable")
+        .unwrap();
+    let rejected = store.get("reject").unwrap();
+    assert_eq!(
+        rejected.attempt.failure.as_deref(),
+        Some("configuration unavailable")
+    );
+    assert_eq!(
+        rejected.start_rejected_by.as_deref(),
+        Some(store.owner_id.as_str())
+    );
+    assert!(rejected.executions.is_empty());
+    assert!(rejected.executions_released());
+    assert!(store.execution_obligations().unwrap().is_empty());
+    for sql in [
+        "DELETE FROM job_start_rejections",
+        "UPDATE job_start_rejections SET owner_id='other'",
+    ] {
+        assert!(store.conn.execute_batch(sql).is_err());
+    }
+    let bare = store.reserve(request("bare-failed")).unwrap();
+    store
+        .finish(
+            "bare-failed",
+            &bare.attempt.attempt_id,
+            JobAttemptState::Failed,
+            Some("not admission evidence"),
+        )
+        .unwrap();
+    assert!(!store.get("bare-failed").unwrap().executions_released());
+    assert!(
+        store
+            .conn
+            .execute(
+                "INSERT INTO job_start_rejections VALUES('bare-failed',?1,'forged')",
+                [&bare.attempt.attempt_id]
+            )
+            .is_err()
+    );
+    assert!(store.execution_obligations().is_err());
+    let running = store.reserve(request("running")).unwrap();
+    store
+        .dispatch("running", &running.attempt.attempt_id, 8)
+        .unwrap();
+    assert!(
+        store
+            .fail_reserved_result("running", &running.attempt.attempt_id, "late failure")
+            .is_err()
+    );
+    let running = store.get("running").unwrap();
+    assert!(running.start_rejected_by.is_none());
+    assert!(!running.executions_released());
+    drop(store);
+    let reopened = JobStore::open(&path).unwrap();
+    assert_eq!(reopened.get("reject").unwrap(), rejected);
+    assert_eq!(reopened.get("running").unwrap(), running);
+    assert!(reopened.execution_obligations().is_err());
+}
+
+#[test]
+fn start_rejection_migration_adds_only_evidence_and_rolls_back_then_reopens_without_cutover() {
+    let temp = tempfile::tempdir().unwrap();
+    let path = temp.path().join("jobs.sqlite3");
+    let mut conn = Connection::open(&path).unwrap();
+    feature_storage::configure_connection(&conn).unwrap();
+    conn.execute_batch(SCHEMA).unwrap();
+    conn.execute_batch(GRANT_SCHEMA).unwrap();
+    conn.execute_batch(RECONCILIATION_SCHEMA).unwrap();
+    conn.execute_batch(recovery::SCHEMA).unwrap();
+    conn.pragma_update(None, "user_version", 6).unwrap();
+    let req = request("already-reserved");
+    let tx = conn.transaction().unwrap();
+    tx.execute("INSERT INTO job_intents(job_id,request_json,fingerprint,max_concurrent,state,current_attempt,acknowledged)
+        VALUES(?1,?2,?3,8,'pending',1,0)", params![req.job_id,serde_json::to_string(&req).unwrap(),req.fingerprint().unwrap()]).unwrap();
+    insert_attempt(&tx, &req, 1).unwrap();
+    tx.commit().unwrap();
+    conn.execute_batch("CREATE TRIGGER job_start_rejection_guard BEFORE UPDATE ON job_intents BEGIN SELECT 1; END;").unwrap();
+    assert!(recovery::initialize(&mut conn).is_err());
+    assert_eq!(
+        conn.pragma_query_value(None, "user_version", |r| r.get::<_, i64>(0))
+            .unwrap(),
+        6
+    );
+    assert!(
+        !conn
+            .query_row(
+                "SELECT EXISTS(SELECT 1 FROM sqlite_schema WHERE name='job_start_rejections')",
+                [],
+                |r| r.get::<_, bool>(0)
+            )
+            .unwrap()
+    );
+    conn.execute_batch("DROP TRIGGER job_start_rejection_guard")
+        .unwrap();
+    recovery::initialize(&mut conn).unwrap();
+    assert_eq!(
+        conn.query_row("SELECT COUNT(*) FROM job_start_rejections", [], |r| r
+            .get::<_, i64>(0))
+            .unwrap(),
+        0
+    );
+    assert_eq!(
+        conn.query_row("SELECT COUNT(*) FROM job_version_stops", [], |r| r
+            .get::<_, i64>(0))
+            .unwrap(),
+        0
+    );
+    drop(conn);
+    let mut store = JobStore::open(&path).unwrap();
+    let saved = store.get(&req.job_id).unwrap();
+    assert_eq!(saved.request, req);
+    assert_eq!(saved.state, JobState::Pending);
+    assert_eq!(saved.attempt.state, JobAttemptState::Reserved);
+    assert!(!saved.version_stopped);
+    assert!(saved.start_rejected_by.is_none());
+    store
+        .fail_reserved_result(
+            &req.job_id,
+            &saved.attempt.attempt_id,
+            "unavailable profile",
+        )
+        .unwrap();
+    let rejected = store.get(&req.job_id).unwrap();
+    drop(store);
+    let reopened = JobStore::open(&path).unwrap();
+    assert_eq!(reopened.get(&req.job_id).unwrap(), rejected);
+    assert!(reopened.execution_obligations().unwrap().is_empty());
+    let fresh = JobStore::open(&temp.path().join("fresh.sqlite3")).unwrap();
+    let schema = |conn: &Connection| {
+        conn.prepare(
+            "SELECT type,name,sql FROM sqlite_schema WHERE sql IS NOT NULL ORDER BY type,name",
+        )
+        .unwrap()
+        .query_map([], |r| {
+            Ok((
+                r.get::<_, String>(0)?,
+                r.get::<_, String>(1)?,
+                r.get::<_, String>(2)?,
+            ))
+        })
+        .unwrap()
+        .collect::<rusqlite::Result<Vec<_>>>()
+        .unwrap()
+    };
+    assert_eq!(schema(&reopened.conn), schema(&fresh.conn));
+}

crates/standalone/src/jobs.rs

@@ -14,6 +14,7 @@
 //! Timeout/shutdown fences unaccepted attempts before cancellation. Only owned runner shutdown
 //! plus joined/revoked Host callbacks discharge execution reservations. Clones cannot reopen
 //! the store after service closure or carry authority over to another Host.
+mod recovery;
 #[cfg(test)]
 mod tests;
 use std::collections::HashMap;
@@ -128,6 +129,10 @@ pub struct StandaloneJobs {
     domain: Arc<Mutex<Option<Arc<dyn JobDomainProvider>>>>,
     model_client: Arc<Mutex<Option<Box<dyn LlmClient>>>>,
     storage_recovered: Arc<Notify>,
+    capacity_changed: Arc<Notify>,
+    // Exact restore admissions and typed causes remain Host-owned, even if a
+    // dispatcher fails. These are identities, never copies of immutable bodies.
+    result_recovery: Arc<Mutex<HashMap<String, Arc<Mutex<Option<JobServiceError>>>>>>,
 }
 
 impl StandaloneJobs {
@@ -171,6 +176,8 @@ impl StandaloneJobs {
             domain: Arc::default(),
             model_client: Arc::default(),
             storage_recovered: Arc::default(),
+            capacity_changed: Arc::default(),
+            result_recovery: Arc::default(),
         })
     }
 
@@ -199,7 +206,9 @@ impl StandaloneJobs {
         Ok(jobs)
     }
 
-    pub(crate) fn reserved_result_job_ids(&self) -> Result<Vec<String>, JobServiceError> {
+    pub(crate) fn reserved_result_attempts(
+        &self,
+    ) -> Result<Vec<(String, String)>, JobServiceError> {
         // Retain IDs, not a second queue of immutable bodies outside the owned
         // execution barrier while awaiting an earlier Job's cleanup.
         Ok(self
@@ -210,23 +219,10 @@ impl StandaloneJobs {
                     && !saved.version_stopped
                     && saved.attempt.state == JobAttemptState::Reserved
             })
-            .map(|saved| saved.request.job_id)
+            .map(|saved| (saved.request.job_id, saved.attempt.attempt_id))
             .collect())
     }
 
-    /// The restore caller captures IDs before returning the Host to clients, then
-    /// runs this once after binding adapters. Granted Jobs keep domain recovery.
-    pub(crate) async fn resume_reserved_result_jobs(
-        &self,
-        ids: Vec<String>,
-    ) -> Result<(), JobServiceError> {
-        for id in ids {
-            self.start(&id).await?;
-            self.wait(&id).await?;
-        }
-        Ok(())
-    }
-
     /// Storage exhaustion is owned by the same execution Task, not a new pending
     /// Job state. The diagnostic retains its typed source until a successful Host
     /// store write wakes reconciliation; no polling or model retry is scheduled.
@@ -481,7 +477,7 @@ impl StandaloneJobs {
     }
 
     pub async fn start(&self, job_id: &str) -> Result<(), JobServiceError> {
-        self.start_with_optional_model_client(job_id, None).await
+        self.start_with_optional_model_client(job_id, None)
     }
 
     /// Injectable model transport for embedders and scripted execution tests.
@@ -496,10 +492,9 @@ impl StandaloneJobs {
         C: LlmClient + 'static,
     {
         self.start_with_optional_model_client(job_id, Some(Box::new(client)))
-            .await
     }
 
-    async fn start_with_optional_model_client(
+    fn start_with_optional_model_client(
         &self,
         job_id: &str,
         client: Option<Box<dyn LlmClient>>,
@@ -639,6 +634,7 @@ impl StandaloneJobs {
         let recovery_issue = Arc::new(Mutex::new(None));
         let task_issue = recovery_issue.clone();
         let storage_recovered = self.storage_recovered.clone();
+        let capacity_changed = self.capacity_changed.clone();
         let task = tokio::spawn(async move {
             let (sender_tx, mut sender_rx) = tokio::sync::oneshot::channel();
             let run = prepared.run_owned(attempt_id.clone(), sink, move |sender| {
@@ -715,6 +711,9 @@ impl StandaloneJobs {
             complete_tx.send_replace(Some(
                 outcome.as_ref().map(|_| ()).map_err(ToString::to_string),
             ));
+            // A wake is only an observation: recovery joins the successful task
+            // return before rechecking capacity. An error does not free its slot.
+            capacity_changed.notify_waiters();
             outcome
         });
         executions.jobs.insert(
@@ -770,6 +769,7 @@ impl StandaloneJobs {
                 .is_some_and(|run| run.attempt_id == attempt)
             {
                 executions.jobs.remove(job_id);
+                self.capacity_changed.notify_waiters();
             }
         }
         self.get(job_id)
@@ -778,6 +778,7 @@ impl StandaloneJobs {
     pub(crate) fn close(&self) {
         let mut executions = self.executions.lock().expect("Job executions poisoned");
         executions.closed = true;
+        self.capacity_changed.notify_waiters();
         for execution in executions.jobs.values() {
             execution.cancel.send_replace(true);
         }

crates/standalone/src/jobs/recovery.rs

@@ -0,0 +1,191 @@
+//! Restore ownership for result-only reservations. Domain recovery keeps its own contract.
+use super::*;
+
+impl StandaloneJobs {
+    /// Capture exact Job/attempt identities before exposing the restored Host.
+    /// Each admission has its own waiter: capacity, storage or cleanup failure in
+    /// one Job must not discard the other accepted reservations.
+    pub(crate) async fn resume_reserved_result_jobs(
+        &self,
+        ids: Vec<(String, String)>,
+    ) -> Result<(), JobServiceError> {
+        let mut owners = tokio::task::JoinSet::new();
+        for (id, attempt) in ids {
+            let issue = Arc::new(Mutex::new(None));
+            {
+                let mut admissions = self.result_recovery.lock().expect("Job recovery poisoned");
+                if admissions.contains_key(&id) {
+                    continue;
+                }
+                admissions.insert(id.clone(), issue.clone());
+            }
+            let jobs = self.clone();
+            owners.spawn(async move {
+                let outcome = tokio::select! {
+                    biased;
+                    _ = jobs.recovery_closed() => return Ok(()),
+                    outcome = jobs.resume_result_attempt(&id, &attempt, &issue) => outcome,
+                };
+                match outcome {
+                    Ok(()) => {
+                        jobs.result_recovery
+                            .lock()
+                            .expect("Job recovery poisoned")
+                            .remove(&id);
+                        Ok(())
+                    }
+                    Err(error) => {
+                        let detail = error.to_string();
+                        *issue.lock().expect("Job recovery issue poisoned") = Some(error);
+                        Err(JobServiceError::Cleanup(format!(
+                            "result recovery {id}/{attempt}: {detail}"
+                        )))
+                    }
+                }
+            });
+        }
+        let mut first_error = None;
+        while let Some(result) = owners.join_next().await {
+            let result =
+                result.unwrap_or_else(|error| Err(JobServiceError::Cleanup(error.to_string())));
+            if let Err(error) = result {
+                first_error.get_or_insert(error);
+            }
+        }
+        first_error.map_or(Ok(()), Err)
+    }
+
+    /// Unresolved restore causes are independent of execution/cleanup evidence.
+    pub fn result_recovery_diagnostic(&self, job_id: &str) -> Option<String> {
+        let admissions = self.result_recovery.lock().expect("Job recovery poisoned");
+        let issue = admissions.get(job_id)?;
+        let issue = issue.lock().expect("Job recovery issue poisoned");
+        Some(issue.as_ref().map_or_else(
+            || "accepted result-only reservation retained; awaiting execution capacity or owner settlement".into(),
+            ToString::to_string,
+        ))
+    }
+
+    async fn recovery_closed(&self) {
+        loop {
+            let changed = self.capacity_changed.notified();
+            tokio::pin!(changed);
+            changed.as_mut().enable();
+            if self
+                .executions
+                .lock()
+                .expect("Job executions poisoned")
+                .closed
+            {
+                return;
+            }
+            changed.await;
+        }
+    }
+
+    async fn resume_result_attempt(
+        &self,
+        id: &str,
+        attempt: &str,
+        issue: &Mutex<Option<JobServiceError>>,
+    ) -> Result<(), JobServiceError> {
+        loop {
+            // Register before checking capacity, including completed-but-not-yet-
+            // returned tasks, so a real release cannot be lost between checks.
+            let changed = self.capacity_changed.notified();
+            tokio::pin!(changed);
+            changed.as_mut().enable();
+            let completed: Vec<_> = self
+                .executions
+                .lock()
+                .expect("Job executions poisoned")
+                .jobs
+                .iter()
+                .filter(|(_, run)| matches!(run.complete.borrow().as_ref(), Some(Ok(()))))
+                .map(|(id, _)| id.clone())
+                .collect();
+            for completed_id in completed {
+                // wait joins task return and only then removes a successful slot.
+                // A failed read here is not evidence against this reservation.
+                let _ = self.wait(&completed_id).await;
+            }
+            let started = recover_owned_settlement(
+                &self.storage_recovered,
+                issue,
+                || self.start_restored_result_attempt(id, attempt),
+                || self.with_store_write(|store| store.probe_reserved_result_write(id, attempt)),
+            )
+            .await;
+            match started {
+                Err(
+                    error @ (JobServiceError::Capacity
+                    | JobServiceError::Store(
+                        JobStoreError::ConcurrentLimit | JobStoreError::SerializationConflict(_),
+                    )),
+                ) => {
+                    *issue.lock().expect("Job recovery issue poisoned") = Some(error);
+                    changed.await;
+                }
+                Err(error) => return Err(error),
+                Ok(()) => {
+                    // Retry only a failed result read, never the already-admitted
+                    // start. Cleanup errors retain their execution ownership.
+                    let outcome = self.wait(id).await;
+                    if let Err(error) = outcome {
+                        if !transient_settlement_storage_error(&error) {
+                            return Err(error);
+                        }
+                        recover_owned_settlement(
+                            &self.storage_recovered,
+                            issue,
+                            || self.get(id).map(|_| ()),
+                            || {
+                                self.with_store_write(|store| {
+                                    store.probe_reserved_result_write(id, attempt)
+                                })
+                            },
+                        )
+                        .await?;
+                    }
+                    return Ok(());
+                }
+            }
+        }
+    }
+
+    fn start_restored_result_attempt(
+        &self,
+        id: &str,
+        attempt: &str,
+    ) -> Result<(), JobServiceError> {
+        let saved = self.get(id)?;
+        if saved.attempt.attempt_id != attempt
+            || saved.domain_grant.is_some()
+            || saved.version_stopped
+        {
+            return Err(JobStoreError::AttemptMismatch.into());
+        }
+        // A client may have cancelled or started this exact attempt since capture.
+        // Never dispatch it twice, and never follow a newly-issued retry attempt.
+        if saved.state != JobState::Pending || saved.attempt.state != JobAttemptState::Reserved {
+            return Ok(());
+        }
+        match self.start_with_optional_model_client(id, None) {
+            Ok(()) => Ok(()),
+            Err(
+                error @ (JobServiceError::Capacity
+                | JobServiceError::Closed
+                | JobServiceError::Store(
+                    JobStoreError::ConcurrentLimit | JobStoreError::SerializationConflict(_),
+                )),
+            ) => Err(error),
+            Err(error) if transient_settlement_storage_error(&error) => Err(error),
+            Err(error) => {
+                // This path has no admitted execution. Persist the individual
+                // rejection without forging cleanup for any prior ownership.
+                let detail = format!("result_recovery_cannot_start: {error}");
+                self.with_store_write(|store| store.fail_reserved_result(id, attempt, &detail))
+            }
+        }
+    }
+}

crates/standalone/src/jobs/recovery_tests.rs

@@ -0,0 +1,245 @@
+//! Restore integration boundary: real SQLite, service admission, Internal Worker,
+//! result sink and cleanup. Only model transport and the WorkerStopping barrier
+//! are scripted. No process exit or absent map is used as shutdown evidence.
+use super::*;
+
+async fn await_capacity_owner(jobs: &StandaloneJobs, id: &str) {
+    tokio::time::timeout(Duration::from_secs(5), async {
+        loop {
+            if jobs
+                .result_recovery_diagnostic(id)
+                .is_some_and(|cause| cause.contains("capacity is exhausted"))
+            {
+                break;
+            }
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+}
+
+fn two_results() -> DomainTestClient {
+    let client = DomainTestClient::success();
+    client.scripts.lock().unwrap().extend(
+        DomainTestClient::success()
+            .scripts
+            .lock()
+            .unwrap()
+            .drain(..),
+    );
+    client
+}
+
+#[tokio::test]
+async fn restored_results_wait_for_domain_cleanup_then_resume_exact_attempts_without_clicks() {
+    let temp = tempfile::tempdir().unwrap();
+    let path = temp.path().join("jobs.sqlite3");
+    let prior = StandaloneJobs::open(&path, temp.path().into()).unwrap();
+    prior
+        .request_granted(domain_request("domain-first"), domain_grant("domain-first"))
+        .unwrap();
+    let first = prior.request(request("result-first")).unwrap();
+    let second = prior.request(request("result-second")).unwrap();
+    drop(prior);
+    let mut jobs = StandaloneJobs::open(&path, temp.path().into()).unwrap();
+    jobs.capacity = 1;
+    let hook = ControlledShutdown {
+        block: true,
+        ..Default::default()
+    };
+    let _release_on_failure = ReleaseShutdownOnDrop(hook.release.clone());
+    jobs.bind_domain(Arc::new(hook.clone())).unwrap();
+    let model = two_results();
+    jobs.bind_model_client(Box::new(model.clone()));
+    let ids = jobs.reserved_result_attempts().unwrap();
+    assert_eq!(ids.len(), 2);
+    jobs.start_with_model_client("domain-first", DomainTestClient::success())
+        .await
+        .unwrap();
+    tokio::time::timeout(Duration::from_secs(5), hook.entered.notified())
+        .await
+        .unwrap();
+    let domain = jobs.get("domain-first").unwrap();
+    assert_eq!(domain.state, JobState::Completed);
+    assert!(!domain.executions_released());
+    let recovery_jobs = jobs.clone();
+    let owner = tokio::spawn(async move { recovery_jobs.resume_reserved_result_jobs(ids).await });
+    await_capacity_owner(&jobs, "result-first").await;
+    await_capacity_owner(&jobs, "result-second").await;
+    assert_eq!(jobs.get("result-first").unwrap(), first);
+    assert_eq!(jobs.get("result-second").unwrap(), second);
+    assert!(model.requests.lock().unwrap().is_empty());
+    assert!(!owner.is_finished());
+    // A reservation accepted after restore capture is not silently swept in.
+    let outside = jobs.request(request("outside-capture")).unwrap();
+    hook.release.notify_one();
+    tokio::time::timeout(Duration::from_secs(10), owner)
+        .await
+        .unwrap()
+        .unwrap()
+        .unwrap();
+    assert!(jobs.get("domain-first").unwrap().executions_released());
+    for original in [first, second] {
+        let saved = jobs.get(&original.request.job_id).unwrap();
+        assert_eq!(saved.state, JobState::Completed);
+        assert_eq!(saved.request, original.request);
+        assert_eq!(saved.input_digest, original.input_digest);
+        assert_eq!(saved.attempt.attempt_id, original.attempt.attempt_id);
+        assert_eq!(saved.attempt.number, 1);
+        assert_eq!(saved.executions.len(), 1);
+        assert!(saved.executions_released());
+        assert_eq!(saved.result, Some(json!({"valid":true})));
+        assert!(
+            jobs.result_recovery_diagnostic(&saved.request.job_id)
+                .is_none()
+        );
+    }
+    assert_eq!(model.requests.lock().unwrap().len(), 4);
+    assert!(model.scripts.lock().unwrap().is_empty());
+    assert_eq!(jobs.get("outside-capture").unwrap(), outside);
+    let db = rusqlite::Connection::open(&path).unwrap();
+    assert_eq!(
+        db.query_row("SELECT COUNT(*) FROM job_intents", [], |row| row
+            .get::<_, i64>(0))
+            .unwrap(),
+        4
+    );
+    assert_eq!(
+        db.query_row("SELECT COUNT(*) FROM job_attempts", [], |row| row
+            .get::<_, i64>(0))
+            .unwrap(),
+        4
+    );
+    jobs.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn restored_results_save_individual_configuration_and_store_rejections_then_continue() {
+    let temp = tempfile::tempdir().unwrap();
+    let path = temp.path().join("jobs.sqlite3");
+    let prior = StandaloneJobs::open(&path, temp.path().into()).unwrap();
+    let mut invalid = request("bad-profile");
+    invalid.profile = "builtin:no-such-job-profile".into();
+    let bad_profile = prior.request(invalid).unwrap();
+    let bad_store = prior.request(request("store-rejection")).unwrap();
+    let good = prior.request(request("later-result")).unwrap();
+    drop(prior);
+    let jobs = StandaloneJobs::open(&path, temp.path().into()).unwrap();
+    let db = rusqlite::Connection::open(&path).unwrap();
+    db.execute_batch(
+        "CREATE TRIGGER reject_one_dispatch BEFORE UPDATE OF state ON job_attempts
+        WHEN NEW.job_id='store-rejection' AND NEW.state='dispatched'
+        BEGIN SELECT RAISE(FAIL, 'scripted individual dispatch rejection'); END;",
+    )
+    .unwrap();
+    let model = DomainTestClient::success();
+    jobs.bind_model_client(Box::new(model.clone()));
+    tokio::time::timeout(
+        Duration::from_secs(10),
+        jobs.resume_reserved_result_jobs(jobs.reserved_result_attempts().unwrap()),
+    )
+    .await
+    .unwrap()
+    .unwrap();
+    for (original, expected) in [
+        (bad_profile, "no-such-job-profile"),
+        (bad_store, "scripted individual dispatch rejection"),
+    ] {
+        let saved = jobs.get(&original.request.job_id).unwrap();
+        assert_eq!(saved.state, JobState::Failed);
+        assert_eq!(saved.attempt.attempt_id, original.attempt.attempt_id);
+        assert_eq!(saved.attempt.number, 1);
+        assert_eq!(saved.request, original.request);
+        assert!(saved.attempt.failure.as_deref().unwrap().contains(expected));
+        assert!(
+            saved
+                .attempt
+                .failure
+                .as_deref()
+                .unwrap()
+                .starts_with("result_recovery_cannot_start:")
+        );
+        assert!(saved.start_rejected_by.is_some());
+        assert!(
+            saved.executions_released(),
+            "explicit pre-dispatch rejection, not Failed alone"
+        );
+        assert!(
+            saved.executions.is_empty(),
+            "dispatch was rejected, not a cleanup receipt"
+        );
+    }
+    let saved = jobs.get("later-result").unwrap();
+    assert_eq!(saved.state, JobState::Completed);
+    assert_eq!(saved.attempt.attempt_id, good.attempt.attempt_id);
+    assert_eq!(saved.executions.len(), 1);
+    assert_eq!(model.requests.lock().unwrap().len(), 2);
+    assert_eq!(
+        db.query_row("SELECT COUNT(*) FROM job_attempts", [], |row| row
+            .get::<_, i64>(0))
+            .unwrap(),
+        3
+    );
+    jobs.shutdown().await.unwrap();
+}
+
+#[tokio::test]
+async fn restored_results_continue_after_wait_error_without_forging_cleanup() {
+    let temp = tempfile::tempdir().unwrap();
+    let jobs = StandaloneJobs::open(&temp.path().join("jobs.sqlite3"), temp.path().into()).unwrap();
+    let failed = jobs.request(request("wait-error")).unwrap();
+    let later = jobs.request(request("later-result")).unwrap();
+    let ids = jobs.reserved_result_attempts().unwrap();
+    let release = Arc::new(Notify::new());
+    let _release_on_failure = ReleaseShutdownOnDrop(release.clone());
+    let bad_model = DomainTestClient {
+        failure_release: Some(release.clone()),
+        ..DomainTestClient::success()
+    };
+    bad_model.scripts.lock().unwrap().pop_back();
+    jobs.start_with_model_client("wait-error", bad_model.clone())
+        .await
+        .unwrap();
+    let model = DomainTestClient::success();
+    jobs.bind_model_client(Box::new(model.clone()));
+    let owner_jobs = jobs.clone();
+    let owner = tokio::spawn(async move { owner_jobs.resume_reserved_result_jobs(ids).await });
+    // Wait for the later model to enter, then observe its real cleanup. The first
+    // owner still cannot finish; it is held before a known provider failure.
+    await_model(&model).await;
+    assert_eq!(
+        await_job(&jobs, "later-result").await.state,
+        JobState::Completed
+    );
+    assert_eq!(
+        jobs.get("later-result").unwrap().attempt.attempt_id,
+        later.attempt.attempt_id
+    );
+    assert!(!owner.is_finished());
+    release.notify_one();
+    assert!(
+        tokio::time::timeout(Duration::from_secs(10), owner)
+            .await
+            .unwrap()
+            .unwrap()
+            .is_err()
+    );
+    let saved = jobs.get("wait-error").unwrap();
+    assert_eq!(saved.state, JobState::Completed);
+    assert_eq!(saved.attempt.attempt_id, failed.attempt.attempt_id);
+    assert_eq!(saved.executions.len(), 1);
+    assert!(!saved.executions_released());
+    assert!(
+        jobs.result_recovery_diagnostic("wait-error")
+            .unwrap()
+            .contains("cleanup")
+    );
+    assert_eq!(model.requests.lock().unwrap().len(), 2);
+    assert_eq!(bad_model.requests.lock().unwrap().len(), 2);
+    assert!(matches!(
+        jobs.shutdown().await,
+        Err(JobServiceError::Cleanup(_))
+    ));
+    assert!(jobs.store.lock().unwrap().is_some());
+}

crates/standalone/src/jobs/tests.rs

@@ -1,3 +1,6 @@
+#[path = "recovery_tests.rs"]
+mod recovery_tests;
+
 use super::*;
 use agen::llm_client::{ClientError, Request, event::Event};
 use async_trait::async_trait;
@@ -731,7 +734,7 @@ async fn host_restart_rejects_live_lease_then_fails_only_unconfirmed_dispatch_wi
     reopened.bind_model_client(Box::new(DomainTestClient::success()));
     tokio::time::timeout(
         Duration::from_secs(5),
-        reopened.resume_reserved_result_jobs(reopened.reserved_result_job_ids().unwrap()),
+        reopened.resume_reserved_result_jobs(reopened.reserved_result_attempts().unwrap()),
     )
     .await
     .unwrap()