fix(standalone): retain exact consolidation admission through recovery

Own accepted never-dispatched consolidation reservations across actual capacity and typed source storage failures. Subscribe before capacity checks, revalidate original binding/attempt/input/grant/retirement and all prior cleanup, join successful resource owners, preserve one deadline and do not redispatch admitted executions. Distinguish settled/admission-required/retained outcomes so Reserved storage recovery transfers the original admission.

Persist pre-dispatch refusal using existing rejection evidence with a v8 guard-only migration, retaining immutable v7 text and old Job cutover semantics. Actual migration rollback/retry/reopen/fresh-schema and execution-negative controls; no operating migration or inferred cleanup.

Validation: old-source D1 red2/D2 red2; focused11 and nonempty finalizer integration; parent normal standalone159+11+9(ignored1)+9/doc0; downstream connected7; root/doc/fmt/names/diff. Child long-temp SUN_LEN failures retained; same-source full parent package passes under prior short isolated temp root. 1230 actual source hashes unchanged during validation. Fresh whole-current/all-seven-AC formal review pending; no integration or Ticket conclusion.

crates/standalone/src/job_store.rs

@@ -33,7 +33,7 @@ mod recovery;
 #[path = "job_store/start_rejection_tests.rs"]
 mod start_rejection_tests;
 
-const SCHEMA_VERSION: i64 = 7;
+const SCHEMA_VERSION: i64 = 8;
 
 /// Availability of the immutable input body, independent of execution outcome.
 #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]

crates/standalone/src/job_store/admission_tests.rs

@@ -0,0 +1,130 @@
+//! Actual v7 guard upgrade: rollback/retry, no cutover, exact values and fresh
+//! schema equality. Temporary SQLite only; this is not an operating DB migration.
+use super::*;
+
+#[test]
+fn granted_admission_guard_upgrade_rolls_back_retries_preserves_reservations_and_reopens() {
+    let temp = tempfile::tempdir().unwrap();
+    let path = temp.path().join("jobs.sqlite3");
+    let mut store = JobStore::open(&path).unwrap();
+    let mut request = super::super::tests::request("granted-refusal");
+    request.serialization_key = None;
+    let granted = store
+        .reserve_granted(request.clone(), Some(serde_json::json!({"exact":true})))
+        .unwrap();
+    request.job_id = "prior-execution".into();
+    let pending = store.reserve(request).unwrap();
+    let dispatched = store
+        .dispatch(&pending.request.job_id, &pending.attempt.attempt_id, 8)
+        .unwrap();
+    drop(store);
+    let mut conn = Connection::open(&path).unwrap();
+    feature_storage::configure_connection(&conn).unwrap();
+    // v8 has no table/row changes. Restore the actual immutable v7 guard text.
+    let old_guard = &START_REJECTION_SCHEMA[START_REJECTION_SCHEMA
+        .find("CREATE TRIGGER job_start_rejection_guard")
+        .unwrap()
+        ..START_REJECTION_SCHEMA
+            .find("CREATE TRIGGER job_start_rejection_immutable")
+            .unwrap()];
+    conn.execute_batch("DROP TRIGGER job_start_rejection_guard")
+        .unwrap();
+    conn.execute_batch(old_guard).unwrap();
+    conn.pragma_update(None, "user_version", 7).unwrap();
+    conn.execute_batch("CREATE TABLE migration_fault(value TEXT REFERENCES job_intents(job_id))")
+        .unwrap();
+    conn.pragma_update(None, "foreign_keys", false).unwrap();
+    conn.execute("INSERT INTO migration_fault VALUES('missing')", [])
+        .unwrap();
+    conn.pragma_update(None, "foreign_keys", true).unwrap();
+    assert!(
+        initialize(&mut conn).is_err(),
+        "actual FK check after DDL must abort the transaction"
+    );
+    assert_eq!(
+        conn.pragma_query_value(None, "user_version", |row| row.get::<_, i64>(0))
+            .unwrap(),
+        7
+    );
+    let guard: String = conn
+        .query_row(
+            "SELECT sql FROM sqlite_schema WHERE name='job_start_rejection_guard'",
+            [],
+            |row| row.get(0),
+        )
+        .unwrap();
+    assert!(
+        guard.contains("result-only"),
+        "failed migration must restore old guard"
+    );
+    assert_eq!(
+        require_snapshot(&conn, &granted.request.job_id).unwrap(),
+        granted
+    );
+    assert_eq!(
+        require_snapshot(&conn, &dispatched.request.job_id).unwrap(),
+        dispatched
+    );
+    conn.execute_batch("DROP TABLE migration_fault").unwrap();
+    initialize(&mut conn).unwrap();
+    assert_eq!(
+        require_snapshot(&conn, &granted.request.job_id).unwrap(),
+        granted
+    );
+    assert_eq!(
+        require_snapshot(&conn, &dispatched.request.job_id).unwrap(),
+        dispatched
+    );
+    assert_eq!(
+        conn.query_row("SELECT count(*) FROM job_version_stops", [], |row| row
+            .get::<_, i64>(0))
+            .unwrap(),
+        0
+    );
+    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([], |row| {
+            Ok((
+                row.get::<_, String>(0)?,
+                row.get::<_, String>(1)?,
+                row.get::<_, String>(2)?,
+            ))
+        })
+        .unwrap()
+        .collect::<Result<Vec<_>, _>>()
+        .unwrap()
+    };
+    let fresh = JobStore::open(&temp.path().join("fresh.sqlite3")).unwrap();
+    assert_eq!(schema(&conn), schema(&fresh.conn));
+    drop(conn);
+    let mut reopened = JobStore::open(&path).unwrap();
+    reopened
+        .fail_reserved_granted(&granted, "permanent configuration refusal")
+        .unwrap();
+    let failed = reopened.get(&granted.request.job_id).unwrap();
+    assert_eq!(
+        failed.attempt.failure.as_deref(),
+        Some("permanent configuration refusal")
+    );
+    assert_eq!(failed.request, granted.request);
+    assert_eq!(failed.domain_grant, granted.domain_grant);
+    assert!(failed.start_rejected_by.is_some());
+    assert!(failed.executions.is_empty());
+    assert_eq!(
+        reopened.get(&dispatched.request.job_id).unwrap(),
+        dispatched
+    );
+    assert!(!dispatched.executions_released());
+    assert!(reopened.conn.execute("INSERT INTO job_start_rejections(job_id,attempt_id,owner_id) VALUES(?1,?2,'not-an-owner')",
+        params![dispatched.request.job_id, dispatched.attempt.attempt_id]).is_err());
+    drop(reopened);
+    let reopened = JobStore::open(&path).unwrap();
+    assert_eq!(reopened.get(&granted.request.job_id).unwrap(), failed);
+    assert_eq!(
+        reopened.get(&dispatched.request.job_id).unwrap(),
+        dispatched
+    );
+}

crates/standalone/src/job_store/recovery.rs

@@ -1,5 +1,9 @@
 use super::*;
 
+#[cfg(test)]
+#[path = "admission_tests.rs"]
+mod admission_tests;
+
 pub(super) const UPDATE_INTERRUPTION: &str = "バージョン更新による実行打切り";
 
 impl JobStore {
@@ -52,6 +56,42 @@ impl JobStore {
         Ok(())
     }
 
+    /// A pre-dispatch refusal is not execution cleanup. Bind it to the exact
+    /// available granted reservation and never clear historical execution owners.
+    pub(crate) fn fail_reserved_granted(
+        &mut self,
+        expected: &JobSnapshot,
+        cause: &str,
+    ) -> Result<(), JobStoreError> {
+        let tx = self
+            .conn
+            .transaction_with_behavior(TransactionBehavior::Immediate)?;
+        let saved = require_snapshot(&tx, &expected.request.job_id)?;
+        check_attempt(&saved, &expected.attempt.attempt_id)?;
+        saved.load_executable_request()?;
+        if saved.input_digest != expected.input_digest
+            || saved.domain_grant != expected.domain_grant
+            || saved.domain_grant.is_none()
+            || saved.version_stopped
+            || saved.state != JobState::Pending
+            || saved.attempt.state != JobAttemptState::Reserved
+            || !saved.executions_released()
+        {
+            return Err(JobStoreError::AttemptMismatch);
+        }
+        tx.execute(
+            "INSERT INTO job_start_rejections(job_id,attempt_id,owner_id) VALUES(?1,?2,?3)",
+            params![
+                saved.request.job_id,
+                saved.attempt.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.
@@ -205,7 +245,7 @@ pub(super) fn initialize(conn: &mut Connection) -> Result<(), JobStoreError> {
                 }
                 migrate(&tx)?;
             }
-            6 => {}
+            6 | 7 => {}
             SCHEMA_VERSION => {}
             other => return Err(JobStoreError::UnsupportedSchema(other)),
         }
@@ -214,6 +254,11 @@ pub(super) fn initialize(conn: &mut Connection) -> Result<(), JobStoreError> {
             // reinterpret preexisting Failed rows as rejected-before-dispatch.
             tx.execute_batch(START_REJECTION_SCHEMA)?;
         }
+        if version <= 7 {
+            // Extend the existing refusal evidence to explicit domain admissions.
+            // No Job is cut over, reissued or retroactively declared unstarted.
+            tx.execute_batch(GRANTED_START_REJECTION_SCHEMA)?;
+        }
         if tx.prepare("PRAGMA foreign_key_check")?.exists([])? {
             return Err(JobStoreError::Corrupt(
                 "Job migration foreign key check failed".into(),
@@ -318,6 +363,19 @@ CREATE TRIGGER job_start_rejection_retained BEFORE DELETE ON job_start_rejection
 BEGIN SELECT RAISE(ABORT,'retained Job start rejection'); END;
 ";
 
+// v8 changes only the guard. Keep the original v7 migration above intact.
+const GRANTED_START_REJECTION_SCHEMA: &str = "
+DROP TRIGGER job_start_rejection_guard;
+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'
+      AND i.input_disposition='available')
+ OR EXISTS(SELECT 1 FROM job_execution_ownership WHERE attempt_id=NEW.attempt_id)
+ OR EXISTS(SELECT 1 FROM job_execution_ownership WHERE job_id=NEW.job_id AND (writes_revoked=0 OR cleanup_receipt IS NULL))
+ OR EXISTS(SELECT 1 FROM job_version_stops WHERE job_id=NEW.job_id)
+BEGIN SELECT RAISE(ABORT,'start rejection requires an available unstarted reservation and prior cleanup'); END;
+";
+
 pub(super) const SCHEMA: &str = "
 CREATE TABLE job_version_history (
     job_id TEXT NOT NULL,

crates/standalone/src/jobs.rs

@@ -41,6 +41,8 @@ pub enum JobServiceError {
     Store(#[from] JobStoreError),
     #[error("standalone Job execution requirements are unavailable: {0}")]
     Configuration(String),
+    #[error(transparent)]
+    DomainStorage(#[from] feature_storage::FeatureStorageError),
     #[error("the standalone Job Host is shutting down")]
     Closed,
     #[error("standalone Job execution capacity is exhausted")]
@@ -82,7 +84,7 @@ pub(crate) trait JobDomainProvider: Send + Sync {
         &self,
         snapshot: &JobSnapshot,
         fence: Arc<dyn JobAttemptFence>,
-    ) -> Result<worker::job::JobFeatureGrant, String>;
+    ) -> Result<worker::job::JobFeatureGrant, JobServiceError>;
 }
 
 type OwnedExecutionReport = Arc<
@@ -520,6 +522,15 @@ impl StandaloneJobs {
         &self,
         job_id: &str,
         client: Option<Box<dyn LlmClient>>,
+    ) -> Result<(), JobServiceError> {
+        self.start_with_admission(job_id, client, None)
+    }
+
+    fn start_with_admission(
+        &self,
+        job_id: &str,
+        client: Option<Box<dyn LlmClient>>,
+        admission: Option<(&JobSnapshot, tokio::time::Instant)>,
     ) -> Result<(), JobServiceError> {
         // Preparation owns copies of the immutable input and domain adapter too. Exclude
         // owner shutdown before creating them, not only before publishing the task.
@@ -535,6 +546,16 @@ impl StandaloneJobs {
                 .map(|c| c.clone_boxed())
         });
         let snapshot = self.get(job_id)?;
+        if let Some((expected, _)) = admission {
+            if snapshot.attempt.attempt_id != expected.attempt.attempt_id
+                || snapshot.input_digest != expected.input_digest
+                || snapshot.domain_grant != expected.domain_grant
+                || snapshot.domain_grant.is_none()
+                || !snapshot.executions_released()
+            {
+                return Err(JobStoreError::AttemptMismatch.into());
+            }
+        }
         snapshot.load_executable_request()?;
         if snapshot.version_stopped
             || snapshot.state != JobState::Pending
@@ -547,8 +568,19 @@ impl StandaloneJobs {
         }
         let effect_gate = Arc::new(Mutex::new(()));
         let revoked = Arc::new(AtomicBool::new(false));
-        let timeout = Duration::from_secs(u64::from(snapshot.request.limits.timeout_seconds));
+        let deadline = admission.map_or_else(
-        let deadline = tokio::time::Instant::now() + timeout;
+            || {
+                tokio::time::Instant::now()
+                    + Duration::from_secs(u64::from(snapshot.request.limits.timeout_seconds))
+            },
+            |(_, deadline)| deadline,
+        );
+        let timeout = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if timeout.is_zero() {
+            return Err(JobServiceError::Configuration(
+                "consolidation_admission_deadline_exceeded".into(),
+            ));
+        }
         let fence: Arc<dyn JobAttemptFence> = Arc::new(BoundAttemptFence {
             revoked: revoked.clone(),
             effect_gate: effect_gate.clone(),
@@ -569,9 +601,7 @@ impl StandaloneJobs {
             &snapshot.domain_grant,
             self.domain.lock().expect("Job domain poisoned").as_ref(),
         ) {
-            (Some(_), Some(provider)) => provider
+            (Some(_), Some(provider)) => provider.grant(&snapshot, fence)?,
-                .grant(&snapshot, fence)
-                .map_err(JobServiceError::Configuration)?,
             (Some(_), None) => {
                 return Err(JobServiceError::Configuration(
                     "persisted Job requires an unavailable explicit domain capability".into(),
@@ -625,6 +655,11 @@ impl StandaloneJobs {
                 return Err(JobServiceError::Capacity);
             }
         }
+        if tokio::time::Instant::now() >= deadline {
+            return Err(JobServiceError::Configuration(
+                "consolidation_admission_deadline_exceeded during preparation".into(),
+            ));
+        }
         let snapshot = self.with_store_write(|store| {
             store.dispatch(job_id, &snapshot.attempt.attempt_id, self.capacity)
         })?;
@@ -669,7 +704,7 @@ impl StandaloneJobs {
                 result = &mut run => result,
                 reason = async {
                     tokio::select! {
-                        _ = tokio::time::sleep(timeout) => "timeout",
+                        _ = tokio::time::sleep_until(deadline) => "timeout",
                         _ = cancellation.wait_for(|cancelled| *cancelled) => "Host cancellation/shutdown",
                     }
                 } => {

crates/standalone/src/jobs/recovery.rs

@@ -66,7 +66,52 @@ impl StandaloneJobs {
         ))
     }
 
-    async fn recovery_closed(&self) {
+    /// Subscribe before attempting admission; callers must enable the returned
+    /// notification before checking capacity to retain releases racing that check.
+    pub(crate) fn admission_changed(&self) -> tokio::sync::futures::OwnedNotified {
+        self.capacity_changed.clone().notified_owned()
+    }
+
+    /// A completion notification precedes task return. Join only successful
+    /// completions before releasing their slot; an error retains actual ownership.
+    pub(crate) async fn join_completed_admissions(&self) {
+        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 id in completed {
+            let _ = self.wait(&id).await;
+        }
+    }
+
+    /// Exact granted admission only, never a lookup of a future attempt by ID.
+    /// The start lock rechecks input, grant and ALL prior execution cleanup.
+    pub(crate) fn start_reserved_granted(
+        &self,
+        expected: &JobSnapshot,
+        deadline: tokio::time::Instant,
+    ) -> Result<(), JobServiceError> {
+        self.start_with_admission(&expected.request.job_id, None, Some((expected, deadline)))
+    }
+
+    pub(crate) fn reject_reserved_granted(
+        &self,
+        expected: &JobSnapshot,
+        cause: &str,
+    ) -> Result<(), JobServiceError> {
+        let executions = self.executions.lock().expect("Job executions poisoned");
+        if executions.closed {
+            return Err(JobServiceError::Closed);
+        }
+        self.with_store_write(|store| store.fail_reserved_granted(expected, cause))
+    }
+
+    pub(crate) async fn recovery_closed(&self) {
         loop {
             let changed = self.capacity_changed.notified();
             tokio::pin!(changed);
@@ -95,20 +140,7 @@ impl StandaloneJobs {
             let changed = self.capacity_changed.notified();
             tokio::pin!(changed);
             changed.as_mut().enable();
-            let completed: Vec<_> = self
+            self.join_completed_admissions().await;
-                .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,

crates/standalone/src/jobs/tests.rs

@@ -121,9 +121,11 @@ impl JobDomainProvider for RecordingDomainProvider {
         &self,
         snapshot: &JobSnapshot,
         fence: Arc<dyn JobAttemptFence>,
-    ) -> Result<worker::job::JobFeatureGrant, String> {
+    ) -> Result<worker::job::JobFeatureGrant, JobServiceError> {
         if snapshot.domain_grant != Some(domain_grant(&snapshot.request.job_id)) {
-            return Err("foreign or mismatched immutable domain grant".into());
+            return Err(JobServiceError::Configuration(
+                "foreign or mismatched immutable domain grant".into(),
+            ));
         }
         self.0.lock().unwrap().push((snapshot.clone(), fence));
         Ok(worker::job::JobFeatureGrant {
@@ -201,7 +203,7 @@ impl JobDomainProvider for ControlledShutdown {
         &self,
         _: &JobSnapshot,
         _: Arc<dyn JobAttemptFence>,
-    ) -> Result<worker::job::JobFeatureGrant, String> {
+    ) -> Result<worker::job::JobFeatureGrant, JobServiceError> {
         Ok(worker::job::JobFeatureGrant {
             features: worker::feature::FeatureRegistryBuilder::new().with_module(self.clone()),
             satisfied_requirements: vec!["feature.subjektiv"],

crates/standalone/src/subjektiv.rs

@@ -18,6 +18,8 @@
 //! delivery failures for accepted results do not trigger model reexecution. Memory application
 //! and surface repair use shared-domain CAS/receipts.
 
+mod admission;
+use admission::{ConsolidationAdmission, schedule_job};
 mod journal;
 pub(crate) use journal::prepare_worker_deletion;
 
@@ -277,6 +279,7 @@ impl StandaloneSubjects {
             self_weak: weak.clone(),
             journal_dispatch: Mutex::new(false),
             consolidation_recovery: Mutex::new(std::collections::HashMap::new()),
+            consolidation_admissions: Mutex::new(std::collections::HashMap::new()),
             #[cfg(test)]
             registration_before_attribution: Mutex::new(None),
             language: self.memory_language.clone(),
@@ -386,6 +389,18 @@ enum ConsolidationRecoveryError {
 }
 
 impl ConsolidationRecoveryError {
+    fn into_service(self) -> crate::jobs::JobServiceError {
+        use crate::jobs::JobServiceError;
+        match self {
+            Self::Service(error) => error,
+            Self::Domain(subjektiv::api::OperationError::Storage(error))
+            | Self::Writer(subjektiv::journal::JournalError::Storage(error))
+            | Self::Writer(subjektiv::journal::JournalError::Subject(
+                subjektiv::SubjektivError::Storage(error),
+            )) => JobServiceError::DomainStorage(error),
+            error => JobServiceError::Configuration(error.to_string()),
+        }
+    }
     fn retryable_read_or_storage(&self) -> bool {
         use crate::jobs::{JobServiceError, JobStoreError};
         fn sqlite(error: &rusqlite::Error) -> bool {
@@ -413,7 +428,8 @@ impl ConsolidationRecoveryError {
         match self {
             Self::Service(JobServiceError::Store(JobStoreError::Sqlite(error))) => sqlite(error),
             Self::Service(JobServiceError::Store(JobStoreError::Io(error))) => io(error),
-            Self::Service(JobServiceError::Store(JobStoreError::Storage(error)))
+            Self::Service(JobServiceError::DomainStorage(error))
+            | Self::Service(JobServiceError::Store(JobStoreError::Storage(error)))
             | Self::Domain(subjektiv::api::OperationError::Storage(error))
             | Self::Writer(subjektiv::journal::JournalError::Storage(error))
             | Self::Writer(subjektiv::journal::JournalError::Subject(
@@ -433,10 +449,18 @@ enum ConsolidationRecoveryProgress {
     Blocked,
 }
 
+#[derive(Debug, PartialEq, Eq)]
+enum ConsolidationRecovery {
+    Settled,
+    AdmissionRequired,
+    Retained,
+}
+
 struct ConsolidationRecoveryOwner {
     progress: ConsolidationRecoveryProgress,
     passes: usize,
     error: ConsolidationRecoveryError,
+    admission_deadline: tokio::time::Instant,
 }
 
 impl ConsolidationRecoveryOwner {
@@ -463,6 +487,8 @@ pub(crate) struct LocalSubjectHost {
     // grant and its unchanged input remain the durable obligation, not this map.
     consolidation_recovery:
         Mutex<std::collections::HashMap<(String, String, String), ConsolidationRecoveryOwner>>,
+    consolidation_admissions:
+        Mutex<std::collections::HashMap<(String, String, String), ConsolidationAdmission>>,
     #[cfg(test)]
     registration_before_attribution: Mutex<Option<Box<dyn FnOnce() + Send>>>,
     catalog: StandaloneSubjects,
@@ -844,17 +870,26 @@ impl LocalSubjectHost {
                     {
                         continue;
                     }
+                    let admission_deadline = tokio::time::Instant::now()
+                        + std::time::Duration::from_secs(u64::from(
+                            job.request.limits.timeout_seconds,
+                        ));
                     match self.recover_consolidation_job(&jobs, &job) {
-                        Ok(true) => {}
+                        Ok(ConsolidationRecovery::AdmissionRequired) => {
-                        Ok(false)
+                            schedule_job(
-                            if job.state == crate::jobs::JobState::Pending
+                                jobs.clone(),
-                                && job.attempt.state == crate::jobs::JobAttemptState::Reserved
+                                job,
-                                && job.executions_released() =>
+                                self.self_weak.clone(),
-                        {
+                                Some(admission_deadline),
-                            schedule_job(jobs.clone(), job.request.job_id, self.self_weak.clone());
+                            );
                         }
-                        Ok(false) => {} // Missing receipts never authorize replay of effects.
+                        Ok(ConsolidationRecovery::Settled | ConsolidationRecovery::Retained) => {}
-                        Err(error) => self.own_consolidation_recovery(jobs.clone(), job, error),
+                        Err(error) => self.own_consolidation_recovery_until(
+                            jobs.clone(),
+                            job,
+                            error,
+                            admission_deadline,
+                        ),
                     }
                 }
             }
@@ -887,7 +922,7 @@ impl LocalSubjectHost {
         &self,
         jobs: &crate::jobs::StandaloneJobs,
         observed: &crate::jobs::JobSnapshot,
-    ) -> Result<bool, ConsolidationRecoveryError> {
+    ) -> Result<ConsolidationRecovery, ConsolidationRecoveryError> {
         self.ensure_consolidation_live()?;
         let current = jobs.get(&observed.request.job_id)?;
         if current.attempt.attempt_id != observed.attempt.attempt_id
@@ -927,13 +962,22 @@ impl LocalSubjectHost {
                 result,
             })?;
             jobs.acknowledge(&settled.request.job_id)?;
-            return Ok(true);
+            return Ok(ConsolidationRecovery::Settled);
         }
         if current.state != crate::jobs::JobState::Failed
             || current.attempt.state != crate::jobs::JobAttemptState::Failed
             || !current.executions_released()
         {
-            return Ok(false);
+            return Ok(
+                if current.state == crate::jobs::JobState::Pending
+                    && current.attempt.state == crate::jobs::JobAttemptState::Reserved
+                    && current.executions_released()
+                {
+                    ConsolidationRecovery::AdmissionRequired
+                } else {
+                    ConsolidationRecovery::Retained
+                },
+            );
         }
         // This inspection reads original candidate/disposition and preparation
         // records transactionally. Missing preparation is valid; stale preparation,
@@ -974,7 +1018,7 @@ impl LocalSubjectHost {
             &current.attempt.attempt_id,
             &format!("consolidation-abandoned-v1:{reference}"),
         )?;
-        Ok(true)
+        Ok(ConsolidationRecovery::Settled)
     }
 
     /// Retain only typed temporary storage failures of this exact operation.
@@ -986,6 +1030,18 @@ impl LocalSubjectHost {
         jobs: crate::jobs::StandaloneJobs,
         observed: crate::jobs::JobSnapshot,
         error: ConsolidationRecoveryError,
+    ) {
+        let admission_deadline = tokio::time::Instant::now()
+            + std::time::Duration::from_secs(u64::from(observed.request.limits.timeout_seconds));
+        self.own_consolidation_recovery_until(jobs, observed, error, admission_deadline);
+    }
+
+    fn own_consolidation_recovery_until(
+        &self,
+        jobs: crate::jobs::StandaloneJobs,
+        observed: crate::jobs::JobSnapshot,
+        error: ConsolidationRecoveryError,
+        admission_deadline: tokio::time::Instant,
     ) {
         let key = (
             observed.request.job_id.clone(),
@@ -1014,7 +1070,9 @@ impl LocalSubjectHost {
             },
             passes: 1,
             error,
+            admission_deadline,
         };
+        let admission_deadline = owner.admission_deadline;
         eprintln!(
             "standalone consolidation Job {} reconciliation {:?} after {} passes: {}; next action: {}",
             key.0,
@@ -1039,14 +1097,24 @@ impl LocalSubjectHost {
                     retained_delay = (retained_delay * 2).min(std::time::Duration::from_secs(30));
                     delay
                 };
-                tokio::time::sleep(delay).await;
+                tokio::select! {
+                    biased;
+                    _ = jobs.recovery_closed() => {
+                        host.consolidation_recovery.lock().expect("consolidation recovery poisoned").remove(&key);
+                        return;
+                    }
+                    _ = tokio::time::sleep(delay) => {}
+                }
                 pass = pass.saturating_add(1);
                 let result = host.recover_consolidation_job(&jobs, &observed);
                 let mut owners = host.consolidation_recovery.lock().expect("consolidation recovery poisoned");
                 match result {
-                    Ok(_) => {
+                    Ok(outcome) => {
                         owners.remove(&key);
                         drop(owners);
+                        if outcome == ConsolidationRecovery::AdmissionRequired {
+                            schedule_job(jobs.clone(), observed.clone(), host.self_weak.clone(), Some(admission_deadline));
+                        }
                         host.wake_journals();
                         return;
                     }
@@ -1496,7 +1564,7 @@ impl SubjektivHost for LocalSubjectHost {
                     .is_some_and(|g| g.binding == self.binding)
             })
         {
-            schedule_job(jobs, pending.request.job_id, self.self_weak.clone());
+            schedule_job(jobs, pending, self.self_weak.clone(), None);
             let (count, bytes) = self
                 .catalog
                 .store
@@ -1559,7 +1627,7 @@ impl SubjektivHost for LocalSubjectHost {
                     .map_err(|e| SubjektivHostError::Unavailable(e.to_string()))?,
             )
             .map_err(|e| SubjektivHostError::Unavailable(e.to_string()))?;
-        schedule_job(jobs, snapshot.request.job_id, self.self_weak.clone());
+        schedule_job(jobs, snapshot, self.self_weak.clone(), None);
         Ok(memory::backend::MemoryConsolidationOutput { status:"started".into(),summary:"Requested a durable generic standalone consolidation Job; processing stops on Host shutdown".into(),candidate_count:count,total_bytes:bytes })
     }
 }
@@ -1582,15 +1650,21 @@ impl crate::jobs::JobDomainProvider for LocalDomainProvider {
         &self,
         snapshot: &crate::jobs::JobSnapshot,
         fence: Arc<dyn crate::jobs::JobAttemptFence>,
-    ) -> Result<worker::job::JobFeatureGrant, String> {
+    ) -> Result<worker::job::JobFeatureGrant, crate::jobs::JobServiceError> {
-        let host = self.0.upgrade().ok_or("Subject Host unavailable")?;
+        let host = self.0.upgrade().ok_or_else(|| {
-        host.ensure_live().map_err(|e| e.to_string())?;
+            crate::jobs::JobServiceError::Configuration("Subject Host unavailable".into())
+        })?;
         if snapshot.request.purpose == journal::PURPOSE {
-            return journal::grant(host, snapshot, fence);
+            host.ensure_live()
+                .map_err(|error| crate::jobs::JobServiceError::Configuration(error.to_string()))?;
+            return journal::grant(host, snapshot, fence)
+                .map_err(crate::jobs::JobServiceError::Configuration);
         }
+        host.ensure_consolidation_live()
+            .map_err(ConsolidationRecoveryError::into_service)?;
         let grant = host
             .consolidation_grant(snapshot)
-            .map_err(|error| error.to_string())?;
+            .map_err(ConsolidationRecoveryError::into_service)?;
         let adapter = Arc::new(LocalConsolidationHost {
             host,
             grant,
@@ -1605,11 +1679,11 @@ impl crate::jobs::JobDomainProvider for LocalDomainProvider {
                 session_id: format!("job:{}", snapshot.attempt.attempt_id),
             },
         )
-        .map_err(|e| e.to_string())?;
+        .map_err(|error| crate::jobs::JobServiceError::Configuration(error.to_string()))?;
         let feature = worker::feature::builtin::memory::SubjektivConsolidationToolsFeature::new(
             connection.clone(),
         )
-        .map_err(|e| e.to_string())?;
+        .map_err(|error| crate::jobs::JobServiceError::Configuration(error.to_string()))?;
         Ok(worker::job::JobFeatureGrant {
             features: worker::feature::FeatureRegistryBuilder::new().with_module(feature),
             satisfied_requirements: vec!["feature.subjektiv"],
@@ -1815,40 +1889,6 @@ impl worker::job::JobResultFinalizer for LocalConsolidationFinalizer {
     }
 }
 
-fn schedule_job(
-    jobs: crate::jobs::StandaloneJobs,
-    id: String,
-    host: std::sync::Weak<LocalSubjectHost>,
-) {
-    // Scheduling/consumer acknowledgement has no Internal Worker ownership.
-    // The generic service fences and retains every execution until cleanup.
-    tokio::spawn(async move {
-        if let Err(error) = jobs.start(&id).await {
-            eprintln!("standalone Subject Job {id} remains pending: {error}");
-            return;
-        }
-        match jobs.wait(&id).await {
-            Ok(result) if result.state == crate::jobs::JobState::Completed => {
-                let _ = jobs.acknowledge(&id);
-            }
-            Ok(result) if result.state == crate::jobs::JobState::Failed => {
-                if let Some(host) = host.upgrade() {
-                    if let Err(error) = host.recover_consolidation_job(&jobs, &result) {
-                        host.own_consolidation_recovery(jobs.clone(), result, error);
-                    }
-                }
-            }
-            Ok(_) => {}
-            Err(error) => eprintln!(
-                "standalone Subject Job {id} cleanup/acknowledgement is unconfirmed: {error}"
-            ),
-        }
-        if let Some(host) = host.upgrade() {
-            host.wake_journals();
-        }
-    });
-}
-
 #[cfg(test)]
 mod tests;
 

crates/standalone/src/subjektiv/admission.rs

@@ -0,0 +1,203 @@
+//! Ownership of never-dispatched consolidation admissions. The durable obligation
+//! is the original Job reservation, not a new queue or a retry execution attempt.
+use super::*;
+use crate::jobs::{JobServiceError, JobSnapshot, JobState, JobStoreError, StandaloneJobs};
+use std::time::Duration;
+use tokio::time::Instant;
+
+pub(super) struct ConsolidationAdmission {
+    pub(super) deadline: Instant,
+    pub(super) issue: Option<ConsolidationRecoveryError>,
+    pub(super) blocked: bool,
+}
+
+pub(super) fn schedule_job(
+    jobs: StandaloneJobs,
+    observed: JobSnapshot,
+    host: std::sync::Weak<LocalSubjectHost>,
+    deadline: Option<Instant>,
+) {
+    let Some(host) = host.upgrade() else {
+        return;
+    };
+    let key = (
+        observed.request.job_id.clone(),
+        observed.attempt.attempt_id.clone(),
+        observed.input_digest.clone(),
+    );
+    let deadline = deadline.unwrap_or_else(|| {
+        Instant::now() + Duration::from_secs(u64::from(observed.request.limits.timeout_seconds))
+    });
+    // An overlapping restore/request must not refill the earlier storage owner's
+    // admission budget while that exact operation is waiting to hand off.
+    let deadline = host
+        .consolidation_recovery
+        .lock()
+        .expect("consolidation recovery poisoned")
+        .get(&key)
+        .map_or(deadline, |owner| owner.admission_deadline.min(deadline));
+    {
+        let mut owners = host
+            .consolidation_admissions
+            .lock()
+            .expect("consolidation admissions poisoned");
+        if owners.contains_key(&key) {
+            return;
+        }
+        owners.insert(
+            key.clone(),
+            ConsolidationAdmission {
+                deadline,
+                issue: None,
+                blocked: false,
+            },
+        );
+    }
+    tokio::spawn(async move {
+        let outcome = tokio::select! {
+            biased;
+            _ = jobs.recovery_closed() => Ok(false),
+            outcome = host.admit_consolidation(&jobs, &observed, &key) => outcome,
+        };
+        match outcome {
+            Err(error) => {
+                eprintln!(
+                    "standalone consolidation admission {}/{} blocked: {error}; original input retained",
+                    key.0, key.1
+                );
+                let mut owners = host
+                    .consolidation_admissions
+                    .lock()
+                    .expect("consolidation admissions poisoned");
+                let owner = owners.get_mut(&key).expect("owned consolidation admission");
+                owner.issue = Some(error);
+                owner.blocked = true;
+                return;
+            }
+            Ok(started) => {
+                // Dispatch is never retried after this point. Execution/cleanup and
+                // settlement remain owned by the existing generic Job service.
+                if started {
+                    match jobs.wait(&observed.request.job_id).await {
+                        Ok(result) if result.state == JobState::Completed => {
+                            if host.ensure_consolidation_live().is_ok() {
+                                let _ = jobs.acknowledge(&result.request.job_id);
+                            }
+                        }
+                        Ok(result) if result.state == JobState::Failed => {
+                            if let Err(error) = host.recover_consolidation_job(&jobs, &result) {
+                                host.own_consolidation_recovery(jobs.clone(), result, error);
+                            }
+                        }
+                        Ok(_) => {}
+                        Err(error) => eprintln!(
+                            "standalone Subject Job {} cleanup/acknowledgement is unconfirmed: {error}",
+                            key.0
+                        ),
+                    }
+                }
+                host.consolidation_admissions
+                    .lock()
+                    .expect("consolidation admissions poisoned")
+                    .remove(&key);
+                host.wake_journals();
+            }
+        }
+    });
+}
+
+impl LocalSubjectHost {
+    async fn admit_consolidation(
+        &self,
+        jobs: &StandaloneJobs,
+        observed: &JobSnapshot,
+        key: &(String, String, String),
+    ) -> Result<bool, ConsolidationRecoveryError> {
+        let deadline = self
+            .consolidation_admissions
+            .lock()
+            .expect("consolidation admissions poisoned")
+            .get(key)
+            .expect("owned consolidation admission")
+            .deadline;
+        let mut delay = Duration::from_millis(100);
+        // Once preparation genuinely refuses, retry ONLY persistence of that
+        // refusal, not preparation/model execution after an unrelated recovery.
+        let mut refusal: Option<String> = None;
+        loop {
+            let changed = jobs.admission_changed();
+            tokio::pin!(changed);
+            changed.as_mut().enable(); // BEFORE capacity checks, including task return.
+            jobs.join_completed_admissions().await;
+            let attempt = (|| {
+                self.ensure_consolidation_live()?;
+                match self.recover_consolidation_job(jobs, observed)? {
+                    ConsolidationRecovery::AdmissionRequired => {}
+                    ConsolidationRecovery::Settled | ConsolidationRecovery::Retained => {
+                        return Ok(false);
+                    }
+                }
+                if Instant::now() >= deadline {
+                    refusal.get_or_insert_with(|| "consolidation_admission_deadline_exceeded: original reservation deadline elapsed before dispatch".into());
+                }
+                if refusal.is_none() {
+                    match jobs.start_reserved_granted(observed, deadline) {
+                        Ok(()) => return Ok(true),
+                        Err(JobServiceError::Configuration(cause)) => {
+                            refusal =
+                                Some(format!("consolidation_admission_cannot_start: {cause}"));
+                        }
+                        Err(error) => return Err(error.into()),
+                    }
+                }
+                // Fresh exact snapshot, real Host binding, original input/grant and
+                // ALL cleanup were just checked. This write proves no runner cleanup.
+                self.ensure_consolidation_live()?;
+                jobs.reject_reserved_granted(
+                    observed,
+                    refusal.as_deref().expect("admission refusal"),
+                )?;
+                let failed = jobs.get(&observed.request.job_id)?;
+                if let Err(error) = self.recover_consolidation_job(jobs, &failed) {
+                    self.own_consolidation_recovery(jobs.clone(), failed, error);
+                }
+                Ok(false)
+            })();
+            match attempt {
+                Ok(started) => return Ok(started),
+                Err(error) => {
+                    let capacity = matches!(
+                        &error,
+                        ConsolidationRecoveryError::Service(
+                            JobServiceError::Capacity
+                                | JobServiceError::Store(
+                                    JobStoreError::ConcurrentLimit
+                                        | JobStoreError::SerializationConflict(_)
+                                )
+                        )
+                    );
+                    if !capacity && !error.retryable_read_or_storage() {
+                        return Err(error);
+                    }
+                    self.consolidation_admissions
+                        .lock()
+                        .expect("consolidation admissions poisoned")
+                        .get_mut(key)
+                        .expect("owned consolidation admission")
+                        .issue = Some(error);
+                    if capacity {
+                        tokio::select! {
+                            _ = &mut changed => {}
+                            _ = tokio::time::sleep_until(deadline) => {}
+                        }
+                    } else {
+                        // Retry the SAME operation against its actual storage source.
+                        // No healthy-other-store probe and no new dispatch attempt.
+                        tokio::time::sleep(delay).await;
+                        delay = (delay * 2).min(Duration::from_secs(30));
+                    }
+                }
+            }
+        }
+    }
+}

crates/standalone/src/subjektiv/admission_tests.rs

@@ -0,0 +1,701 @@
+//! Real Host admission with real Job/domain SQLite stores. Only model responses
+//! and the held Job's release timing are substituted; no process-death proof.
+use super::*;
+use agen::llm_client::{ClientError, LlmClient, Request, event::Event, types::Item};
+use std::sync::atomic::AtomicUsize;
+use std::time::Duration;
+
+#[derive(Clone, Default)]
+struct AdmissionModel {
+    clones: Arc<AtomicUsize>,
+    entered: Arc<tokio::sync::Notify>,
+    release: Arc<tokio::sync::Notify>,
+    submissions: Arc<Mutex<Vec<String>>>,
+}
+#[async_trait::async_trait]
+impl LlmClient for AdmissionModel {
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        self.clones.fetch_add(1, Ordering::SeqCst);
+        Box::new(self.clone())
+    }
+    async fn stream(
+        &self,
+        request: Request,
+    ) -> Result<agen::llm_client::client::ResponseStream, ClientError> {
+        let envelope = request
+            .items
+            .iter()
+            .filter_map(Item::as_text)
+            .find(|text| text.contains("input_json: "))
+            .expect("immutable Job input");
+        let id = envelope
+            .lines()
+            .find_map(|line| line.strip_prefix("job_id: "))
+            .unwrap();
+        let input: serde_json::Value = serde_json::from_str(
+            envelope
+                .lines()
+                .find_map(|line| line.strip_prefix("input_json: "))
+                .unwrap(),
+        )
+        .unwrap();
+        let events = if request.items.iter().any(|item| {
+            matches!(item,
+            Item::ToolResult { call_id, is_error: false, .. } if call_id == "admission-result")
+        }) {
+            vec![
+                Event::text_block_start(0),
+                Event::text_delta(0, "done"),
+                Event::text_block_stop(0, None),
+            ]
+        } else {
+            {
+                let mut submissions = self.submissions.lock().unwrap();
+                assert!(
+                    !submissions.iter().any(|prior| prior == id),
+                    "duplicate execution/model submission"
+                );
+                submissions.push(id.to_owned());
+            }
+            if id == "held-capacity" {
+                self.entered.notify_one();
+                self.release.notified().await;
+            }
+            vec![
+                Event::tool_use_start(0, "admission-result", "SubmitJobResult"),
+                Event::tool_input_delta(0, &serde_json::json!({"result": input}).to_string()),
+                Event::tool_use_stop(0),
+            ]
+        };
+        Ok(Box::pin(futures::stream::iter(events.into_iter().map(Ok))))
+    }
+}
+
+struct AdmissionFixture {
+    body: SubjectConnection,
+    jobs: crate::jobs::StandaloneJobs,
+    model: AdmissionModel,
+    _temp: tempfile::TempDir,
+}
+impl Drop for AdmissionFixture {
+    fn drop(&mut self) {
+        self.body.host.live.store(false, Ordering::Release);
+    }
+}
+impl AdmissionFixture {
+    fn new() -> Self {
+        let temp = tempfile::tempdir().unwrap();
+        let catalog = StandaloneSubjects::open(temp.path().join("state")).unwrap();
+        let subject = catalog.create("admission", None).unwrap();
+        let body = catalog
+            .connect(&subject.id, protocol::WorkerId::now_v7())
+            .unwrap();
+        let jobs =
+            crate::jobs::StandaloneJobs::open(&body.jobs_path, temp.path().to_path_buf()).unwrap();
+        let model = AdmissionModel::default();
+        jobs.bind_model_client(Box::new(model.clone()));
+        body.host.bind_jobs(jobs.clone()).unwrap();
+        Self {
+            body,
+            jobs,
+            model,
+            _temp: temp,
+        }
+    }
+    fn reserve(&self) -> crate::jobs::JobSnapshot {
+        self.reserve_with_profile("builtin:standalone-subjektiv-consolidation")
+    }
+    fn reserve_with_profile(&self, profile: &str) -> crate::jobs::JobSnapshot {
+        self.jobs.request_granted(job::JobRequest {
+            job_id: "reserved-consolidation".into(), purpose: "subjektiv_consolidation".into(),
+            input_ref: format!("subjektiv://{}/consolidation", self.body.binding.subject_id),
+            input: serde_json::json!({"subject_id": self.body.binding.subject_id, "candidate_ids": []}),
+            instruction: "Return the exact immutable input with SubmitJobResult".into(),
+            profile: profile.into(),
+            serialization_key: Some(format!("subjektiv:{}", self.body.binding.subject_id)),
+            limits: Default::default(),
+        }, serde_json::to_value(LocalConsolidationGrant {
+            binding: self.body.binding.clone(), candidate_ids: vec![],
+        }).unwrap()).unwrap()
+    }
+    fn domain_path(&self) -> PathBuf {
+        self.body
+            .host
+            .catalog
+            .root
+            .join("features")
+            .join(&self.body.host.catalog.scope_id)
+            .join("features/subjektiv.sqlite")
+    }
+    async fn hold_capacity(&self) {
+        self.jobs
+            .request(job::JobRequest {
+                job_id: "held-capacity".into(),
+                purpose: "test".into(),
+                input_ref: "fixture:held".into(),
+                input: serde_json::json!({"held": true}),
+                instruction: "Return the input".into(),
+                profile: "builtin:job".into(),
+                serialization_key: None,
+                limits: job::JobLimits {
+                    max_concurrent_jobs: 1,
+                    ..Default::default()
+                },
+            })
+            .unwrap();
+        self.jobs.start("held-capacity").await.unwrap();
+        tokio::time::timeout(Duration::from_secs(10), self.model.entered.notified())
+            .await
+            .unwrap();
+        let held = self.jobs.get("held-capacity").unwrap();
+        assert_eq!(held.attempt.state, crate::jobs::JobAttemptState::Dispatched);
+        assert!(!held.executions_released());
+    }
+    async fn assert_completed_once(&self, before: &crate::jobs::JobSnapshot) {
+        let id = &before.request.job_id;
+        until(
+            "same reserved consolidation must progress after ONLY fault/capacity release",
+            || {
+                self.jobs.get(id).is_ok_and(|saved| {
+                    saved.state == crate::jobs::JobState::Completed
+                        && saved.acknowledged
+                        && saved.executions_released()
+                })
+            },
+        )
+        .await;
+        let after = self.jobs.wait(id).await.unwrap();
+        assert_eq!(after.request, before.request);
+        assert_eq!(after.input_digest, before.input_digest);
+        assert_eq!(after.domain_grant, before.domain_grant);
+        assert_eq!(after.attempt.attempt_id, before.attempt.attempt_id);
+        assert_eq!(after.attempt.number, before.attempt.number);
+        assert_eq!(after.executions.len(), 1);
+        assert!(after.result.as_ref().unwrap().get("surface").is_some());
+        assert_eq!(
+            self.model
+                .submissions
+                .lock()
+                .unwrap()
+                .iter()
+                .filter(|prior| *prior == id)
+                .count(),
+            1
+        );
+        let job_db = rusqlite::Connection::open(&self.body.jobs_path).unwrap();
+        let attempts: i64 = job_db
+            .query_row(
+                "SELECT count(*) FROM job_attempts WHERE job_id=?1",
+                [id],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(attempts, 1);
+        let domain = rusqlite::Connection::open(self.domain_path()).unwrap();
+        let generations: i64 = domain
+            .query_row(
+                "SELECT count(*) FROM surface_generation_runs WHERE job_id=?1 AND attempt_id=?2",
+                rusqlite::params![id, before.attempt.attempt_id],
+                |row| row.get(0),
+            )
+            .unwrap();
+        assert_eq!(generations, 1, "finalizer must prepare exactly once");
+        assert_eq!(
+            subjektiv::job::lookup_result(
+                &self.body.host.catalog.store,
+                &self.body.binding.subject_id,
+                &[],
+                id,
+                &before.attempt.attempt_id
+            )
+            .unwrap(),
+            after.result
+        );
+    }
+}
+async fn until(message: &str, mut check: impl FnMut() -> bool) {
+    tokio::time::timeout(Duration::from_secs(10), async {
+        while !check() {
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await
+    .expect(message);
+}
+
+#[tokio::test]
+async fn consolidation_admission_capacity_only_release_starts_original_request() {
+    capacity_only_release(false).await;
+}
+#[tokio::test]
+async fn consolidation_admission_capacity_only_release_starts_restored_reservation() {
+    capacity_only_release(true).await;
+}
+async fn capacity_only_release(restore: bool) {
+    let fixture = AdmissionFixture::new();
+    fixture.hold_capacity().await;
+    let clones = fixture.model.clones.load(Ordering::SeqCst);
+    let before = if restore {
+        let before = fixture.reserve();
+        fixture.body.host.recover_jobs();
+        before
+    } else {
+        let result = fixture
+            .body
+            .host
+            .request_consolidation(
+                &SubjektivHostContext {
+                    worker_id: fixture.body.host.worker_key.clone(),
+                    session_id: session_store::new_session_id().to_string(),
+                },
+                memory::backend::MemoryConsolidateStagingOperation { force: true },
+            )
+            .unwrap();
+        assert_eq!(result.status, "started");
+        fixture
+            .jobs
+            .pending()
+            .unwrap()
+            .into_iter()
+            .find(|saved| saved.domain_grant.is_some())
+            .unwrap()
+    };
+    // Preparation clones the configured transport before checking actual capacity.
+    // This barrier proves the production scheduler attempted admission while held.
+    until("scheduler must reach capacity check", || {
+        fixture.model.clones.load(Ordering::SeqCst) > clones
+    })
+    .await;
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    assert!(before.executions.is_empty());
+    assert!(matches!(
+        fixture.jobs.start(&before.request.job_id).await,
+        Err(crate::jobs::JobServiceError::Capacity)
+    ));
+    fixture.model.release.notify_one();
+    let held = tokio::time::timeout(Duration::from_secs(10), fixture.jobs.wait("held-capacity"))
+        .await
+        .unwrap()
+        .unwrap();
+    assert!(held.executions_released());
+    fixture.assert_completed_once(&before).await;
+}
+
+#[tokio::test]
+async fn consolidation_admission_reserved_domain_busy_release_before_fast_budget() {
+    reserved_domain_busy_release(false).await;
+}
+#[tokio::test]
+async fn consolidation_admission_reserved_domain_busy_release_beyond_fast_budget() {
+    reserved_domain_busy_release(true).await;
+}
+async fn reserved_domain_busy_release(beyond: bool) {
+    let fixture = AdmissionFixture::new();
+    let before = fixture.reserve();
+    let connection = rusqlite::Connection::open(fixture.domain_path()).unwrap();
+    connection.execute_batch("BEGIN IMMEDIATE").unwrap();
+    // Production restore entry, not a manually armed retry or failed-attempt fixture.
+    fixture.body.host.recover_jobs();
+    let required_passes = if beyond { 8 } else { 1 };
+    // Real SQLite busy_timeout makes eight attempts longer than the short wait.
+    tokio::time::timeout(Duration::from_secs(60), async {
+        loop {
+            let owners = fixture.body.host.consolidation_recovery.lock().unwrap();
+            if let Some(owner) = owners
+                .values()
+                .find(|owner| owner.passes >= required_passes)
+            {
+                super::assert_real_busy(&owner.error, true);
+                if beyond {
+                    assert_eq!(
+                        owner.progress,
+                        ConsolidationRecoveryProgress::RetainedStorage
+                    );
+                }
+                break;
+            }
+            drop(owners);
+            tokio::time::sleep(Duration::from_millis(5)).await;
+        }
+    })
+    .await
+    .expect("restore must own the actual reserved-domain Busy operation");
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    assert!(fixture.model.submissions.lock().unwrap().is_empty());
+    eprintln!(
+        "D2 actual reserved Feature Busy retained; beyond_fast_budget={beyond}; rollback ONLY"
+    );
+    connection.execute_batch("ROLLBACK").unwrap();
+    // No restore/recover_jobs/reclick/new request after releasing the SQLite fault.
+    fixture.assert_completed_once(&before).await;
+}
+
+async fn waiting_capacity(fixture: &AdmissionFixture) {
+    until(
+        "exact admission must own the actual capacity refusal",
+        || {
+            fixture
+                .body
+                .host
+                .consolidation_admissions
+                .lock()
+                .unwrap()
+                .values()
+                .any(|owner| {
+                    matches!(
+                        owner.issue,
+                        Some(ConsolidationRecoveryError::Service(
+                            crate::jobs::JobServiceError::Capacity
+                        ))
+                    )
+                })
+        },
+    )
+    .await;
+}
+
+#[tokio::test]
+async fn consolidation_admission_duplicate_notifications_preserve_deadline_and_never_forge_cleanup()
+{
+    let fixture = AdmissionFixture::new();
+    fixture.hold_capacity().await;
+    let before = fixture.reserve();
+    let deadline = tokio::time::Instant::now() + Duration::from_millis(300);
+    schedule_job(
+        fixture.jobs.clone(),
+        before.clone(),
+        fixture.body.host.self_weak.clone(),
+        Some(deadline),
+    );
+    waiting_capacity(&fixture).await;
+    // A late notification carrying a fresh timeout must not reset the original.
+    schedule_job(
+        fixture.jobs.clone(),
+        before.clone(),
+        fixture.body.host.self_weak.clone(),
+        Some(deadline + Duration::from_secs(60)),
+    );
+    {
+        let owners = fixture.body.host.consolidation_admissions.lock().unwrap();
+        assert_eq!(owners.len(), 1);
+        assert_eq!(owners.values().next().unwrap().deadline, deadline);
+    }
+    until(
+        "the original admission deadline must persist a specific pre-dispatch failure",
+        || {
+            fixture.jobs.get(&before.request.job_id).unwrap().state == crate::jobs::JobState::Failed
+                && fixture
+                    .body
+                    .host
+                    .consolidation_admissions
+                    .lock()
+                    .unwrap()
+                    .is_empty()
+        },
+    )
+    .await;
+    let failed = fixture.jobs.get(&before.request.job_id).unwrap();
+    assert!(
+        failed
+            .attempt
+            .failure
+            .as_deref()
+            .unwrap()
+            .contains("consolidation_admission_deadline_exceeded")
+    );
+    assert_eq!(failed.attempt.attempt_id, before.attempt.attempt_id);
+    assert_eq!(failed.domain_grant, before.domain_grant);
+    assert_eq!(failed.input_digest, before.input_digest);
+    assert!(failed.executions.is_empty());
+    assert!(failed.start_rejected_by.is_some());
+    assert!(
+        !fixture
+            .jobs
+            .get("held-capacity")
+            .unwrap()
+            .executions_released(),
+        "refusal must not invent prior cleanup"
+    );
+    assert_eq!(
+        *fixture.model.submissions.lock().unwrap(),
+        ["held-capacity"]
+    );
+    fixture.model.release.notify_one();
+    assert!(
+        fixture
+            .jobs
+            .wait("held-capacity")
+            .await
+            .unwrap()
+            .executions_released()
+    );
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), failed);
+}
+
+#[tokio::test]
+async fn consolidation_admission_capacity_wait_rechecks_retirement_binding_and_version_stop() {
+    for case in ["erased", "version_stopped", "inactive", "released_writer"] {
+        let fixture = AdmissionFixture::new();
+        fixture.hold_capacity().await;
+        let before = fixture.reserve();
+        fixture.body.host.recover_jobs();
+        waiting_capacity(&fixture).await;
+        match case {
+            "erased" => {
+                fixture
+                    .jobs
+                    .reclaim_input(
+                        &before.request.job_id,
+                        &before.input_digest,
+                        crate::jobs::JobInputRetirementReason::ExplicitErasure,
+                    )
+                    .unwrap();
+            }
+            "version_stopped" => {
+                rusqlite::Connection::open(&fixture.body.jobs_path)
+                    .unwrap()
+                    .execute(
+                        "INSERT INTO job_version_stops(job_id) VALUES(?1)",
+                        [&before.request.job_id],
+                    )
+                    .unwrap();
+            }
+            "inactive" => fixture.body.host.live.store(false, Ordering::Release),
+            "released_writer" => fixture
+                .body
+                .host
+                .catalog
+                .store
+                .release_journal_writer(&fixture.body.host.writer)
+                .unwrap(),
+            _ => unreachable!(),
+        }
+        let expected = fixture.jobs.get(&before.request.job_id).unwrap();
+        fixture.model.release.notify_one();
+        fixture.jobs.wait("held-capacity").await.unwrap();
+        until(
+            "capacity wake must retain the specific refusal without dispatch",
+            || {
+                fixture
+                    .body
+                    .host
+                    .consolidation_admissions
+                    .lock()
+                    .unwrap()
+                    .values()
+                    .any(|owner| owner.blocked)
+            },
+        )
+        .await;
+        assert_eq!(
+            fixture.jobs.get(&before.request.job_id).unwrap(),
+            expected,
+            "{case}"
+        );
+        assert!(expected.executions.is_empty(), "{case}");
+        assert_eq!(
+            *fixture.model.submissions.lock().unwrap(),
+            ["held-capacity"],
+            "{case}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn consolidation_admission_stale_attempt_input_or_grant_never_starts_current_reservation() {
+    for case in ["attempt", "digest", "grant"] {
+        let fixture = AdmissionFixture::new();
+        let before = fixture.reserve();
+        let mut stale = before.clone();
+        match case {
+            "attempt" => stale.attempt.attempt_id = "different-attempt".into(),
+            "digest" => stale.input_digest = "different-digest".into(),
+            "grant" => stale.domain_grant = Some(serde_json::json!({"foreign":true})),
+            _ => unreachable!(),
+        }
+        schedule_job(
+            fixture.jobs.clone(),
+            stale,
+            fixture.body.host.self_weak.clone(),
+            None,
+        );
+        until(
+            "stale exact admission must be diagnosed, never followed by ID",
+            || {
+                fixture
+                    .body
+                    .host
+                    .consolidation_admissions
+                    .lock()
+                    .unwrap()
+                    .values()
+                    .any(|owner| owner.blocked)
+            },
+        )
+        .await;
+        assert_eq!(
+            fixture.jobs.get(&before.request.job_id).unwrap(),
+            before,
+            "{case}"
+        );
+        assert!(
+            fixture.model.submissions.lock().unwrap().is_empty(),
+            "{case}"
+        );
+    }
+}
+
+#[tokio::test]
+async fn consolidation_admission_shutdown_releases_waiter_without_dispatch_or_cleanup_claim() {
+    let fixture = AdmissionFixture::new();
+    fixture.hold_capacity().await;
+    let before = fixture.reserve();
+    fixture.body.host.recover_jobs();
+    waiting_capacity(&fixture).await;
+    // Real service shutdown closes admission and signals the actual held runner.
+    fixture.jobs.close();
+    until("shutdown must release its volatile admission owner", || {
+        fixture
+            .body
+            .host
+            .consolidation_admissions
+            .lock()
+            .unwrap()
+            .is_empty()
+    })
+    .await;
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    assert_eq!(
+        *fixture.model.submissions.lock().unwrap(),
+        ["held-capacity"]
+    );
+    fixture.model.release.notify_one();
+    tokio::time::timeout(Duration::from_secs(10), fixture.jobs.wait("held-capacity"))
+        .await
+        .unwrap()
+        .unwrap();
+}
+
+#[tokio::test]
+async fn consolidation_admission_permanent_preparation_refusal_is_saved_without_execution() {
+    let fixture = AdmissionFixture::new();
+    // A real existing Profile that does not satisfy consolidation's explicit
+    // policy requirement; no fake transport/storage error stands in for refusal.
+    let before = fixture.reserve_with_profile("builtin:job");
+    fixture.body.host.recover_jobs();
+    until(
+        "permanent preparation refusal must become a specific durable failure",
+        || {
+            fixture.jobs.get(&before.request.job_id).unwrap().state == crate::jobs::JobState::Failed
+                && fixture
+                    .body
+                    .host
+                    .consolidation_admissions
+                    .lock()
+                    .unwrap()
+                    .is_empty()
+        },
+    )
+    .await;
+    let failed = fixture.jobs.get(&before.request.job_id).unwrap();
+    assert!(
+        failed
+            .attempt
+            .failure
+            .as_deref()
+            .unwrap()
+            .contains("consolidation_admission_cannot_start")
+    );
+    assert!(failed.start_rejected_by.is_some());
+    assert!(failed.executions.is_empty());
+    assert_eq!(failed.request, before.request);
+    assert_eq!(failed.domain_grant, before.domain_grant);
+    assert_eq!(failed.input_digest, before.input_digest);
+    assert_eq!(failed.attempt.attempt_id, before.attempt.attempt_id);
+    assert!(fixture.model.submissions.lock().unwrap().is_empty());
+    assert!(
+        crate::job_store::JobStore::open(&fixture.body.jobs_path)
+            .unwrap()
+            .execution_obligations()
+            .unwrap()
+            .is_empty()
+    );
+}
+
+#[tokio::test]
+async fn consolidation_admission_reserved_storage_handoff_preserves_deadline_while_capacity_held() {
+    let fixture = AdmissionFixture::new();
+    fixture.hold_capacity().await;
+    let before = fixture.reserve();
+    let connection = rusqlite::Connection::open(fixture.domain_path()).unwrap();
+    connection.execute_batch("BEGIN IMMEDIATE").unwrap();
+    fixture.body.host.recover_jobs();
+    let deadline = {
+        let owners = fixture.body.host.consolidation_recovery.lock().unwrap();
+        assert_eq!(owners.len(), 1);
+        let owner = owners.values().next().unwrap();
+        super::assert_real_busy(&owner.error, true);
+        owner.admission_deadline
+    };
+    connection.execute_batch("ROLLBACK").unwrap();
+    waiting_capacity(&fixture).await;
+    {
+        let owners = fixture.body.host.consolidation_admissions.lock().unwrap();
+        assert_eq!(owners.len(), 1);
+        assert_eq!(owners.values().next().unwrap().deadline, deadline);
+    }
+    // Duplicate restoration while capacity remains held cannot reset the budget.
+    fixture.body.host.recover_jobs();
+    assert_eq!(
+        fixture
+            .body
+            .host
+            .consolidation_admissions
+            .lock()
+            .unwrap()
+            .values()
+            .next()
+            .unwrap()
+            .deadline,
+        deadline
+    );
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    fixture.model.release.notify_one();
+    fixture.jobs.wait("held-capacity").await.unwrap();
+    fixture.assert_completed_once(&before).await;
+}
+
+#[tokio::test]
+async fn consolidation_admission_missing_prior_cleanup_never_dispatches_or_creates_release_proof() {
+    let fixture = super::ConsolidationRecoveryFixture::new();
+    let before = fixture.saved.clone();
+    assert!(!before.executions_released());
+    schedule_job(
+        fixture.jobs.clone(),
+        before.clone(),
+        fixture.body.host.self_weak.clone(),
+        None,
+    );
+    until(
+        "non-reserved work must leave admission without replay",
+        || {
+            fixture
+                .body
+                .host
+                .consolidation_admissions
+                .lock()
+                .unwrap()
+                .is_empty()
+        },
+    )
+    .await;
+    assert_eq!(fixture.jobs.get(&before.request.job_id).unwrap(), before);
+    assert!(fixture.domain_closure().is_none());
+    assert!(
+        !fixture
+            .jobs
+            .get(&before.request.job_id)
+            .unwrap()
+            .executions_released()
+    );
+}

crates/standalone/src/subjektiv/tests.rs

@@ -4,6 +4,9 @@
 //! cleanup receipt, not the domain lookup. No test here proves an OS crash/drain.
 use super::*;
 
+#[path = "admission_tests.rs"]
+mod admission_tests;
+
 struct ConsolidationRecoveryFixture {
     body: SubjectConnection,
     jobs: crate::jobs::StandaloneJobs,
@@ -279,12 +282,13 @@ async fn failed_consolidation_closes_only_attempt_and_preserves_unresolved_input
             memory
         );
         // Stable exact replay; a reference does not mean all candidate work is done.
-        assert!(
+        assert_eq!(
             fixture
                 .body
                 .host
                 .recover_consolidation_job(&fixture.jobs, &before)
-                .unwrap()
+                .unwrap(),
+            ConsolidationRecovery::Settled
         );
         assert_eq!(fixture.domain_closure(), Some(receipt));
         let reopened = crate::job_store::JobStore::open(&fixture.body.jobs_path).unwrap();
@@ -356,7 +360,10 @@ async fn failed_consolidation_retains_unreleased_stale_inactive_migrated_and_era
             .body
             .host
             .recover_consolidation_job(&fixture.jobs, &observed);
-        assert!(!matches!(outcome, Ok(true)), "{case}");
+        assert!(
+            !matches!(outcome, Ok(ConsolidationRecovery::Settled)),
+            "{case}"
+        );
         assert!(fixture.domain_closure().is_none(), "{case}");
         assert_eq!(
             fixture.jobs.get(&fixture.saved.request.job_id).unwrap(),