diff --git a/crates/worker/src/feature/builtin/backend_job_input.rs b/crates/worker/src/feature/builtin/backend_job_input.rs index 9103fcdd4..508463a23 100644 --- a/crates/worker/src/feature/builtin/backend_job_input.rs +++ b/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; diff --git a/crates/worker/src/worker.rs b/crates/worker/src/worker.rs index 6388fc2c6..45b002f12 100644 --- a/crates/worker/src/worker.rs +++ b/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 { + 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 { 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, +} +impl ProtectedSubjectJobInput { + fn from_binding(binding: &crate::BackendJobExecutionBinding) -> Option { + 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, + #[serde(default, skip_serializing_if = "std::collections::BTreeSet::is_empty")] + subject_jobs: std::collections::BTreeSet, } 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>, prompt_source: Option>, context_source: Option>, + // 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)>, } impl ProtectedInputSupply { @@ -3841,15 +3908,15 @@ impl ProtectedInputSupply { &self, history: impl IntoIterator, ) -> 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, - ) -> 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 Worker { 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 Worker { { 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, allowed: bool) { if allowed { diff --git a/crates/worker/src/worker_job_input_supply_tests.rs b/crates/worker/src/worker_job_input_supply_tests.rs new file mode 100644 index 000000000..2a98954c1 --- /dev/null +++ b/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; +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 { + 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>>, + compactor_calls: Arc, + revoke_on_compact: Option>, + invalid_first_summary: bool, +} +#[async_trait] +impl LlmClient for Provider { + async fn stream( + &self, + request: agen::llm_client::Request, + ) -> Result { + 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 { + 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, + provider: Provider, + consolidation: bool, +) -> Worker { + 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, + provider: Provider, +) -> Worker, TestStore> { + Worker::, _>::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" + ); + } +} diff --git a/crates/worker/tests/controller/subject_job_input_supply.rs b/crates/worker/tests/controller/subject_job_input_supply.rs new file mode 100644 index 000000000..2edcd8a19 --- /dev/null +++ b/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 { + 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, +) -> (Worker, 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); +#[async_trait] +impl Hook for AuthorizeThenRevoke { + async fn call(&self, _: &PreRequestContext) -> Result { + 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" + ); + } +} diff --git a/crates/worker/tests/controller_test.rs b/crates/worker/tests/controller_test.rs index 22d3c0d49..49a07d224 100644 --- a/crates/worker/tests/controller_test.rs +++ b/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 [ diff --git a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs b/crates/workspace-server/src/server/tests/subject_authorization_tests.rs index 1df4c8306..6b80d8f9f 100644 --- a/crates/workspace-server/src/server/tests/subject_authorization_tests.rs +++ b/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, diff --git a/crates/workspace-server/src/server/tests/subject_job_model_supply_tests.rs b/crates/workspace-server/src/server/tests/subject_job_model_supply_tests.rs new file mode 100644 index 000000000..10e418e9b --- /dev/null +++ b/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; + +#[derive(Clone, Default)] +struct Model { + calls: Arc, + input: Arc>>, +} +#[async_trait] +impl LlmClient for Model { + async fn stream( + &self, + request: ModelRequest, + ) -> std::result::Result { + 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 { + 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) { + 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 for PolicyRevocation { + async fn call( + &self, + _: &PreRequestContext, + ) -> std::result::Result { + // 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" + ); + } + } +} diff --git a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs b/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs index 967e4aebb..5f8bd3216 100644 --- a/crates/workspace-server/src/server/tests/subjektiv_journal_tests.rs +++ b/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) -> Self { + Self::with_history_and_origin(history, true).await + } + async fn with_history_and_origin( + history: Vec, + 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() diff --git a/docs/report/2026-10-28-t787-authorization-boundary-inventory.md b/docs/report/2026-10-28-t787-authorization-boundary-inventory.md index f1f3754a4..1ee416d6f 100644 --- a/docs/report/2026-10-28-t787-authorization-boundary-inventory.md +++ b/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の完了を推論しない。