fix: retain Host-bound Job input authority through every model supply

Bind immutable Job kind/id/input digest to retained input provenance. Recheck current Host authority after hooks and at each compactor request; restored or rewritten input cannot become a local grant or switch Job identities. Add actual Controller, lifecycle, correction and signed Host/model regressions for Journal and consolidation.

crates/worker/src/feature/builtin/backend_job_input.rs

@@ -1,4 +1,9 @@
-//! Rechecks frozen Subject Job input before each model request, including restore.
+//! Early rejection of frozen Subject Job input in the normal request hook chain.
+//!
+//! This hook cannot authorize later hook output or compactor requests. The
+//! Worker's retained-input guard independently binds the immutable Job/input
+//! identity and rechecks current Host authority after hooks and at every
+//! compactor dispatch, including restored and rewritten Sessions.
 use std::sync::Arc;
 
 use async_trait::async_trait;

crates/worker/src/worker.rs

@@ -3607,9 +3607,29 @@ enum ProtectedInputKind {
     Journal,
     JournalStartup,
     JournalCompaction,
+    SubjectConsolidationJob,
+    SubjectJournalJob,
 }
 
 impl ProtectedInputKind {
+    fn for_job(binding: &crate::BackendJobExecutionBinding) -> Option<Self> {
+        if binding.subjektiv_consolidation {
+            Some(Self::SubjectConsolidationJob)
+        } else if binding.subjektiv_journal {
+            Some(Self::SubjectJournalJob)
+        } else {
+            None
+        }
+    }
+
+    fn job_operation(self) -> Option<&'static str> {
+        match self {
+            Self::SubjectConsolidationJob => Some("memory.consolidation.supply"),
+            Self::SubjectJournalJob => Some("journal.job.input.supply"),
+            _ => None,
+        }
+    }
+
     fn parse(kind: &str) -> Result<Self, String> {
         match kind {
             "workspace-memory" => Ok(Self::WorkspaceMemory),
@@ -3617,6 +3637,8 @@ impl ProtectedInputKind {
             "journal" => Ok(Self::Journal),
             "journal-startup" => Ok(Self::JournalStartup),
             "journal-compaction" => Ok(Self::JournalCompaction),
+            "subject-consolidation-job" => Ok(Self::SubjectConsolidationJob),
+            "subject-journal-job" => Ok(Self::SubjectJournalJob),
             _ => Err("unknown protected input kind".into()),
         }
     }
@@ -3628,6 +3650,8 @@ impl ProtectedInputKind {
             Self::Journal => "journal",
             Self::JournalStartup => "journal-startup",
             Self::JournalCompaction => "journal-compaction",
+            Self::SubjectConsolidationJob => "subject-consolidation-job",
+            Self::SubjectJournalJob => "subject-journal-job",
         }
     }
 }
@@ -3662,11 +3686,33 @@ impl JournalCompactionRetention {
     }
 }
 
+// Input identity is retained across rewrites, not execution authority. Attempt
+// identity remains a live Host fence; a retry may bind a new attempt for this
+// same immutable Job/input, never a different Job's current Subject permission.
+#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ProtectedSubjectJobInput {
+    kind: ProtectedInputKind,
+    job_id: String,
+    input_digest: Option<String>,
+}
+impl ProtectedSubjectJobInput {
+    fn from_binding(binding: &crate::BackendJobExecutionBinding) -> Option<Self> {
+        ProtectedInputKind::for_job(binding).map(|kind| Self {
+            kind,
+            job_id: binding.job_id.clone(),
+            input_digest: binding.input_digest.clone(),
+        })
+    }
+}
+
 #[derive(Debug, Clone, serde::Serialize, serde::Deserialize)]
 #[serde(deny_unknown_fields)]
 struct ProtectedInputProvenance {
     version: u8,
     kinds: std::collections::BTreeSet<ProtectedInputKind>,
+    #[serde(default, skip_serializing_if = "std::collections::BTreeSet::is_empty")]
+    subject_jobs: std::collections::BTreeSet<ProtectedSubjectJobInput>,
 }
 
 impl Default for ProtectedInputProvenance {
@@ -3674,6 +3720,7 @@ impl Default for ProtectedInputProvenance {
         Self {
             version: 1,
             kinds: Default::default(),
+            subject_jobs: Default::default(),
         }
     }
 }
@@ -3778,8 +3825,25 @@ impl ProtectedInputProvenance {
                     "unsupported protected input provenance version".into(),
                 ));
             }
+            if saved.kinds.iter().any(|kind| {
+                kind.job_operation().is_some()
+                    && !saved.subject_jobs.iter().any(|job| job.kind == *kind)
+            }) || saved.subject_jobs.iter().any(|job| {
+                job.kind.job_operation().is_none()
+                    || !saved.kinds.contains(&job.kind)
+                    || job.job_id.trim().is_empty()
+                    || job
+                        .input_digest
+                        .as_ref()
+                        .is_some_and(|digest| digest.trim().is_empty())
+            }) {
+                return Err(WorkerError::InvalidState(
+                    "invalid protected Job input provenance".into(),
+                ));
+            }
             found = true;
             provenance.kinds.extend(saved.kinds);
+            provenance.subject_jobs.extend(saved.subject_jobs);
             Ok(())
         };
         for entry in entries {
@@ -3827,6 +3891,9 @@ pub(crate) struct ProtectedInputSupply {
     retained: Arc<Mutex<ProtectedInputProvenance>>,
     prompt_source: Option<Arc<dyn SystemPromptContributionSource>>,
     context_source: Option<Arc<dyn SessionContextSource>>,
+    // Reconstructed only from the immutable Host execution binding, never from
+    // saved provenance. Every use asks the Host for current attempt/origin/policy.
+    subject_job: Option<(ProtectedSubjectJobInput, Arc<dyn WorkspaceClient>)>,
 }
 
 impl ProtectedInputSupply {
@@ -3841,15 +3908,15 @@ impl ProtectedInputSupply {
         &self,
         history: impl IntoIterator<Item = &'a SessionHistoryMetadata>,
     ) -> Result<(), String> {
-        let kinds = {
+        let provenance = {
             let mut retained = self
                 .retained
                 .lock()
                 .expect("protected input provenance poisoned");
             retained.observe_history(history);
-            retained.kinds.clone()
+            retained.clone()
         };
-        self.authorize_kinds(kinds).await
+        self.authorize_retained(provenance).await
     }
 
     pub(crate) async fn authorize_pending<'a>(
@@ -3864,14 +3931,38 @@ impl ProtectedInputSupply {
             .expect("protected input provenance poisoned")
             .clone();
         pending.observe_history(history);
-        self.authorize_kinds(pending.kinds).await
+        self.authorize_retained(pending).await
     }
 
-    async fn authorize_kinds(
-        &self,
-        kinds: std::collections::BTreeSet<ProtectedInputKind>,
-    ) -> Result<(), String> {
-        for kind in kinds {
+    async fn authorize_retained(&self, provenance: ProtectedInputProvenance) -> Result<(), String> {
+        for input in &provenance.subject_jobs {
+            if self.subject_job.as_ref().map(|(bound, _)| bound) != Some(input) {
+                return Err(
+                    "retained protected Job input requires its matching Host Job/input binding"
+                        .into(),
+                );
+            }
+        }
+        for kind in provenance.kinds {
+            if let Some(operation) = kind.job_operation() {
+                let (bound_kind, client) = self.subject_job.as_ref().ok_or_else(|| {
+                    format!(
+                        "retained protected input {} requires its Host Job binding",
+                        kind.as_str()
+                    )
+                })?;
+                if bound_kind.kind != kind || !provenance.subject_jobs.contains(bound_kind) {
+                    return Err(
+                        "retained protected Job input does not match its Host Job binding".into(),
+                    );
+                }
+                client
+                    .authorize_subject_operation(operation)
+                    .map_err(|error| {
+                        format!("protected Job input supply rejected for {operation}: {error}")
+                    })?;
+                continue;
+            }
             let journal = matches!(
                 kind,
                 ProtectedInputKind::Journal
@@ -4830,6 +4921,10 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
             retained: self.protected_inputs.clone(),
             prompt_source: self.feature_resident_summary_source.clone(),
             context_source: self.session_context_source.clone(),
+            subject_job: self
+                .backend_job()
+                .and_then(ProtectedSubjectJobInput::from_binding)
+                .map(|input| (input, self.workspace_client_handle())),
         }
     }
 
@@ -4943,9 +5038,21 @@ impl<C: LlmClient + 'static, St: Store> Worker<C, St> {
         {
             return Err("Backend Job capability requires Backend Workspace API authority");
         }
+        let retained_input = ProtectedSubjectJobInput::from_binding(&binding);
         self.backend_job
             .set(Some(binding))
-            .map_err(|_| "Backend Job capability is already bound")
+            .map_err(|_| "Backend Job capability is already bound")?;
+        // A retention obligation, not a grant. Persist it with every head/rewrite
+        // so removing Feature configuration or the binding cannot expose input.
+        if let Some(input) = retained_input {
+            let mut retained = self
+                .protected_inputs
+                .lock()
+                .expect("protected input provenance poisoned");
+            retained.kinds.insert(input.kind);
+            retained.subject_jobs.insert(input);
+        }
+        Ok(())
     }
 
     /// The Worker's manifest.
@@ -20900,6 +21007,10 @@ permission = "write"
     // Real initial-seed/replay/fork/compaction and Memory source wiring. Only the
     // Host HTTP boundary and provider transport are fixtures; Host policy evaluation
     // itself is covered by workspace-server authorization tests, not asserted here.
+    mod subject_job_input_tests {
+        include!("worker_job_input_supply_tests.rs");
+    }
+
     mod protected_input_tests {
         fn assert_supply_outcome(outcome: Result<WorkerRunResult, WorkerError>, allowed: bool) {
             if allowed {

crates/worker/src/worker_job_input_supply_tests.rs

@@ -0,0 +1,363 @@
+use super::*;
+use agen::llm_client::event::{Event, ResponseStatus, StatusEvent};
+
+type TestStore = session_store::CombinedStore<session_store::FsStore, session_store::FsWorkerStore>;
+const SECRET: &str = "retained-frozen-subject-job-input";
+
+#[derive(Debug)]
+struct Host {
+    operation: &'static str,
+    status: AtomicUsize,
+    checks: AtomicUsize,
+}
+impl Host {
+    fn new(consolidation: bool) -> Self {
+        Self {
+            operation: if consolidation {
+                "memory.consolidation.supply"
+            } else {
+                "journal.job.input.supply"
+            },
+            status: AtomicUsize::new(200),
+            checks: AtomicUsize::new(0),
+        }
+    }
+}
+impl WorkspaceClient for Host {
+    fn workspace_id(&self) -> Option<&str> {
+        Some("workspace-test")
+    }
+    fn kind(&self) -> &str {
+        "retained-job-host"
+    }
+    fn is_available(&self) -> bool {
+        true
+    }
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        assert_eq!(
+            request.path,
+            "/api/w/workspace-test/authorization/bound-check"
+        );
+        let input: server_api::authorization::BoundOperationAuthorizationRequest =
+            serde_json::from_str(request.body.as_deref().unwrap()).unwrap();
+        assert_eq!(input.operation, self.operation);
+        assert_eq!(
+            input.target,
+            server_api::authorization::BoundAuthorizationTarget::Subject {}
+        );
+        self.checks.fetch_add(1, Ordering::SeqCst);
+        Ok(WorkspaceResponse { status: self.status.load(Ordering::SeqCst) as u16,
+            body: r#"{"scope":{"kind":"subject","workspace_id":"workspace-test","subject_id":"subject"}}"#.into() })
+    }
+}
+
+#[derive(Clone, Default)]
+struct Provider {
+    requests: Arc<Mutex<Vec<agen::llm_client::Request>>>,
+    compactor_calls: Arc<AtomicUsize>,
+    revoke_on_compact: Option<Arc<Host>>,
+    invalid_first_summary: bool,
+}
+#[async_trait]
+impl LlmClient for Provider {
+    async fn stream(
+        &self,
+        request: agen::llm_client::Request,
+    ) -> Result<agen::llm_client::ResponseStream, agen::llm_client::ClientError> {
+        let compact = request
+            .tools
+            .iter()
+            .any(|tool| tool.name == "write_summary");
+        self.requests.lock().unwrap().push(request);
+        let events = if compact && self.compactor_calls.fetch_add(1, Ordering::SeqCst) % 2 == 0 {
+            if let Some(host) = &self.revoke_on_compact {
+                host.status.store(403, Ordering::SeqCst);
+            }
+            vec![
+                Event::tool_use_start(0, "summary", "write_summary"),
+                Event::tool_input_delta(
+                    0,
+                    if self.invalid_first_summary {
+                        serde_json::json!({"text":5}).to_string()
+                    } else {
+                        serde_json::json!({"text":format!("Retained {SECRET}")}).to_string()
+                    },
+                ),
+                Event::tool_use_stop(0),
+            ]
+        } else {
+            vec![
+                Event::text_block_start(0),
+                Event::text_delta(0, "done"),
+                Event::text_block_stop(0, None),
+            ]
+        };
+        Ok(Box::pin(futures::stream::iter(
+            events
+                .into_iter()
+                .chain(std::iter::once(Event::Status(StatusEvent {
+                    status: ResponseStatus::Completed,
+                })))
+                .map(Ok),
+        )))
+    }
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+}
+
+fn binding(consolidation: bool) -> crate::BackendJobExecutionBinding {
+    crate::BackendJobExecutionBinding {
+        job_id: "bound-job".into(),
+        attempt_id: "bound-attempt".into(),
+        input_digest: Some("digest".into()),
+        subjektiv_consolidation: consolidation,
+        subjektiv_journal: !consolidation,
+    }
+}
+fn store(temp: &tempfile::TempDir) -> TestStore {
+    session_store::CombinedStore::new(
+        session_store::FsStore::new(temp.path().join("sessions")).unwrap(),
+        session_store::FsWorkerStore::new(temp.path().join("workers")).unwrap(),
+    )
+}
+fn manifest() -> WorkerManifest {
+    let mut manifest = minimal_manifest();
+    manifest.worker.name = format!("retained-job-{}", uuid::Uuid::now_v7());
+    manifest.scope.allow.clear();
+    manifest.compaction = Some(manifest::CompactionConfig {
+        model: None,
+        retained_tokens: 1,
+        ..Default::default()
+    });
+    manifest
+}
+async fn boot(
+    manifest: WorkerManifest,
+    store: TestStore,
+    host: Arc<Host>,
+    provider: Provider,
+    consolidation: bool,
+) -> Worker<Provider, TestStore> {
+    let mut worker = Worker::new(
+        manifest,
+        Engine::<_, Mutable, SessionHistoryMetadata>::new_annotated(provider),
+        store,
+        WorkerWorkspaceContext::with_client(None, host),
+        WorkerFilesystemAuthority::None,
+        Scope::empty(),
+    )
+    .await
+    .unwrap();
+    worker.bind_backend_job(binding(consolidation)).unwrap();
+    worker
+        .engine_mut()
+        .set_system_prompt("Host-bound frozen Job input");
+    worker
+}
+async fn restore(
+    manifest: WorkerManifest,
+    store: TestStore,
+    session: SessionId,
+    segment: SegmentId,
+    host: Arc<Host>,
+    provider: Provider,
+) -> Worker<Box<dyn LlmClient>, TestStore> {
+    Worker::<Box<dyn LlmClient>, _>::restore_from_manifest_with_context_and_model_client(
+        session,
+        segment,
+        manifest,
+        store,
+        PromptCatalogSource::builtins_only(),
+        WorkerWorkspaceContext::with_client(None, host),
+        WorkerFilesystemAuthority::None,
+        Some(Box::new(provider)),
+    )
+    .await
+    .unwrap()
+}
+fn assert_retained(store: &TestStore, session: SessionId, segment: SegmentId, consolidation: bool) {
+    let kind = ProtectedInputKind::for_job(&binding(consolidation))
+        .unwrap()
+        .as_str();
+    assert!(store.read_all(session, segment).unwrap().iter().any(
+        |entry| matches!(entry, LogEntry::Extension { domain, payload, .. }
+            if domain == PROTECTED_INPUTS_DOMAIN && payload["kinds"] == serde_json::json!([kind]))
+    ));
+}
+
+#[tokio::test]
+async fn subject_job_requirement_survives_fork_compaction_rewind_and_missing_or_wrong_restore_binding()
+ {
+    let _environment = worker_allocation::test_util::ENV_LOCK
+        .lock()
+        .unwrap_or_else(|e| e.into_inner());
+    for consolidation in [false, true] {
+        let temp = tempfile::tempdir().unwrap();
+        let store = store(&temp);
+        let manifest = manifest();
+        let host = Arc::new(Host::new(consolidation));
+        let provider = Provider::default();
+        let requests = provider.requests.clone();
+        let mut worker = boot(
+            manifest.clone(),
+            store.clone(),
+            host.clone(),
+            provider,
+            consolidation,
+        )
+        .await;
+        worker.run_text(SECRET.repeat(100)).await.unwrap();
+        assert!(format!("{:?}", requests.lock().unwrap()[0]).contains(SECRET));
+        let session = worker.session_id();
+        let first = worker.segment_id();
+        assert_retained(&store, session, first, consolidation);
+        store
+            .append(
+                session,
+                first,
+                &LogEntry::Extension {
+                    ts: 1,
+                    domain: "test.concurrent-write".into(),
+                    payload: serde_json::json!({}),
+                },
+            )
+            .unwrap();
+        worker.run_text("fork frozen Job input").await.unwrap();
+        let fork = worker.segment_id();
+        assert_ne!(fork, first);
+        assert_retained(&store, session, fork, consolidation);
+        let compact = worker.compact(0).await.unwrap();
+        assert_ne!(compact, fork);
+        assert_retained(&store, session, compact, consolidation);
+        worker.run_text("keep this turn").await.unwrap();
+        worker.run_text("rewind this turn").await.unwrap();
+        let target = rewind_test_inputs(&worker)[0].0.clone();
+        worker.rewind_to(target).await.unwrap();
+        let rewind = worker.segment_id();
+        assert_retained(&store, session, rewind, consolidation);
+        drop(worker);
+        for segment in [first, fork, compact, rewind] {
+            for mode in [
+                "missing",
+                "wrong-kind",
+                "wrong-job",
+                "wrong-digest",
+                "revoked",
+                "allowed",
+            ] {
+                let provider = Provider::default();
+                let calls = provider.requests.clone();
+                let host = Arc::new(Host::new(consolidation));
+                if mode == "revoked" {
+                    host.status.store(403, Ordering::SeqCst);
+                }
+                let mut restored = restore(
+                    manifest.clone(),
+                    store.clone(),
+                    session,
+                    segment,
+                    host.clone(),
+                    provider,
+                )
+                .await;
+                if mode != "missing" {
+                    let mut bound = binding(if mode == "wrong-kind" {
+                        !consolidation
+                    } else {
+                        consolidation
+                    });
+                    if mode == "wrong-job" {
+                        bound.job_id = "other-job".into();
+                    }
+                    if mode == "wrong-digest" {
+                        bound.input_digest = Some("other-input".into());
+                    }
+                    restored.bind_backend_job(bound).unwrap();
+                }
+                let outcome = restored
+                    .run_text("restored frozen Job input")
+                    .await
+                    .unwrap();
+                assert_eq!(
+                    matches!(outcome, WorkerRunResult::Finished),
+                    mode == "allowed",
+                    "{mode}: {outcome:?}"
+                );
+                assert_eq!(
+                    calls.lock().unwrap().len(),
+                    usize::from(mode == "allowed"),
+                    "{mode}"
+                );
+                if mode == "allowed" || mode == "revoked" {
+                    assert!(host.checks.load(Ordering::SeqCst) > 0);
+                }
+            }
+        }
+    }
+}
+
+#[test]
+fn malformed_subject_job_provenance_never_becomes_a_local_or_new_job_grant() {
+    for payload in [
+        serde_json::json!({"version":1,"kinds":["subject-journal-job"]}),
+        serde_json::json!({"version":1,"kinds":["subject-journal-job"],"subject_jobs":[{
+            "kind":"subject-journal-job","job_id":"","input_digest":"digest"}]}),
+        serde_json::json!({"version":1,"kinds":["subject-journal-job"],"subject_jobs":[{
+            "kind":"workspace-memory","job_id":"job","input_digest":"digest"}]}),
+    ] {
+        let entry = LogEntry::Extension {
+            ts: 1,
+            domain: PROTECTED_INPUTS_DOMAIN.into(),
+            payload,
+        };
+        assert!(ProtectedInputProvenance::restore(&[entry], &[], None).is_err());
+    }
+}
+
+#[tokio::test]
+async fn subject_job_compactor_reauthorizes_after_each_request_before_correction_or_retry() {
+    let _environment = worker_allocation::test_util::ENV_LOCK
+        .lock()
+        .unwrap_or_else(|e| e.into_inner());
+    for (consolidation, invalid_summary) in
+        [(false, false), (false, true), (true, false), (true, true)]
+    {
+        let temp = tempfile::tempdir().unwrap();
+        let host = Arc::new(Host::new(consolidation));
+        let provider = Provider {
+            revoke_on_compact: Some(host.clone()),
+            invalid_first_summary: invalid_summary,
+            ..Default::default()
+        };
+        let requests = provider.requests.clone();
+        let compacts = provider.compactor_calls.clone();
+        let mut worker = boot(manifest(), store(&temp), host, provider, consolidation).await;
+        worker.run_text(SECRET.repeat(100)).await.unwrap();
+        let segment = worker.segment_id();
+        assert!(worker.compact(0).await.is_err());
+        assert_eq!(
+            worker.segment_id(),
+            segment,
+            "failed compaction cannot publish a summary"
+        );
+        assert_eq!(
+            compacts.load(Ordering::SeqCst),
+            1,
+            "revocation stops every later compactor dispatch"
+        );
+        let requests = requests.lock().unwrap();
+        assert_eq!(
+            requests.len(),
+            2,
+            "one allowed foreground and one pre-revocation compactor request only"
+        );
+        assert!(
+            format!("{:?}", requests[1]).contains(SECRET),
+            "the allowed compactor actually received frozen input"
+        );
+    }
+}

crates/worker/tests/controller/subject_job_input_supply.rs

@@ -0,0 +1,191 @@
+use super::*;
+use std::sync::atomic::AtomicBool;
+use worker::hook::{Hook, HookError, HookPreRequestAction, PreLlmRequest, PreRequestContext};
+
+const FROZEN_INPUT: &str = "frozen-subject-job-input-must-not-leak";
+
+// Real Controller/Worker + production supply guard; only the Host transport and
+// model are replaced. Host policy/origin/attempt fences have signed Server tests.
+#[derive(Debug)]
+struct JobHost {
+    operation: &'static str,
+    denied: AtomicBool,
+    checks: AtomicUsize,
+}
+impl JobHost {
+    fn new(consolidation: bool) -> Self {
+        Self {
+            operation: if consolidation {
+                "memory.consolidation.supply"
+            } else {
+                "journal.job.input.supply"
+            },
+            denied: AtomicBool::new(false),
+            checks: AtomicUsize::new(0),
+        }
+    }
+}
+impl WorkspaceClient for JobHost {
+    fn workspace_id(&self) -> Option<&str> {
+        Some("workspace-test")
+    }
+    fn kind(&self) -> &str {
+        "job-supply-host"
+    }
+    fn is_available(&self) -> bool {
+        true
+    }
+    fn execute(
+        &self,
+        request: WorkspaceRequest,
+    ) -> Result<WorkspaceResponse, WorkspaceClientError> {
+        assert_eq!(
+            request.path,
+            "/api/w/workspace-test/authorization/bound-check"
+        );
+        let input: server_api::authorization::BoundOperationAuthorizationRequest =
+            serde_json::from_str(request.body.as_deref().unwrap()).unwrap();
+        assert_eq!(input.operation, self.operation);
+        assert_eq!(
+            input.target,
+            server_api::authorization::BoundAuthorizationTarget::Subject {}
+        );
+        self.checks.fetch_add(1, Ordering::SeqCst);
+        Ok(WorkspaceResponse {
+            status: if self.denied.load(Ordering::SeqCst) { 403 } else { 200 },
+            body: r#"{"scope":{"kind":"subject","workspace_id":"workspace-test","subject_id":"bound-subject"}}"#.into(),
+        })
+    }
+}
+
+async fn job_worker(
+    consolidation: bool,
+    host: Arc<JobHost>,
+) -> (Worker<MockClient, TestStore>, MockClient) {
+    let workspace = tempfile::tempdir().unwrap();
+    let profile = if consolidation {
+        "subjektiv-memory-consolidation"
+    } else {
+        "subjektiv-journal-recording"
+    };
+    let mut manifest = ProfileResolver::new()
+        .with_workspace_base(workspace.path())
+        .resolve(
+            &ProfileSelector::source_named(ProfileRegistrySource::Builtin, profile),
+            ProfileResolveOptions::with_worker_name("subject-job-supply-test"),
+        )
+        .unwrap()
+        .manifest;
+    if consolidation {
+        manifest
+            .feature
+            .subjektiv
+            .bind_workspace_settings(manifest::WorkspaceMemorySettingsSnapshot {
+                workspace_id: "workspace-test".into(),
+                language: "English".into(),
+            })
+            .unwrap();
+    }
+    manifest.compaction = Some(manifest::CompactionConfig {
+        retained_tokens: 1,
+        model: None,
+        ..Default::default()
+    });
+    let client = MockClient::new(simple_text_events());
+    let (mut worker, _) = make_worker_with_manifest_and_workspace_context(
+        client.clone(),
+        manifest,
+        WorkerWorkspaceContext::with_client(None, host),
+    )
+    .await;
+    worker
+        .bind_backend_job(worker::BackendJobExecutionBinding {
+            job_id: "bound-job".into(),
+            attempt_id: "bound-attempt".into(),
+            input_digest: Some("bound-digest".into()),
+            subjektiv_consolidation: consolidation,
+            subjektiv_journal: !consolidation,
+        })
+        .unwrap();
+    (worker, client)
+}
+
+#[tokio::test]
+async fn idle_controller_compaction_rejects_revoked_subject_job_input_before_any_model_request() {
+    for consolidation in [false, true] {
+        let host = Arc::new(JobHost::new(consolidation));
+        let (worker, client) = job_worker(consolidation, host.clone()).await;
+        let handle = spawn_controller(worker).await;
+        let mut events = handle.subscribe();
+        let id = protocol::new_submission_request_id();
+        handle
+            .send(Method::submit_text(id.clone(), FROZEN_INPUT.repeat(200)))
+            .await
+            .unwrap();
+        assert!(
+            drain_until(
+                &mut events,
+                std::time::Duration::from_secs(2),
+                |event| matches!(event,
+            Event::WorkerState { snapshot } if snapshot.catalog_status() == WorkerStatus::Idle
+                && snapshot.last_finished_submission_request_id.as_deref() == Some(id.as_str()))
+            )
+            .await
+        );
+        let before = client.captured_requests().len();
+        assert_eq!(before, 1);
+        assert!(format!("{:?}", client.captured_requests()[0]).contains(FROZEN_INPUT));
+        let checks = host.checks.load(Ordering::SeqCst);
+        assert!(checks > 0);
+        host.denied.store(true, Ordering::SeqCst);
+        handle
+            .send(Method::Compact {
+                command: worker_command(&handle),
+            })
+            .await
+            .unwrap();
+        assert!(drain_until(&mut events, std::time::Duration::from_secs(2), |event| matches!(event,
+            Event::Error { message, .. } if message.contains(host.operation) && message.contains("403"))).await,
+            "revoked Job compaction must report its operation denial");
+        assert!(host.checks.load(Ordering::SeqCst) > checks);
+        assert_eq!(
+            client.captured_requests().len(),
+            before,
+            "revoked frozen input reached compactor"
+        );
+    }
+}
+
+struct AuthorizeThenRevoke(Arc<JobHost>);
+#[async_trait]
+impl Hook<PreLlmRequest> for AuthorizeThenRevoke {
+    async fn call(&self, _: &PreRequestContext) -> Result<HookPreRequestAction, HookError> {
+        self.0
+            .authorize_subject_operation(self.0.operation)
+            .unwrap();
+        self.0.denied.store(true, Ordering::SeqCst);
+        Ok(HookPreRequestAction::Continue)
+    }
+}
+
+#[tokio::test]
+async fn later_hook_revocation_rejects_subject_job_input_at_final_supply_boundary() {
+    for consolidation in [false, true] {
+        let host = Arc::new(JobHost::new(consolidation));
+        let (mut worker, client) = job_worker(consolidation, host.clone()).await;
+        worker.add_pre_llm_request_hook(AuthorizeThenRevoke(host.clone()));
+        let outcome = worker.run_text(FROZEN_INPUT).await.unwrap();
+        assert!(
+            host.checks.load(Ordering::SeqCst) > 0,
+            "the hook first observes allowed Host authority"
+        );
+        assert!(
+            matches!(outcome, worker::WorkerRunResult::Interrupted { .. }),
+            "{outcome:?}"
+        );
+        assert!(
+            client.captured_requests().is_empty(),
+            "same-request revocation leaked Job input"
+        );
+    }
+}

crates/worker/tests/controller_test.rs

@@ -1474,6 +1474,9 @@ async fn consolidation_job_preserves_custom_instruction_and_explicit_tool_policy
     }
 }
 
+#[path = "controller/subject_job_input_supply.rs"]
+mod subject_job_input_supply;
+
 #[tokio::test]
 async fn denied_subject_job_input_supply_stops_before_model_dispatch() {
     for (profile, consolidation, operation) in [

crates/workspace-server/src/server/tests/subject_authorization_tests.rs

@@ -5,6 +5,8 @@
 //! runs a real ProfileRuntimeWorkerFactory/Controller and materialized Workdir;
 //! only its external model stream is scripted. Neither fixture claims OS isolation.
 use super::*;
+#[path = "subject_job_model_supply_tests.rs"]
+mod subject_job_model_supply;
 use agen::tool::{Tool, ToolExecutionContext, ToolOutput};
 use server_api::authorization::{
     PermissionResource, ScopedPermission, SubjectPermissions, SubjectPermissionsUpdate,

crates/workspace-server/src/server/tests/subject_job_model_supply_tests.rs

@@ -0,0 +1,390 @@
+use super::*;
+use agen::llm_client::{ClientError, LlmClient, Request as ModelRequest, ResponseStream};
+use async_trait::async_trait;
+use std::sync::atomic::{AtomicUsize, Ordering};
+use worker::hook::{Hook, HookError, HookPreRequestAction, PreLlmRequest, PreRequestContext};
+
+type ModelStore =
+    session_store::CombinedStore<session_store::FsStore, session_store::FsWorkerStore>;
+
+#[derive(Clone, Default)]
+struct Model {
+    calls: Arc<AtomicUsize>,
+    input: Arc<Mutex<Vec<String>>>,
+}
+#[async_trait]
+impl LlmClient for Model {
+    async fn stream(
+        &self,
+        request: ModelRequest,
+    ) -> std::result::Result<ResponseStream, ClientError> {
+        use agen::llm_client::event::{Event, ResponseStatus, StatusEvent};
+        self.calls.fetch_add(1, Ordering::SeqCst);
+        self.input.lock().unwrap().push(format!("{request:?}"));
+        Ok(Box::pin(futures::stream::iter(
+            [
+                Event::text_block_start(0),
+                Event::text_delta(0, "done"),
+                Event::text_block_stop(0, None),
+                Event::Status(StatusEvent {
+                    status: ResponseStatus::Completed,
+                }),
+            ]
+            .into_iter()
+            .map(Ok),
+        )))
+    }
+    fn clone_boxed(&self) -> Box<dyn LlmClient> {
+        Box::new(self.clone())
+    }
+}
+
+async fn job_fixture(
+    consolidation: bool,
+    missing_origin: bool,
+) -> (
+    tempfile::TempDir,
+    WorkspaceApi,
+    crate::backend_job::BackendJobReservation,
+) {
+    if !consolidation {
+        return super::super::subjektiv_journal_tests::model_supply_job_fixture(missing_origin)
+            .await;
+    }
+    let f = Fixture::new().await;
+    f.assign(
+        vec![subject_permission(
+            &f.subject,
+            "memory.consolidation.supply",
+        )],
+        vec![],
+    )
+    .await;
+    if missing_origin {
+        let historical = BackendJobRequest {
+            job_id: format!("historical-job-{}", Uuid::now_v7()),
+            purpose: "subjektiv_consolidation".into(),
+            input_ref: format!("subjektiv://{}/consolidation", f.subject),
+            input: json!({"subject_id": f.subject, "candidate_ids": []}),
+            instruction: "retained legacy input".into(),
+            profile: "builtin:subjektiv-memory-consolidation".into(),
+            grants: crate::backend_job::BackendJobGrants {
+                subjektiv_consolidation: Some(crate::backend_job::SubjektivConsolidationGrant {
+                    subject_id: f.subject.clone(),
+                    candidate_ids: vec![],
+                }),
+                ..Default::default()
+            },
+            serialization_key: None,
+            source_worker: None,
+            notification_target: None,
+            limits: Default::default(),
+        };
+        let saved = f
+            .api
+            .store
+            .reserve_backend_job(TEST_WORKSPACE_ID, &historical, &Utc::now().to_rfc3339())
+            .unwrap();
+        return (f._dir, f.api, saved);
+    }
+    subjektiv_job::dispatch(&f.api, &f.subject, Some(f.source), 0, 0).unwrap();
+    let job = f
+        .api
+        .store
+        .find_active_backend_job_for_resource(
+            TEST_WORKSPACE_ID,
+            &crate::backend_job::subjektiv_serialization_key(&f.subject),
+        )
+        .unwrap()
+        .unwrap();
+    let attempt = f
+        .api
+        .store
+        .get_backend_job_attempt(
+            TEST_WORKSPACE_ID,
+            &job.request.job_id,
+            &crate::backend_job::attempt_id(&job.request.job_id, job.current_attempt),
+        )
+        .unwrap()
+        .unwrap();
+    (
+        f._dir,
+        f.api,
+        crate::backend_job::BackendJobReservation {
+            job,
+            attempt,
+            replayed: false,
+            resource_reused: false,
+        },
+    )
+}
+
+fn replace_policy(api: &WorkspaceApi, subject: &str, operations: &[&str]) {
+    let store = open_subjektiv_store(api).unwrap();
+    let current = store.subject_permissions(subject).unwrap();
+    store
+        .set_subject_permissions(
+            subject,
+            &SubjectPermissionsUpdate {
+                operation_id: Uuid::now_v7().to_string(),
+                expected_change_id: current.map(|p| p.change_id),
+                execution: operations
+                    .iter()
+                    .map(|op| subject_permission(subject, op))
+                    .collect(),
+                delegation: vec![],
+            },
+        )
+        .unwrap();
+}
+
+async fn model_worker(
+    root: &std::path::Path,
+    api: &mut WorkspaceApi,
+    reservation: &crate::backend_job::BackendJobReservation,
+    consolidation: bool,
+) -> (worker::Worker<Model, ModelStore>, Model) {
+    let target = reservation.attempt.worker.as_ref().unwrap();
+    let identity = Arc::new(RuntimeIdentityMaterial::generate(&target.runtime_id).unwrap());
+    configure_runtime_request_auth(api, &identity, &target.runtime_id);
+    let client = Arc::new(RouterClient {
+        api: api.clone(),
+        identity,
+        worker: target.clone(),
+        handle: tokio::runtime::Handle::current(),
+        requests: Mutex::new(vec![]),
+    });
+    let mut manifest = local_task_manifest();
+    manifest.worker.name = format!("signed-job-supply-{}", Uuid::now_v7());
+    manifest.compaction = Some(manifest::CompactionConfig {
+        model: None,
+        retained_tokens: 1,
+        ..Default::default()
+    });
+    let model = Model::default();
+    let engine =
+        agen::Engine::<_, agen::state::Mutable, worker::SessionHistoryMetadata>::new_annotated(
+            model.clone(),
+        );
+    let store = session_store::CombinedStore::new(
+        session_store::FsStore::new(root.join("sessions")).unwrap(),
+        session_store::FsWorkerStore::new(root.join("workers")).unwrap(),
+    );
+    let mut worker = worker::Worker::new(
+        manifest,
+        engine,
+        store,
+        worker::WorkerWorkspaceContext::with_client(
+            Some(worker::WorkspaceId::new(TEST_WORKSPACE_ID).unwrap()),
+            client,
+        ),
+        worker::WorkerFilesystemAuthority::None,
+        manifest::Scope::empty(),
+    )
+    .await
+    .unwrap();
+    worker
+        .bind_backend_job(worker::BackendJobExecutionBinding {
+            job_id: reservation.job.request.job_id.clone(),
+            attempt_id: reservation.attempt.attempt_id.clone(),
+            input_digest: Some(reservation.job.request.input_digest().unwrap()),
+            subjektiv_consolidation: consolidation,
+            subjektiv_journal: !consolidation,
+        })
+        .unwrap();
+    worker
+        .engine_mut()
+        .set_system_prompt("Bound Job model fixture");
+    (worker, model)
+}
+
+struct PolicyRevocation {
+    api: WorkspaceApi,
+    subject: String,
+}
+#[async_trait]
+impl Hook<PreLlmRequest> for PolicyRevocation {
+    async fn call(
+        &self,
+        _: &PreRequestContext,
+    ) -> std::result::Result<HookPreRequestAction, HookError> {
+        // Change the real durable policy after the guard's initial signed allow.
+        // This observes new-access denial, not Runtime cancellation completion.
+        replace_policy(&self.api, &self.subject, &[]);
+        Ok(HookPreRequestAction::Continue)
+    }
+}
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn signed_job_host_policy_origin_and_attempt_fences_stop_frozen_input_model_supply() {
+    for consolidation in [false, true] {
+        for mode in [
+            "policy",
+            "attempt",
+            "origin",
+            "late-hook",
+            "outside-grant",
+            "missing-origin",
+        ] {
+            let (_dir, mut api, mut reservation) =
+                job_fixture(consolidation, mode == "missing-origin").await;
+            let subject = reservation
+                .job
+                .request
+                .grants
+                .subjektiv_consolidation
+                .as_ref()
+                .map(|g| g.subject_id.clone())
+                .or_else(|| {
+                    reservation
+                        .job
+                        .request
+                        .grants
+                        .subjektiv_journal
+                        .as_ref()
+                        .map(|g| g.subject_id.clone())
+                })
+                .unwrap();
+            replace_policy(
+                &api,
+                &subject,
+                &["memory.consolidation.supply", "journal.job.input.supply"],
+            );
+            if mode == "missing-origin" {
+                // Retained pre-migration input through the real ledger. New
+                // dispatch rejects this; it must also fail at later model supply.
+                let historical = reservation.job.request.clone();
+                assert!(historical.source_worker.is_none());
+                let now = Utc::now().to_rfc3339();
+                let saved = reservation.clone();
+                api.store
+                    .claim_backend_job_attempt_dispatch(
+                        TEST_WORKSPACE_ID,
+                        &historical.job_id,
+                        &saved.attempt.attempt_id,
+                        &now,
+                    )
+                    .unwrap();
+                let target = spawn_ticket_check_source(&api, "historical-job-model");
+                api.store
+                    .bind_backend_job_attempt_worker(
+                        TEST_WORKSPACE_ID,
+                        &historical.job_id,
+                        &saved.attempt.attempt_id,
+                        &target,
+                        None,
+                        &now,
+                    )
+                    .unwrap();
+                reservation.job = api
+                    .store
+                    .get_backend_job(TEST_WORKSPACE_ID, &historical.job_id)
+                    .unwrap()
+                    .unwrap();
+                reservation.attempt = api
+                    .store
+                    .get_backend_job_attempt(
+                        TEST_WORKSPACE_ID,
+                        &historical.job_id,
+                        &saved.attempt.attempt_id,
+                    )
+                    .unwrap()
+                    .unwrap();
+            }
+            let local = tempfile::tempdir().unwrap();
+            let (mut worker, model) = model_worker(
+                local.path(),
+                &mut api,
+                &reservation,
+                if mode == "outside-grant" {
+                    !consolidation
+                } else {
+                    consolidation
+                },
+            )
+            .await;
+            if mode == "late-hook" {
+                worker.add_pre_llm_request_hook(PolicyRevocation {
+                    api: api.clone(),
+                    subject: subject.clone(),
+                });
+            }
+            let frozen = serde_json::to_string(&reservation.job.request.input).unwrap();
+            let outcome = worker.run_text(frozen.clone()).await.unwrap();
+            if matches!(mode, "late-hook" | "outside-grant" | "missing-origin") {
+                assert!(
+                    matches!(outcome, worker::WorkerRunResult::Interrupted { .. }),
+                    "{mode}: {outcome:?}"
+                );
+                assert_eq!(model.calls.load(Ordering::SeqCst), 0, "{mode}");
+                continue;
+            }
+            assert!(
+                matches!(outcome, worker::WorkerRunResult::Finished),
+                "{mode}: {outcome:?}"
+            );
+            assert_eq!(model.calls.load(Ordering::SeqCst), 1);
+            assert!(
+                model.input.lock().unwrap()[0].contains(&subject),
+                "allowed real frozen input must reach the model"
+            );
+            match mode {
+                "policy" => replace_policy(&api, &subject, &[]),
+                "attempt" => {
+                    api.finish_backend_job_attempt(
+                        &reservation.job.request.job_id,
+                        &reservation.attempt.attempt_id,
+                        BackendJobAttemptState::Failed,
+                        "cancelled",
+                        "test cancellation",
+                    )
+                    .unwrap();
+                }
+                "origin" => {
+                    // Historical disconnect input in a deterministic execution
+                    // fixture, not a claim of live Controller release/cancellation.
+                    let source = reservation.job.request.source_worker.as_ref().unwrap();
+                    let lease = api
+                        .store
+                        .require_current_worker_singleton_owner(TEST_WORKSPACE_ID, source)
+                        .unwrap()
+                        .unwrap();
+                    let binding = "historical-disconnected-job-origin";
+                    let operation = "historical-job-origin-release";
+                    api.config_store
+                        .reserve_subject_release(TEST_WORKSPACE_ID, &lease, binding, operation)
+                        .unwrap();
+                    api.config_store
+                        .confirm_subject_release_proof(
+                            TEST_WORKSPACE_ID,
+                            &lease,
+                            binding,
+                            operation,
+                            false,
+                        )
+                        .unwrap();
+                    api.config_store
+                        .release_subject_singleton(TEST_WORKSPACE_ID, &lease, binding, operation)
+                        .unwrap();
+                }
+                _ => unreachable!(),
+            }
+            let error = worker.compact(0).await.unwrap_err();
+            assert!(error.to_string().contains("supply"), "{mode}: {error}");
+            assert_eq!(
+                model.calls.load(Ordering::SeqCst),
+                1,
+                "{mode}: frozen input reached compactor after revocation"
+            );
+            assert_eq!(
+                api.store
+                    .get_backend_job(TEST_WORKSPACE_ID, &reservation.job.request.job_id)
+                    .unwrap()
+                    .unwrap()
+                    .request,
+                reservation.job.request,
+                "revocation never rewrites the frozen grant/input/origin"
+            );
+        }
+    }
+}

crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs

@@ -683,6 +683,36 @@ fn message(text: &str) -> session_store::LoggedItem {
         content: vec![session_store::LoggedContentPart::Text { text: text.into() }],
     }
 }
+pub(super) async fn model_supply_job_fixture(
+    missing_origin: bool,
+) -> (tempfile::TempDir, WorkspaceApi, BackendJobReservation) {
+    let fixture = if missing_origin {
+        RuntimeFixture::with_history_and_origin(
+            vec![session_store::LoggedHistoryEntry {
+                item: message(SOURCE_SENTINEL),
+                metadata: session_store::LoggedSessionHistoryMetadata {
+                    input_metadata: None,
+                    entry_id: session_store::LoggedSessionHistoryEntryId(
+                        "historical-source".into(),
+                    ),
+                    origin: session_store::LoggedSessionHistoryOrigin::LocalHumanInput,
+                    derivation: None,
+                },
+            }],
+            false,
+        )
+        .await
+    } else {
+        RuntimeFixture::new().await
+    };
+    let reservation = if missing_origin {
+        fixture.reserved
+    } else {
+        fixture.dispatch_input()
+    };
+    (fixture.dir, fixture.api, reservation)
+}
+
 impl RuntimeFixture {
     async fn new() -> Self {
         Self::with_history(vec![session_store::LoggedHistoryEntry {
@@ -697,6 +727,12 @@ impl RuntimeFixture {
         .await
     }
     async fn with_history(history: Vec<session_store::LoggedHistoryEntry>) -> Self {
+        Self::with_history_and_origin(history, true).await
+    }
+    async fn with_history_and_origin(
+        history: Vec<session_store::LoggedHistoryEntry>,
+        frozen_origin: bool,
+    ) -> Self {
         let dir = tempfile::tempdir().unwrap();
         let (api, execution) = test_api_with_recording_backend(dir.path()).await;
         let store = open_subjektiv_store(&api).unwrap();
@@ -789,7 +825,10 @@ impl RuntimeFixture {
         store
             .accept_journal_sources(&host, "bounded captured prefix", &input.selected_ranges)
             .unwrap();
-        let request = subjektiv_journal::recording_request(&host, attr, &id, input).unwrap();
+        let mut request = subjektiv_journal::recording_request(&host, attr, &id, input).unwrap();
+        if !frozen_origin {
+            request.source_worker = None;
+        }
         assert!(
             request.input["capture"]["entries"]
                 .as_array()

docs/report/2026-10-28-t787-authorization-boundary-inventory.md

@@ -76,4 +76,12 @@
 - 安定候補でroot `cargo check`、変更crateの通常suite（`server-api`、`tools`、`worker`、`worker-runtime`、`yoi-workspace-server`）、`server-api --features typescript` のsuite、変更crateのrustdoc生成、format・差分・状態契約名検査が成功した。restoreを含むsuiteにはcheckout内 `target/restore-test-environment/` 以下の専用config/data/runtimeを使用した。最初のServer再実行は隔離環境変数不足でpreflight失敗し、設定後の全suite再実行が成功した。preflightを削除して通してはいない。
 - OpenAPIと影響する各OpenAPI closure出力は所有exampleの `--check` で一致した。Webの型検査と変更型を使う関連APIテストが成功した。描画変更はなく、browser/E2E・稼働環境更新は実施していない。通常suiteで元から明示選択のignoredテストを実行済みとは扱わない。
 
+## 初回レビューで確認されたJob入力供給の不足
+
+初回候補の通常suite成功は、凍結Job入力をcompactorへ再供給する経路と、Jobの通常hook後に権限が失効する経路を証明していなかった。正式レビューは実Controllerと捕捉モデルで漏れを再現し、request_changesとした。上記の成功記録をこの境界の成功へ拡張してはならない。
+
+修正では、Job種別・Job ID・入力digestの保持義務を既存の保護入力provenanceに追加し、Host由来の実行bindingとの一致と現行認可を共通供給guardで検査する。保存値はgrantではない。attemptは現在のHost正本で毎回検査し、同じ凍結入力への正当な再試行と別Jobへの付替えを区別する。通常hookは早期診断に留まり、hook後とcompactorの各dispatchを代替しない。
+
+恒久回帰は `worker_job_input_supply_tests.rs`（rewrite/restoreとcompactor継続）、`tests/controller/subject_job_input_supply.rs`（実Controllerと後続hook失効）、`server/tests/subject_job_model_supply_tests.rs`（署名Host・実policy/Job台帳・捕捉モデル）へ追加した。Journal/consolidation双方で許可時の実供給と拒否時のモデル要求0を観測する。origin切断の履歴fixtureは稼働Controllerの停止確認を主張せず、既存のconnected release検証とは分ける。修正前の恒久回帰失敗と修正後の成功、最終suiteの結果はMRで区別して記録する。
+
 件数・生ログ・source識別はMRの検証記録に置く。これらの成功だけで受入条件の網羅、正式承認、統合、Ticketの完了を推論しない。