feat: caller-selected Backend Jobs and bounded subjektiv consolidation (T-704)

crates/workspace-server/src/backend_job.rs

@@ -199,6 +199,34 @@ const fn default_max_attempts() -> u8 {
     DEFAULT_MAX_ATTEMPTS
 }
 
+/// Explicit Backend-issued domain capabilities, never inferred from purpose or profile.
+#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct BackendJobGrants {
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub subjektiv_consolidation: Option<SubjektivConsolidationGrant>,
+}
+
+impl BackendJobGrants {
+    pub fn is_empty(&self) -> bool {
+        self.subjektiv_consolidation.is_none()
+    }
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(deny_unknown_fields)]
+pub struct SubjektivConsolidationGrant {
+    pub subject_id: String,
+    /// Exact immutable candidate scope. Empty is a surface-only refresh grant:
+    /// the subject is still locked, but no candidate access is authorized.
+    pub candidate_ids: Vec<String>,
+}
+
+/// Common runner lock identity for a subject, independent of the candidate snapshot.
+pub fn subjektiv_serialization_key(subject_id: &str) -> String {
+    format!("subjektiv-consolidation:{subject_id}")
+}
+
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
 #[serde(deny_unknown_fields)]
 pub struct BackendJobRequest {
@@ -216,6 +244,13 @@ pub struct BackendJobRequest {
     pub instruction: String,
     /// Existing configured profile selector. Job dispatch never accepts raw profile source.
     pub profile: String,
+    /// Immutable domain grants; profile resolution alone never grants access.
+    #[serde(default, skip_serializing_if = "BackendJobGrants::is_empty")]
+    pub grants: BackendJobGrants,
+    /// Optional common-runner single-flight resource, not a separate domain state machine.
+    /// Consolidation grants always derive their subject lock; an explicit key must match it.
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    pub serialization_key: Option<String>,
     /// Writer provenance and optional advisory destination, not parent ownership.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub source_worker: Option<RuntimeWorkerRef>,
@@ -233,10 +268,34 @@ impl BackendJobRequest {
         validate_bounded_text("input_ref", &self.input_ref, MAX_JOB_REFERENCE_BYTES)?;
         validate_bounded_text("instruction", &self.instruction, MAX_JOB_INSTRUCTION_BYTES)?;
         validate_identifier("profile", &self.profile, 256)?;
-        if self.profile != "builtin:backend-job" {
-            return Err(Error::InvalidInput(
-                "Backend Jobs must use the dedicated builtin:backend-job profile".to_string(),
-            ));
+        self.profile_selector()?;
+        if let Some(key) = &self.serialization_key {
+            validate_identifier("serialization_key", key, MAX_JOB_REFERENCE_BYTES)?;
+        }
+        if let Some(grant) = &self.grants.subjektiv_consolidation {
+            validate_identifier("grant subject_id", &grant.subject_id, 256)?;
+            if grant.candidate_ids.len() > 256 {
+                return Err(Error::InvalidInput(
+                    "Backend Job consolidation grant requires 0..=256 candidate ids".into(),
+                ));
+            }
+            let mut seen = std::collections::HashSet::new();
+            for id in &grant.candidate_ids {
+                validate_identifier("grant candidate_id", id, 256)?;
+                if !seen.insert(id) {
+                    return Err(Error::InvalidInput(
+                        "Backend Job consolidation grant has duplicate candidate ids".into(),
+                    ));
+                }
+            }
+            if let Some(key) = &self.serialization_key
+                && *key != subjektiv_serialization_key(&grant.subject_id)
+            {
+                return Err(Error::InvalidInput(
+                    "Backend Job serialization_key must match the consolidation subject lock"
+                        .into(),
+                ));
+            }
         }
         let input = serde_json::to_vec(&self.input).map_err(|error| {
             Error::InvalidInput(format!("serialize Backend Job input: {error}"))
@@ -249,6 +308,44 @@ impl BackendJobRequest {
         self.limits.validate()
     }
 
+    /// Validate selector syntax only. Registry existence and resolution are dispatch authority.
+    pub fn profile_selector(&self) -> Result<manifest::ProfileSelector> {
+        use manifest::ProfileSelector;
+        validate_identifier("profile", &self.profile, 256)?;
+        let selector = ProfileSelector::parse_cli(&self.profile);
+        let valid = match &selector {
+            ProfileSelector::Default => true,
+            ProfileSelector::Path { .. } => false,
+            ProfileSelector::Named { source, name } => {
+                !name.is_empty()
+                    && !name.starts_with('.')
+                    && !name.contains(['/', '\\', ':'])
+                    && name
+                        .chars()
+                        .all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.'))
+                    && ![".toml", ".json", ".dcdl", ".nix"]
+                        .iter()
+                        .any(|suffix| name.ends_with(suffix))
+                    && (source.is_some() || name != "inherit")
+            }
+        };
+        if !valid {
+            return Err(Error::InvalidInput(
+                "Backend Job profile must be a manifest registry selector, not raw source or a path".into(),
+            ));
+        }
+        Ok(selector)
+    }
+
+    /// Effective single-flight key. Typed grants cannot bypass their subject fence.
+    pub fn resource_key(&self) -> Option<String> {
+        self.grants
+            .subjektiv_consolidation
+            .as_ref()
+            .map(|grant| subjektiv_serialization_key(&grant.subject_id))
+            .or_else(|| self.serialization_key.clone())
+    }
+
     pub fn fingerprint(&self) -> Result<String> {
         self.validate()?;
         let encoded = serde_json::to_vec(self).map_err(|error| {
@@ -346,7 +443,12 @@ pub struct BackendJobAttemptRecord {
 pub struct BackendJobReservation {
     pub job: BackendJobRecord,
     pub attempt: BackendJobAttemptRecord,
+    /// Exact replay of the caller-selected immutable intent.
     pub replayed: bool,
+    /// A different intent already fences this resource. The returned request is
+    /// the original snapshot; the caller must defer its new input to a followup.
+    #[serde(default)]
+    pub resource_reused: bool,
 }
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq)]
@@ -486,6 +588,8 @@ mod tests {
             input: serde_json::json!({"title": "Check me"}),
             instruction: "Check this immutable Ticket snapshot.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: BackendJobGrants::default(),
+            serialization_key: None,
             source_worker: Some(RuntimeWorkerRef::new("runtime-a", "worker-a")),
             notification_target: Some(RuntimeWorkerRef::new("runtime-a", "worker-a")),
             limits: BackendJobLimits::default(),
@@ -509,6 +613,133 @@ mod tests {
         }
     }
 
+    #[test]
+    fn registry_profile_syntax_accepts_configured_selectors_without_an_allowlist() {
+        for profile in [
+            "builtin:backend-job",
+            "builtin:subjektiv-memory-consolidation",
+            "project:custom",
+            "user:custom",
+            "custom",
+            "default",
+            "builtin:not-yet-configured",
+        ] {
+            let mut request = request();
+            request.profile = profile.into();
+            request.validate().unwrap();
+        }
+        for profile in [
+            "",
+            " builtin:backend-job",
+            "path:custom",
+            "./custom.toml",
+            "/tmp/custom",
+            "project:../custom",
+            "project:custom.toml",
+            "builtin:",
+            "unknown:custom",
+            "project:a:b",
+            "C:\\custom",
+            "{\"worker\":{}}",
+            "inherit",
+        ] {
+            let mut request = request();
+            request.profile = profile.into();
+            assert!(request.validate().is_err(), "accepted {profile}");
+        }
+    }
+
+    #[test]
+    fn explicit_grants_are_immutable_bounded_and_derive_a_subject_fence() {
+        let mut request = request();
+        let original = request.fingerprint().unwrap();
+        request.grants.subjektiv_consolidation = Some(SubjektivConsolidationGrant {
+            subject_id: "subject-a".into(),
+            candidate_ids: vec!["candidate-1".into()],
+        });
+        assert_ne!(original, request.fingerprint().unwrap());
+        assert_eq!(
+            request.resource_key(),
+            Some(subjektiv_serialization_key("subject-a"))
+        );
+        let granted = request.fingerprint().unwrap();
+        request
+            .grants
+            .subjektiv_consolidation
+            .as_mut()
+            .unwrap()
+            .candidate_ids
+            .push("candidate-2".into());
+        assert_ne!(granted, request.fingerprint().unwrap());
+        request.serialization_key = Some("another-subject".into());
+        assert!(request.validate().is_err());
+        request.serialization_key = None;
+        request
+            .grants
+            .subjektiv_consolidation
+            .as_mut()
+            .unwrap()
+            .candidate_ids
+            .push("candidate-2".into());
+        assert!(request.validate().is_err());
+    }
+
+    #[test]
+    fn empty_consolidation_grant_keeps_subject_lock_without_candidate_authority() {
+        let mut request = request();
+        request.grants.subjektiv_consolidation = Some(SubjektivConsolidationGrant {
+            subject_id: "subject-a".into(),
+            candidate_ids: vec![],
+        });
+        request.validate().unwrap();
+        assert!(
+            !request.grants.is_empty(),
+            "empty candidate scope still carries a subject grant"
+        );
+        assert_eq!(
+            request.resource_key(),
+            Some(subjektiv_serialization_key("subject-a"))
+        );
+        let encoded = serde_json::to_string(&request).unwrap();
+        assert!(encoded.contains("\"candidate_ids\":[]"));
+        let decoded: BackendJobRequest = serde_json::from_str(&encoded).unwrap();
+        assert_eq!(decoded, request);
+        let surface_only = request.fingerprint().unwrap();
+        request
+            .grants
+            .subjektiv_consolidation
+            .as_mut()
+            .unwrap()
+            .candidate_ids = (0..256).map(|index| format!("candidate-{index}")).collect();
+        request.validate().unwrap();
+        assert_ne!(surface_only, request.fingerprint().unwrap());
+        request
+            .grants
+            .subjektiv_consolidation
+            .as_mut()
+            .unwrap()
+            .candidate_ids
+            .push("candidate-256".into());
+        assert!(
+            request.validate().is_err(),
+            "candidate batch remains bounded"
+        );
+    }
+
+    #[test]
+    fn legacy_request_replay_keeps_its_exact_serialization() {
+        let request = request();
+        let original_json = serde_json::to_string(&request).unwrap();
+        assert!(!original_json.contains("grants"));
+        assert!(!original_json.contains("serialization_key"));
+        let decoded: BackendJobRequest = serde_json::from_str(&original_json).unwrap();
+        assert_eq!(decoded.grants, BackendJobGrants::default());
+        assert_eq!(
+            request.fingerprint().unwrap(),
+            decoded.fingerprint().unwrap()
+        );
+    }
+
     #[test]
     fn tracked_runtime_ids_are_fixed_size_for_maximum_job_ids() {
         let job_id = "j".repeat(256);

crates/workspace-server/src/hosts.rs

@@ -564,6 +564,8 @@ pub enum WorkerSpawnIntent {
         job_id: String,
         attempt_id: String,
         purpose: String,
+        input_revision: String,
+        subjektiv_consolidation: bool,
     },
     TicketRole {
         ticket_id: String,
@@ -5708,6 +5710,21 @@ fn runtime_create_worker_request(
         workspace_api: Some(workspace_api),
         memory_settings: request.resolved_memory_settings.clone(),
         subjektiv_attached: request.resolved_subjektiv_attached,
+        backend_job: match &request.intent {
+            WorkerSpawnIntent::BackendJob {
+                job_id,
+                attempt_id,
+                input_revision,
+                subjektiv_consolidation,
+                ..
+            } => Some(worker_runtime::catalog::BackendJobExecutionBinding {
+                job_id: job_id.clone(),
+                attempt_id: attempt_id.clone(),
+                input_revision: Some(input_revision.clone()),
+                subjektiv_consolidation: *subjektiv_consolidation,
+            }),
+            _ => None,
+        },
     }
 }
 

crates/workspace-server/src/latest_schema.sql

@@ -845,6 +845,7 @@ CREATE TABLE backend_jobs (
             purpose TEXT NOT NULL,
             input_revision TEXT NOT NULL,
             input_ref TEXT NOT NULL,
+            resource_key TEXT,
             request_json TEXT NOT NULL,
             intent_fingerprint TEXT NOT NULL,
             state TEXT NOT NULL CHECK (state IN ('pending', 'completed', 'failed', 'unknown')),
@@ -863,6 +864,8 @@ CREATE TABLE backend_jobs (
         );
 CREATE INDEX backend_jobs_active
         ON backend_jobs(workspace_id, state, updated_at);
+CREATE INDEX backend_jobs_resource
+        ON backend_jobs(workspace_id, resource_key, created_at);
 CREATE TABLE backend_job_attempts (
             workspace_id TEXT NOT NULL,
             job_id TEXT NOT NULL,

crates/workspace-server/src/profile_settings.rs

@@ -251,6 +251,27 @@ pub fn selector_for_workspace_candidate(
         .then(|| worker_runtime::catalog::ProfileSelector::Named(profile.to_string()))
 }
 
+/// Resolve a Host-owned launch selector against the complete existing registry.
+/// Browser candidates remain a narrower UI policy; this does not widen that surface.
+pub(crate) fn selector_for_registered_profile(
+    projection: &ProfileConfigProjection,
+    profile: &str,
+) -> Option<worker_runtime::catalog::ProfileSelector> {
+    let catalog = builtin_profile_catalog_snapshot();
+    let builtin = if profile.starts_with("builtin:") {
+        profile.to_string()
+    } else {
+        format!("builtin:{profile}")
+    };
+    if catalog.entrypoints.contains_key(&builtin) {
+        return Some(worker_runtime::catalog::ProfileSelector::Builtin(builtin));
+    }
+    projection
+        .entries
+        .contains_key(profile)
+        .then(|| worker_runtime::catalog::ProfileSelector::Named(profile.into()))
+}
+
 fn validate_prompt_projection_matches_state(
     workspace_id: &str,
     state: &WorkspaceConfigState,
@@ -401,8 +422,12 @@ pub fn build_virtual_profile_config_bundle_with_prompt_projection(
 ) -> Result<Option<ConfigBundle>> {
     validate_prompt_projection_matches_state(workspace_id, state, prompt_projection)?;
     let prompt_catalog = prompt_projection.catalog().clone();
-    let profile_selector = selector_for_builtin_candidate(selector)
-        .unwrap_or_else(|| worker_runtime::catalog::ProfileSelector::Named(selector.to_string()));
+    let profile_selector =
+        selector_for_registered_profile(projection, selector).ok_or_else(|| {
+            Error::InvalidInput(format!(
+                "Profile `{selector}` is not a registered launch selector"
+            ))
+        })?;
     let archive = match projection.entries.get(selector) {
         Some(entry) => Some(build_virtual_profile_archive(
             selector,

crates/workspace-server/src/runtime_subscription_tests.rs

@@ -74,6 +74,7 @@ fn create_request(name: &str) -> CreateWorkerRequest {
             language: "English".to_string(),
         }),
         subjektiv_attached: false,
+        backend_job: None,
     }
 }
 

crates/workspace-server/src/server.rs

@@ -1,3 +1,5 @@
+mod backend_job_profile;
+mod subjektiv_job;
 mod worker_operations;
 pub(crate) use worker_operations::{
     WorkerOperationContext, WorkspaceWorker, WorkspaceWorkerMethodSender,
@@ -3500,6 +3502,7 @@ impl WorkspaceApi {
         &self,
         request: &BackendJobRequest,
     ) -> Result<BackendJobReservation> {
+        self.resolve_backend_job_profile(request)?;
         let now = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
         let reservation =
             self.store
@@ -3548,6 +3551,19 @@ impl WorkspaceApi {
             return Ok(reservation);
         }
         let request = &reservation.job.request;
+        let (profile, config_bundle) = match self.resolve_backend_job_profile(request) {
+            Ok(resolved) => resolved,
+            Err(error) => {
+                self.finish_backend_job_attempt(
+                    &request.job_id,
+                    &reservation.attempt.attempt_id,
+                    BackendJobAttemptState::Failed,
+                    "profile_requirements",
+                    &error.to_string(),
+                )?;
+                return Err(error);
+            }
+        };
         let worker_input = request.worker_input(&reservation.attempt.attempt_id)?;
         let control_operation = WorkerControlOperation {
             operation_id: allocation_key(&reservation.attempt.attempt_id),
@@ -3562,22 +3578,24 @@ impl WorkspaceApi {
                     job_id: request.job_id.clone(),
                     attempt_id: reservation.attempt.attempt_id.clone(),
                     purpose: request.purpose.clone(),
+                    input_revision: request.input_revision.clone(),
+                    subjektiv_consolidation: request.grants.subjektiv_consolidation.is_some(),
                 },
                 acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
                     expected_segments: 0,
                 },
-                profile: ProfileSelector::Builtin(request.profile.clone()),
+                profile,
                 ticket_assignment: None,
                 initial_submit: Vec::new(),
                 workdir_attachment_requests: Vec::new(),
                 resolved_workdir_attachment_requests: Vec::new(),
                 resolved_workdir_attachments: Vec::new(),
-                resolved_config_bundle: None,
+                resolved_config_bundle: Some(config_bundle),
                 resolved_worker_observation_enabled: false,
                 resolved_worker_observation_grants: Vec::new(),
                 resolved_workspace_api: None,
                 resolved_memory_settings: None,
-                resolved_subjektiv_attached: false,
+                resolved_subjektiv_attached: request.grants.subjektiv_consolidation.is_some(),
                 resolved_control_operation: Some(control_operation),
             },
         ) {
@@ -3647,6 +3665,7 @@ impl WorkspaceApi {
                 job,
                 attempt,
                 replayed: true,
+                resource_reused: false,
             });
         }
 
@@ -3738,6 +3757,7 @@ impl WorkspaceApi {
             job,
             attempt,
             replayed: reservation.replayed,
+            resource_reused: reservation.resource_reused,
         })
     }
 
@@ -3801,6 +3821,7 @@ impl WorkspaceApi {
                     job,
                     attempt: attempt.clone(),
                     replayed: true,
+                    resource_reused: false,
                 }) {
                     Ok(reservation) => {
                         made_progress |=
@@ -4187,6 +4208,7 @@ impl WorkspaceApi {
                             job,
                             attempt,
                             replayed: true,
+                            resource_reused: false,
                         })
                     {
                         tracing::warn!(%error, "Backend Job reserved-attempt recovery failed");
@@ -4292,6 +4314,7 @@ impl WorkspaceApi {
         submission: &BackendJobResultSubmission,
     ) -> Result<BackendJobResultAcceptance> {
         let accepted_at = Utc::now().to_rfc3339_opts(SecondsFormat::Nanos, true);
+        let mut verified_submission = submission.clone();
         if let Some(job) = self
             .store
             .get_backend_job(&self.config.workspace_id, &submission.job_id)?
@@ -4334,10 +4357,41 @@ impl WorkspaceApi {
                 return Err(error);
             }
         }
+        if let Some(job) = self
+            .store
+            .get_backend_job(&self.config.workspace_id, &submission.job_id)?
+            && job.request.grants.subjektiv_consolidation.is_some()
+        {
+            let attempt = self
+                .store
+                .get_backend_job_attempt(
+                    &self.config.workspace_id,
+                    &submission.job_id,
+                    &submission.attempt_id,
+                )?
+                .ok_or_else(|| Error::InvalidInput("unknown consolidation attempt".into()))?;
+            if attempt.worker.as_ref() != Some(source_worker)
+                || job.current_attempt != attempt.attempt
+                || attempt.input_revision != submission.input_revision
+                || !matches!(
+                    attempt.state,
+                    BackendJobAttemptState::Dispatched | BackendJobAttemptState::Completed
+                )
+            {
+                return Err(Error::WorkspacePermissionDenied(
+                    "consolidation result requires the exact bound attempt".into(),
+                ));
+            }
+            if attempt.state != BackendJobAttemptState::Completed {
+                subjektiv_job::validate_result(self, &job.request, &attempt, &submission.result)?;
+            }
+            verified_submission.result =
+                subjektiv_job::with_dispositions(self, &job.request, &submission.result)?;
+        }
         let acceptance = self.store.accept_backend_job_result(
             &self.config.workspace_id,
             source_worker,
-            submission,
+            &verified_submission,
             &accepted_at,
         )?;
         if let Some(target) = acceptance.job.request.notification_target.as_ref() {
@@ -18175,6 +18229,13 @@ fn subjektiv_subject_scope(
         )
     })?;
     let worker = RuntimeWorkerRef::new(&source.runtime_id, worker_id);
+    if let Some(grant) = subjektiv_job::active_grant(api, workspace_id, &worker)? {
+        return Ok((
+            grant.subject_id,
+            worker,
+            SubjektivWorkerAuthority::Consolidation,
+        ));
+    }
     let lease = api
         .store
         .require_current_worker_singleton_owner(workspace_id, &worker)?
@@ -20006,7 +20067,8 @@ async fn scoped_subjektiv_memory_backend(
     Json(request): Json<server_api::SubjektivMemoryBackendRequest>,
 ) -> ApiResult<Json<server_api::SubjektivMemoryBackendResponse>> {
     validate_workspace_scope(&api, &path.workspace_id)?;
-    let (subject_id, _, authority) = subjektiv_subject_scope(&api, &path.workspace_id, &context)?;
+    let (subject_id, source_worker, authority) =
+        subjektiv_subject_scope(&api, &path.workspace_id, &context)?;
     let store = open_subjektiv_store(&api)?;
     let response = match request.operation {
         server_api::SubjektivMemoryBackendOperation::ResidentSummary(_) => {
@@ -20118,9 +20180,30 @@ async fn scoped_subjektiv_memory_backend(
                 "candidate listing",
             )?;
             let limit = bounded_subjektiv_limit(input.limit, SUBJEKTIV_QUERY_DEFAULT_LIMIT)?;
-            let mut candidates = store
-                .pending_staging_candidates(&subject_id, limit.saturating_add(1))
-                .map_err(subjektiv_store_error)?;
+            let mut candidates = if let Some(grant) =
+                subjektiv_job::active_grant(&api, &path.workspace_id, &source_worker)?
+            {
+                let mut pending = Vec::new();
+                for id in &grant.candidate_ids {
+                    if store
+                        .staging_resolution(&subject_id, id)
+                        .map_err(subjektiv_store_error)?
+                        .is_none()
+                    {
+                        if let Some(candidate) = store
+                            .staging_candidate(&subject_id, id)
+                            .map_err(subjektiv_store_error)?
+                        {
+                            pending.push(candidate);
+                        }
+                    }
+                }
+                pending
+            } else {
+                store
+                    .pending_staging_candidates(&subject_id, limit.saturating_add(1))
+                    .map_err(subjektiv_store_error)?
+            };
             let has_more = candidates.len() > limit;
             candidates.truncate(limit);
             server_api::SubjektivMemoryBackendResponse::Candidates(
@@ -20139,6 +20222,12 @@ async fn scoped_subjektiv_memory_backend(
                 SubjektivWorkerAuthority::Consolidation,
                 "candidate read",
             )?;
+            subjektiv_job::require_candidate_grant(
+                &api,
+                &path.workspace_id,
+                &source_worker,
+                &input.candidate_id,
+            )?;
             let candidate = store
                 .staging_candidate(&subject_id, &input.candidate_id)
                 .map_err(subjektiv_store_error)?
@@ -20167,6 +20256,12 @@ async fn scoped_subjektiv_memory_backend(
                 SubjektivWorkerAuthority::Consolidation,
                 "candidate decision",
             )?;
+            subjektiv_job::require_candidate_grant(
+                &api,
+                &path.workspace_id,
+                &source_worker,
+                &input.candidate_id,
+            )?;
             let receipt = store
                 .decide_candidate(&subject_id, subjektiv_candidate_decision(input)?)
                 .map_err(subjektiv_store_error)?;
@@ -20180,12 +20275,42 @@ async fn scoped_subjektiv_memory_backend(
                 SubjektivWorkerAuthority::Consolidation,
                 "surface generation preparation",
             )?;
+            if let Some(grant) =
+                subjektiv_job::active_grant(&api, &path.workspace_id, &source_worker)?
+            {
+                for id in &grant.candidate_ids {
+                    if store
+                        .staging_resolution(&subject_id, id)
+                        .map_err(subjektiv_store_error)?
+                        .is_none()
+                    {
+                        return Err(Error::InvalidInput(format!("surface generation requires a durable disposition for batch candidate `{id}`")).into());
+                    }
+                }
+            }
+            let binding = api
+                .store
+                .worker_registry_projection(&path.workspace_id, &source_worker)?
+                .and_then(|p| p.job);
             let generation = store
-                .prepare_surface_generation(&subject_id)
+                .prepare_job_surface_generation(
+                    &subject_id,
+                    binding
+                        .as_ref()
+                        .map(|b| (b.job_id.as_str(), b.attempt_id.as_str())),
+                )
                 .map_err(subjektiv_store_error)?;
-            server_api::SubjektivMemoryBackendResponse::SurfacePrepared(
-                subjektiv_surface_prepare_response(generation),
-            )
+            let resident = store
+                .resident_surface(&subject_id)
+                .map_err(subjektiv_store_error)?;
+            let mut response = subjektiv_surface_prepare_response(generation);
+            if resident.availability == crate::subjektiv::SurfaceAvailability::Ready {
+                response.current_snapshot_id = resident
+                    .snapshot
+                    .filter(|s| s.built_from_store_revision == response.store_revision)
+                    .map(|s| s.id);
+            }
+            server_api::SubjektivMemoryBackendResponse::SurfacePrepared(response)
         }
         server_api::SubjektivMemoryBackendOperation::PublishSurface(input) => {
             require_subjektiv_worker_authority(
@@ -20193,6 +20318,20 @@ async fn scoped_subjektiv_memory_backend(
                 SubjektivWorkerAuthority::Consolidation,
                 "surface publication",
             )?;
+            if let Some(binding) = api
+                .store
+                .worker_registry_projection(&path.workspace_id, &source_worker)?
+                .and_then(|p| p.job)
+            {
+                store
+                    .require_job_surface_generation(
+                        &subject_id,
+                        &input.generation_id,
+                        &binding.job_id,
+                        &binding.attempt_id,
+                    )
+                    .map_err(subjektiv_store_error)?;
+            }
             let points = input
                 .points
                 .into_iter()
@@ -20225,13 +20364,52 @@ async fn scoped_subjektiv_memory_backend(
                 SubjektivWorkerAuthority::Consolidation,
                 "surface generation failure",
             )?;
+            if let Some(binding) = api
+                .store
+                .worker_registry_projection(&path.workspace_id, &source_worker)?
+                .and_then(|p| p.job)
+            {
+                store
+                    .require_job_surface_generation(
+                        &subject_id,
+                        &input.generation_id,
+                        &binding.job_id,
+                        &binding.attempt_id,
+                    )
+                    .map_err(subjektiv_store_error)?;
+            }
             let revision = store
                 .fail_surface_generation(&subject_id, &input.generation_id, &input.reason_code)
                 .map_err(subjektiv_store_error)?;
+            let status = if let Some(binding) = api
+                .store
+                .worker_registry_projection(&path.workspace_id, &source_worker)?
+                .and_then(|p| p.job)
+            {
+                store
+                    .validate_job_surface_outcome(
+                        &subject_id,
+                        &input.generation_id,
+                        revision,
+                        "failed",
+                        None,
+                        Some(&input.reason_code),
+                        &binding.job_id,
+                        &binding.attempt_id,
+                    )
+                    .map_err(|e| {
+                        Error::RepositoryConflict(format!(
+                            "surface failure was not recorded for this attempt: {e}"
+                        ))
+                    })?;
+                "failed_confirmed"
+            } else {
+                "failed"
+            };
             server_api::SubjektivMemoryBackendResponse::SurfaceFailed(
                 server_api::SubjektivSurfaceFailureResponse {
                     store_revision: revision,
-                    status: "failed".into(),
+                    status: status.into(),
                 },
             )
         }
@@ -20244,6 +20422,7 @@ fn subjektiv_surface_prepare_response(
 ) -> server_api::SubjektivSurfacePrepareResponse {
     server_api::SubjektivSurfacePrepareResponse {
         generation_id: generation.id,
+        current_snapshot_id: None,
         store_revision: generation.store_revision,
         active_memory_count: generation.active_memory_count,
         materials: generation
@@ -21448,11 +21627,9 @@ async fn scoped_subjektiv_memory_consolidation(
         SubjektivWorkerAuthority::Subject,
         "consolidation request",
     )?;
-    Ok(Json(start_subjektiv_staging_consolidation(
-        api,
-        &subject_id,
-        operation,
-    )?))
+    Ok(Json(
+        start_subjektiv_staging_consolidation(api, &subject_id, operation).await?,
+    ))
 }
 
 const SUBJECT_CONSOLIDATION_THRESHOLD_FILES: usize = 5;
@@ -21511,7 +21688,7 @@ mod subject_surface_consolidation_policy_tests {
     }
 }
 
-fn start_subjektiv_staging_consolidation(
+async fn start_subjektiv_staging_consolidation(
     api: WorkspaceApi,
     subject_id: &str,
     operation: MemoryConsolidateStagingOperation,
@@ -21520,6 +21697,13 @@ fn start_subjektiv_staging_consolidation(
     let (candidate_count, total_bytes) = store
         .pending_staging_backlog(subject_id)
         .map_err(subjektiv_store_error)?;
+    if !subjektiv_job::retire_legacy_worker(&api, subject_id).await? {
+        return Ok(MemoryConsolidationOutput {
+            status: "skipped_legacy_inflight".into(),
+            summary: "Legacy owned subject consolidator has not completed foreground work; no Job may run concurrently.".into(),
+            candidate_count, total_bytes,
+        });
+    }
     let surface_availability = store
         .resident_surface(subject_id)
         .map_err(subjektiv_store_error)?
@@ -21550,135 +21734,7 @@ fn start_subjektiv_staging_consolidation(
         });
     }
 
-    let runtime_id = select_memory_consolidation_runtime(&api)?;
-    let singleton_key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}");
-    let input_content = format!(
-        "Process the delegated subject's {candidate_count} pending Memory candidate(s) ({total_bytes} bytes) with MemoryStagingList, MemoryStagingRead, subject Memory reads, and MemoryApplyCandidate. After this committed consolidation turn, the Host will rebuild the {:?} resident surface from confirmed Memory in a separate clean context.",
-        surface_availability
-    );
-    if let Some(output) = try_reuse_subject_consolidation_worker(
-        &api,
-        &singleton_key,
-        &input_content,
-        candidate_count,
-        total_bytes,
-    )? {
-        return Ok(output);
-    }
-    let result = api.spawn_workspace_worker(
-        &runtime_id,
-        WorkerSpawnRequest {
-            requested_worker_name: Some(SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE.to_string()),
-            singleton_key: Some(singleton_key),
-            intent: WorkerSpawnIntent::WorkspaceOrchestrator,
-            acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
-                expected_segments: 1,
-            },
-            profile: ProfileSelector::Builtin(SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE.to_string()),
-            ticket_assignment: None,
-            initial_submit: vec![Segment::text(input_content)],
-            workdir_attachment_requests: Vec::new(),
-            resolved_workdir_attachment_requests: Vec::new(),
-            resolved_workdir_attachments: Vec::new(),
-            resolved_config_bundle: None,
-            resolved_worker_observation_enabled: false,
-            resolved_worker_observation_grants: Vec::new(),
-            resolved_workspace_api: None,
-            resolved_memory_settings: None,
-            resolved_subjektiv_attached: true,
-            resolved_control_operation: None,
-        },
-    )?;
-    if result.state != InternalWorkerOperationState::Accepted {
-        return Ok(MemoryConsolidationOutput {
-            status: "skipped_spawn_rejected".into(),
-            summary: "Runtime rejected subject Memory consolidater spawn.".into(),
-            candidate_count,
-            total_bytes,
-        });
-    }
-    let Some(worker) = result.worker else {
-        return Ok(MemoryConsolidationOutput {
-            status: "skipped_spawn_missing_worker".into(),
-            summary: "Runtime accepted subject Memory consolidater spawn without returning a Worker summary.".into(),
-            candidate_count,
-            total_bytes,
-        });
-    };
-    Ok(MemoryConsolidationOutput {
-        status: "started".into(),
-        summary: format!(
-            "Started subject Memory consolidater '{}' for {candidate_count} staging candidate(s).",
-            worker.worker.worker_id
-        ),
-        candidate_count,
-        total_bytes,
-    })
-}
-
-fn try_reuse_subject_consolidation_worker(
-    api: &WorkspaceApi,
-    singleton_key: &str,
-    input_content: &str,
-    candidate_count: usize,
-    total_bytes: u64,
-) -> ApiResult<Option<MemoryConsolidationOutput>> {
-    let Some(owner) = api
-        .store
-        .current_worker_singleton_owner(&api.config.workspace_id, singleton_key)?
-    else {
-        return Ok(None);
-    };
-    let mut worker = api
-        .runtime
-        .worker(&owner.worker)
-        .map_err(|error| error.into_error())?;
-    worker.singleton_key = Some(owner.key);
-    require_dedicated_subjektiv_consolidation_worker(
-        &worker,
-        &api.config.workspace_id,
-        singleton_key,
-    )?;
-    if worker.state != "idle" {
-        return Ok(Some(MemoryConsolidationOutput {
-            status: "skipped_existing_not_idle".into(),
-            summary: format!(
-                "Existing subject Memory consolidater '{}' is '{}', not confirmed idle.",
-                worker.worker.worker_id, worker.state
-            ),
-            candidate_count,
-            total_bytes,
-        }));
-    }
-    let input = api
-        .runtime
-        .send_input(
-            &worker.worker,
-            WorkerInputRequest {
-                kind: WorkerInputKind::User,
-                content: input_content.to_string(),
-                submission_request_id: None,
-                segments: None,
-            },
-        )
-        .map_err(|error| error.into_error())?;
-    if input.state != InternalWorkerOperationState::Accepted {
-        return Ok(Some(MemoryConsolidationOutput {
-            status: "skipped_existing_input_rejected".into(),
-            summary: "Existing subject Memory consolidater rejected new input.".into(),
-            candidate_count,
-            total_bytes,
-        }));
-    }
-    Ok(Some(MemoryConsolidationOutput {
-        status: "reused".into(),
-        summary: format!(
-            "Reused subject Memory consolidater '{}' for {candidate_count} staging candidate(s).",
-            worker.worker.worker_id
-        ),
-        candidate_count,
-        total_bytes,
-    }))
+    subjektiv_job::dispatch(&api, subject_id, candidate_count, total_bytes)
 }
 
 async fn scoped_memory_consolidation(
@@ -33409,6 +33465,7 @@ impl IntoResponse for ApiError {
 
 #[cfg(test)]
 mod tests {
+    mod subjektiv_jobs_tests;
     mod worker_operations_tests;
     include!("server_workspace_config_tests.rs");
     use super::*;
@@ -44862,6 +44919,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -44963,6 +45022,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: Some(notification_target.clone()),
             notification_target: Some(notification_target.clone()),
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45069,6 +45130,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45105,10 +45168,10 @@ mod tests {
             Some("backend_job_cleanup_removal_rejected")
         );
 
-        let second = api.retry_backend_job(&request.job_id).unwrap();
-        let second_worker = second.attempt.worker.clone().unwrap();
-        seed_worker_session_for_cleanup(temp.path(), &second_worker);
-        assert_ne!(first_worker, second_worker);
+        assert!(
+            api.retry_backend_job(&request.job_id).is_err(),
+            "retry must not overlap an incompletely cleaned attempt"
+        );
         let _ = set_worker_retention(
             api.clone(),
             first_worker.runtime_id.clone(),
@@ -45125,6 +45188,11 @@ mod tests {
             BackendJobWorkerCleanupState::Completed,
         )
         .await;
+        let second = api.retry_backend_job(&request.job_id).unwrap();
+        let second_worker = second.attempt.worker.clone().unwrap();
+        seed_worker_session_for_cleanup(temp.path(), &second_worker);
+        assert_ne!(first_worker, second_worker);
+        api.recover_backend_jobs().unwrap();
         assert!(api.runtime.worker(&second_worker).is_ok());
 
         api.accept_backend_job_result(
@@ -45171,6 +45239,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45212,6 +45282,7 @@ mod tests {
                     workspace_api: Some(test_worker_workspace_api(EMBEDDED_WORKER_RUNTIME_ID)),
                     memory_settings: Some(test_worker_memory_settings()),
                     subjektiv_attached: false,
+                    backend_job: None,
                 },
             )
             .unwrap();
@@ -45348,6 +45419,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45410,6 +45483,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45543,6 +45618,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: Some(target.clone()),
             limits: crate::backend_job::BackendJobLimits {
@@ -45625,6 +45702,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45663,6 +45742,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45728,6 +45809,8 @@ mod tests {
                 input: serde_json::json!({"index": index}),
                 instruction: "Return a structured fixture result.".to_string(),
                 profile: "builtin:backend-job".to_string(),
+                grants: Default::default(),
+                serialization_key: None,
                 source_worker: None,
                 notification_target: None,
                 limits: crate::backend_job::BackendJobLimits {
@@ -45773,6 +45856,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45816,6 +45901,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Return a structured check result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits::default(),
@@ -45905,6 +45992,8 @@ mod tests {
             input: serde_json::json!({"value": "immutable"}),
             instruction: "Wait for a structured result.".to_string(),
             profile: "builtin:backend-job".to_string(),
+            grants: Default::default(),
+            serialization_key: None,
             source_worker: None,
             notification_target: None,
             limits: crate::backend_job::BackendJobLimits {
@@ -55932,6 +56021,7 @@ mod tests {
             workspace_api: None,
             memory_settings: Some(memory_settings),
             subjektiv_attached: false,
+            backend_job: None,
         }
     }
 
@@ -60280,14 +60370,9 @@ VALUES ('0192f0e8-4d84-7d6e-a000-000000000001', 'ticket', 3);
         let error = subjektiv_subject_scope(&api, TEST_WORKSPACE_ID, &forged_context).unwrap_err();
         assert_eq!(error.into_response().status(), StatusCode::FORBIDDEN);
         assert!(
-            try_reuse_subject_consolidation_worker(
-                &api,
-                forged_key,
-                "process pending subject Memory",
-                1,
-                1,
-            )
-            .is_err()
+            subjektiv_job::retire_legacy_worker(&api, "forged-subject")
+                .await
+                .is_err()
         );
     }
 

crates/workspace-server/src/server/backend_job_profile.rs

@@ -0,0 +1,101 @@
+//! Profile recipes are selected by the caller. Execution grants remain Host owned.
+use super::*;
+
+impl WorkspaceApi {
+    pub(super) fn resolve_backend_job_profile(
+        &self,
+        request: &BackendJobRequest,
+    ) -> Result<(ProfileSelector, worker_runtime::config_bundle::ConfigBundle)> {
+        request.validate()?;
+        let state = self
+            .config_store
+            .load_workspace_config(self.workspace_id())?
+            .ok_or_else(|| {
+                Error::Config("Backend Job requires Workspace Profile configuration".into())
+            })?;
+        let projection = crate::profile_settings::project_profiles_from_workspace_config(
+            self.workspace_id(),
+            &state,
+        )?;
+        let requested_profile = if request.profile == "default" {
+            projection
+                .settings
+                .default_profile
+                .as_deref()
+                .ok_or_else(|| {
+                    Error::InvalidInput("Workspace has no configured default Job Profile".into())
+                })?
+        } else {
+            request.profile.as_str()
+        };
+        let selector = crate::profile_settings::selector_for_registered_profile(
+            &projection,
+            requested_profile,
+        )
+        .ok_or_else(|| {
+            Error::InvalidInput(format!(
+                "Backend Job Profile `{}` is not a published registry candidate",
+                request.profile
+            ))
+        })?;
+        let profile_key = match &selector {
+            ProfileSelector::Builtin(key) | ProfileSelector::Named(key) => key.as_str(),
+        };
+        let prompts = self
+            .prompt_projection_cache
+            .resolve(self.workspace_id(), &state)?;
+        let bundle =
+            crate::profile_settings::build_virtual_profile_config_bundle_with_prompt_projection(
+                &projection,
+                &state,
+                self.workspace_id(),
+                &self.config.workspace_created_at,
+                profile_key,
+                prompts.as_ref(),
+            )?
+            .ok_or_else(|| Error::Config("Backend Job Profile bundle is missing".into()))?;
+        let manifest = crate::profile_settings::resolve_profile_manifest_from_config_bundle(
+            &bundle,
+            profile_key,
+        )?;
+        validate_backend_job_profile_requirements(&manifest, request)?;
+        Ok((selector, bundle))
+    }
+}
+
+fn validate_backend_job_profile_requirements(
+    manifest: &manifest::WorkerManifest,
+    request: &BackendJobRequest,
+) -> Result<()> {
+    let f = &manifest.feature;
+    // These tools require domain/parent/Workdir authority which a bounded Job
+    // deliberately does not inherit. Reject, rather than disabling features or
+    // substituting a different recipe under the caller's selector.
+    if f.memory.enabled()
+        || f.worker.enabled
+        || f.sub_worker.enabled
+        || f.flow.enabled
+        || f.workspace_worker_discovery.enabled
+        || f.objective.enabled
+        || f.manage_workdir.enabled
+        || f.workspace_config.enabled
+        || f.ticket.enabled
+        || f.merge_request.show
+        || f.merge_request.open
+        || f.merge_request.review
+        || f.merge_request.readiness_check
+        || f.merge_request.complete
+        || f.orchestration.enabled
+        || !manifest.mcp.stdio_servers.is_empty()
+    {
+        return Err(Error::WorkspacePermissionDenied("Backend Job Profile requires authority not granted to this Job (Workspace mutation, Memory, child/peer, Workdir, Flow or MCP)".into()));
+    }
+    let consolidation = request.grants.subjektiv_consolidation.is_some();
+    if f.subjektiv.enabled() != consolidation
+        || f.subjektiv.profile.consolidation_tools != consolidation
+        || (consolidation && f.subjektiv.profile.extraction.enabled)
+    {
+        return Err(Error::WorkspacePermissionDenied("Backend Job Profile subjektiv requirements do not match the explicit consolidation grant".into()));
+    }
+    Ok(())
+}

crates/workspace-server/src/server/subjektiv_job.rs

@@ -0,0 +1,311 @@
+//! subjektiv is a Job client, not a second runner or Worker registry.
+use super::*;
+use crate::backend_job::{BackendJobGrants, SubjektivConsolidationGrant};
+
+const PURPOSE: &str = "subjektiv_consolidation";
+const CANDIDATE_BATCH_LIMIT: usize = 100;
+
+/// Load Host authority from the authenticated Worker registry and current attempt.
+/// Model input, Profile tags, display names and a subject-body lease are not grants.
+pub(super) fn active_grant(
+    api: &WorkspaceApi,
+    workspace_id: &str,
+    worker: &RuntimeWorkerRef,
+) -> ApiResult<Option<SubjektivConsolidationGrant>> {
+    let Some(binding) = api
+        .store
+        .worker_registry_projection(workspace_id, worker)?
+        .and_then(|p| p.job)
+    else {
+        return Ok(None);
+    };
+    let job = api
+        .store
+        .get_backend_job(workspace_id, &binding.job_id)?
+        .ok_or_else(|| Error::WorkspacePermissionDenied("Job binding has no intent".into()))?;
+    let attempt = api
+        .store
+        .get_backend_job_attempt(workspace_id, &binding.job_id, &binding.attempt_id)?
+        .ok_or_else(|| Error::WorkspacePermissionDenied("Job binding has no attempt".into()))?;
+    if job.state != BackendJobState::Pending
+        || attempt.state != BackendJobAttemptState::Dispatched
+        || attempt.worker.as_ref() != Some(worker)
+        || job.current_attempt != attempt.attempt
+        || attempt.input_revision != job.request.input_revision
+        || chrono::DateTime::parse_from_rfc3339(&attempt.deadline_at)
+            .map_err(|e| Error::Store(format!("invalid Job deadline: {e}")))?
+            <= Utc::now()
+    {
+        return Err(Error::WorkspacePermissionDenied(
+            "subjektiv grant requires the current live Job attempt".into(),
+        )
+        .into());
+    }
+    Ok(job.request.grants.subjektiv_consolidation)
+}
+
+pub(super) fn require_candidate_grant(
+    api: &WorkspaceApi,
+    workspace_id: &str,
+    worker: &RuntimeWorkerRef,
+    candidate_id: &str,
+) -> ApiResult<()> {
+    if let Some(grant) = active_grant(api, workspace_id, worker)? {
+        if !grant.candidate_ids.iter().any(|id| id == candidate_id) {
+            return Err(Error::WorkspacePermissionDenied(
+                "candidate is outside this Job's immutable batch".into(),
+            )
+            .into());
+        }
+    }
+    Ok(()) // legacy owned consolidators may finish before the cutover fence
+}
+
+/// Retire only the persisted old consolidation owner. Never interrupt a foreground
+/// candidate decision. Once Idle is confirmed, stop cancels any legacy surface
+/// background work; the new Job regenerates it and does not infer surface success.
+pub(super) async fn retire_legacy_worker(api: &WorkspaceApi, subject_id: &str) -> ApiResult<bool> {
+    let key = format!("{SUBJEKTIV_CONSOLIDATION_SINGLETON_PREFIX}{subject_id}");
+    let Some(owner) = api
+        .store
+        .current_worker_singleton_owner(api.workspace_id(), &key)?
+    else {
+        return Ok(true);
+    };
+    let mut summary = api
+        .runtime
+        .worker(&owner.worker)
+        .map_err(|e| e.into_error())?;
+    summary.singleton_key = Some(owner.key.clone());
+    require_dedicated_subjektiv_consolidation_worker(&summary, api.workspace_id(), &key)?;
+    let stopped = summary.state.eq_ignore_ascii_case("stopped");
+    let idle = summary.state.eq_ignore_ascii_case("idle")
+        && summary
+            .worker_state
+            .as_ref()
+            .is_some_and(|s| s.state == protocol::WorkerState::Idle);
+    if !stopped && !idle {
+        return Ok(false);
+    }
+    if api
+        .store
+        .current_worker_singleton_owner(api.workspace_id(), &key)?
+        .as_ref()
+        .map(|o| &o.worker)
+        != Some(&owner.worker)
+    {
+        return Err(Error::RepositoryConflict(
+            "legacy consolidation owner moved during cutover".into(),
+        )
+        .into());
+    }
+    if !summary.state.eq_ignore_ascii_case("stopped") {
+        let stopped = api
+            .runtime
+            .stop_worker(
+                &owner.worker,
+                WorkerLifecycleRequest {
+                    reason: Some(
+                        "Retire legacy subjektiv consolidator after foreground completion".into(),
+                    ),
+                    ticket_assignment: None,
+                },
+            )
+            .map_err(|e| e.into_error())?;
+        if stopped.state != InternalWorkerOperationState::Accepted {
+            return Ok(false);
+        }
+    }
+    if !api
+        .runtime
+        .worker(&owner.worker)
+        .map_err(|e| e.into_error())?
+        .state
+        .eq_ignore_ascii_case("stopped")
+    {
+        return Ok(false);
+    }
+    let removal = WorkerRemovalService::new(api)
+        .execute_target_removal(
+            api.runtime.as_ref(),
+            &owner.worker,
+            "subjektiv consolidation Job cutover",
+            None,
+        )
+        .await
+        .map_err(Error::Store)?;
+    if removal.status != StatusCode::OK.as_u16() {
+        return Err(Error::Store(format!(
+            "legacy consolidation cleanup rejected: {}",
+            removal.body
+        ))
+        .into());
+    }
+    Ok(api
+        .store
+        .current_worker_singleton_owner(api.workspace_id(), &key)?
+        .is_none())
+}
+
+pub(super) fn dispatch(
+    api: &WorkspaceApi,
+    subject_id: &str,
+    candidate_count: usize,
+    total_bytes: u64,
+) -> ApiResult<MemoryConsolidationOutput> {
+    let store = open_subjektiv_store(api)?;
+    let subject = store
+        .subject(subject_id)
+        .map_err(subjektiv_store_error)?
+        .ok_or_else(|| Error::SubjektivSubjectNotFound(subject_id.into()))?;
+    let candidate_ids: Vec<_> = store
+        .pending_staging_candidates(subject_id, CANDIDATE_BATCH_LIMIT)
+        .map_err(subjektiv_store_error)?
+        .into_iter()
+        .map(|c| c.id)
+        .collect();
+    let batch_len = candidate_ids.len();
+    let request = BackendJobRequest {
+        job_id: format!("subjektiv-consolidation:{}", Uuid::new_v4()),
+        purpose: PURPOSE.into(),
+        input_revision: subject.store_revision.to_string(),
+        input_ref: format!("subjektiv://{subject_id}/consolidation"),
+        input: serde_json::json!({"subject_id": subject_id, "candidate_ids": candidate_ids}),
+        instruction: "Resolve ONLY the immutable candidate_ids batch using MemoryStagingList/Read, confirmed Memory reads and MemoryApplyCandidate. Already resolved candidates are retained receipts, not work to repeat. Leave newly arriving candidates to a subsequent Job. When every batch candidate has a durable disposition, call SubmitBackendJobResult with {subject_id, candidate_ids}; its Host capability awaits bounded clean-context surface generation and adds the actual surface outcome before submission. Final prose is not completion.".into(),
+        // Selection belongs here, never in the common runner's purpose switch.
+        profile: format!("builtin:{SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE}"),
+        grants: BackendJobGrants { subjektiv_consolidation: Some(SubjektivConsolidationGrant { subject_id: subject_id.into(), candidate_ids }) },
+        serialization_key: None,
+        source_worker: None,
+        notification_target: None,
+        limits: crate::backend_job::BackendJobLimits { timeout_seconds: 600, ..Default::default() },
+    };
+    let reservation = api.dispatch_backend_job(&request)?;
+    Ok(MemoryConsolidationOutput {
+        status: if reservation.resource_reused {
+            "skipped_existing_job"
+        } else {
+            "started"
+        }
+        .into(),
+        summary: format!(
+            "Subject consolidation Job '{}' ({:?}) owns an immutable batch of {} candidate(s); remaining/new candidates stay pending for the next request. No persistent subject consolidator is reused.",
+            reservation.job.request.job_id,
+            reservation.attempt.state,
+            if reservation.resource_reused {
+                reservation
+                    .job
+                    .request
+                    .grants
+                    .subjektiv_consolidation
+                    .as_ref()
+                    .map_or(0, |g| g.candidate_ids.len())
+            } else {
+                batch_len
+            }
+        ),
+        candidate_count,
+        total_bytes,
+    })
+}
+
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct ConsolidationResult {
+    subject_id: String,
+    candidate_ids: Vec<String>,
+    surface: SurfaceResult,
+}
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct SurfaceResult {
+    availability: String,
+    generation_id: String,
+    store_revision: u64,
+    #[serde(default)]
+    snapshot_id: Option<String>,
+    #[serde(default)]
+    reason_code: Option<String>,
+}
+
+pub(super) fn validate_result(
+    api: &WorkspaceApi,
+    request: &BackendJobRequest,
+    attempt: &BackendJobAttemptRecord,
+    value: &serde_json::Value,
+) -> Result<()> {
+    let Some(grant) = &request.grants.subjektiv_consolidation else {
+        return Ok(());
+    };
+    let result: ConsolidationResult = serde_json::from_value(value.clone())
+        .map_err(|e| Error::InvalidInput(format!("invalid consolidation result: {e}")))?;
+    if result.subject_id != grant.subject_id || result.candidate_ids != grant.candidate_ids {
+        return Err(Error::InvalidInput(
+            "consolidation result does not match the immutable Subject/batch grant".into(),
+        ));
+    }
+    let store =
+        crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration)
+            .map_err(|e| Error::Store(e.to_string()))?;
+    for candidate_id in &grant.candidate_ids {
+        if store
+            .staging_resolution(&grant.subject_id, candidate_id)
+            .map_err(|e| Error::Store(e.to_string()))?
+            .is_none()
+        {
+            return Err(Error::InvalidInput(format!(
+                "consolidation candidate `{candidate_id}` is unresolved"
+            )));
+        }
+    }
+    store
+        .validate_job_surface_outcome(
+            &grant.subject_id,
+            &result.surface.generation_id,
+            result.surface.store_revision,
+            &result.surface.availability,
+            result.surface.snapshot_id.as_deref(),
+            result.surface.reason_code.as_deref(),
+            &request.job_id,
+            &attempt.attempt_id,
+        )
+        .map_err(|e| Error::InvalidInput(format!("consolidation surface result mismatch: {e}")))
+}
+
+/// Persist actual candidate dispositions alongside the separate surface outcome.
+/// Dispositions are immutable existing records, not model assertions or another
+/// apply ledger. This enrichment is deterministic on exact result replay.
+pub(super) fn with_dispositions(
+    api: &WorkspaceApi,
+    request: &BackendJobRequest,
+    value: &serde_json::Value,
+) -> Result<serde_json::Value> {
+    let grant = request
+        .grants
+        .subjektiv_consolidation
+        .as_ref()
+        .ok_or_else(|| Error::InvalidInput("missing consolidation grant".into()))?;
+    let store =
+        crate::subjektiv::SubjektivStore::open(&api.feature_storage, &api.subjektiv_registration)
+            .map_err(|e| Error::Store(e.to_string()))?;
+    let mut dispositions = Vec::new();
+    for id in &grant.candidate_ids {
+        let resolution = store
+            .staging_resolution(&grant.subject_id, id)
+            .map_err(|e| Error::Store(e.to_string()))?
+            .ok_or_else(|| {
+                Error::InvalidInput(format!("candidate `{id}` has no durable disposition"))
+            })?;
+        dispositions
+            .push(serde_json::to_value(resolution).map_err(|e| Error::Store(e.to_string()))?);
+    }
+    let mut enriched = value.clone();
+    enriched
+        .as_object_mut()
+        .ok_or_else(|| Error::InvalidInput("consolidation result must be an object".into()))?
+        .insert(
+            "candidate_dispositions".into(),
+            serde_json::Value::Array(dispositions),
+        );
+    Ok(enriched)
+}

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

@@ -0,0 +1,609 @@
+use super::*;
+use crate::subjektiv::{
+    CandidateDecision, CandidateDecisionRequest, MemoryDraft, MemoryRevisionTarget, SubjectRole,
+    SubjectStagingRecord, SurfaceAvailability,
+};
+use memory::extract::{CandidateKind, ExtractedCandidate, StagingEvidence, StagingRecord};
+use memory::schema::SourceRef;
+use memory::schema::{EvidenceKind, EvidenceOrigin, EvidenceOriginKind};
+
+fn stage(store: &crate::subjektiv::SubjektivStore, subject: &str, id: &str) {
+    let origin = EvidenceOrigin {
+        kind: EvidenceOriginKind::HumanInput,
+        account_id: Some("test-human".into()),
+        workspace_id: Some(TEST_WORKSPACE_ID.into()),
+        runtime_id: None,
+        worker_id: None,
+        flow_selector: None,
+        flow_definition_id: None,
+        flow_definition_revision: None,
+    };
+    let record = StagingRecord::from_candidate(
+        id,
+        "test-extract",
+        SourceRef {
+            segment_id: "segment-1".into(),
+            range: [1, 2],
+        },
+        ExtractedCandidate {
+            kind: CandidateKind::Constraint,
+            claim: "Keep receipts and confirmed memory".into(),
+            why_useful: "No repeated application".into(),
+            staleness: None,
+            evidence_ids: vec!["e1".into()],
+        },
+        vec![StagingEvidence {
+            id: "e1".into(),
+            kind: EvidenceKind::new(EvidenceKind::MESSAGE),
+            entry_range: Some([1, 2]),
+            origin: Some(origin),
+            excerpt: Some("Keep confirmed memory".into()),
+            summary: None,
+        }],
+        vec![],
+    );
+    store
+        .stage_candidate(SubjectStagingRecord::attach(subject, record))
+        .unwrap();
+}
+fn context(worker: &RuntimeWorkerRef) -> server_api::ServerRequestContext {
+    server_api::ServerRequestContext {
+        actor: None,
+        worker_source: None,
+        runtime_source: Some(server_api::ServerRuntimeSource {
+            runtime_id: worker.runtime_id.clone(),
+            worker_id: Some(worker.worker_id.clone()),
+        }),
+        origin: None,
+        transport_headers: vec![],
+    }
+}
+async fn op(
+    api: &WorkspaceApi,
+    worker: &RuntimeWorkerRef,
+    operation: server_api::SubjektivMemoryBackendOperation,
+) -> ApiResult<Json<server_api::SubjektivMemoryBackendResponse>> {
+    scoped_subjektiv_memory_backend(
+        State(api.clone()),
+        AxumPath(ScopedWorkspacePath {
+            workspace_id: TEST_WORKSPACE_ID.into(),
+        }),
+        context(worker),
+        Json(server_api::SubjektivMemoryBackendRequest { operation }),
+    )
+    .await
+}
+fn resource(api: &WorkspaceApi, subject: &str) -> crate::backend_job::BackendJobRecord {
+    api.store
+        .find_active_backend_job_for_resource(
+            TEST_WORKSPACE_ID,
+            &crate::backend_job::subjektiv_serialization_key(subject),
+        )
+        .unwrap()
+        .unwrap()
+}
+fn current_attempt(
+    api: &WorkspaceApi,
+    job: &crate::backend_job::BackendJobRecord,
+) -> BackendJobAttemptRecord {
+    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()
+}
+
+#[tokio::test]
+async fn subjektiv_job_bounds_batch_auth_partial_receipts_surface_and_cleanup() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(dir.path()).await;
+    let store = open_subjektiv_store(&api).unwrap();
+    let subject = store
+        .create_subject(SubjectRole::new("test-subject").unwrap())
+        .unwrap();
+    stage(&store, &subject.id, "candidate-first");
+    let started = start_subjektiv_staging_consolidation(
+        api.clone(),
+        &subject.id,
+        MemoryConsolidateStagingOperation { force: true },
+    )
+    .await
+    .unwrap();
+    assert_eq!(started.status, "started");
+    let job = resource(&api, &subject.id);
+    assert_eq!(
+        job.request.profile,
+        "builtin:subjektiv-memory-consolidation"
+    );
+    let attempt = current_attempt(&api, &job);
+    let worker = attempt.worker.as_ref().unwrap();
+    assert_eq!(execution.take_inputs().len(), 1);
+    assert!(
+        api.store
+            .require_current_worker_singleton_owner(TEST_WORKSPACE_ID, worker)
+            .unwrap()
+            .is_none()
+    );
+    assert_eq!(
+        subjektiv_subject_scope(&api, TEST_WORKSPACE_ID, &context(worker))
+            .unwrap()
+            .0,
+        subject.id
+    );
+    assert!(subjektiv_subject_scope(&api, "foreign", &context(worker)).is_err());
+    let ordinary = spawn_ticket_check_source(&api, "not-a-consolidator");
+    assert!(subjektiv_subject_scope(&api, TEST_WORKSPACE_ID, &context(&ordinary)).is_err());
+    // A premature result cannot generate a surface or complete unresolved work.
+    assert!(
+        op(
+            &api,
+            worker,
+            server_api::SubjektivMemoryBackendOperation::PrepareSurface(
+                server_api::SubjektivSurfacePrepareRequest {}
+            )
+        )
+        .await
+        .is_err()
+    );
+    stage(&store, &subject.id, "candidate-arrived-later");
+    let again = start_subjektiv_staging_consolidation(
+        api.clone(),
+        &subject.id,
+        MemoryConsolidateStagingOperation { force: true },
+    )
+    .await
+    .unwrap();
+    assert_eq!(again.status, "skipped_existing_job");
+    assert!(execution.take_inputs().is_empty());
+    assert_eq!(resource(&api, &subject.id).request, job.request);
+    assert!(
+        subjektiv_job::require_candidate_grant(
+            &api,
+            TEST_WORKSPACE_ID,
+            worker,
+            "candidate-arrived-later"
+        )
+        .is_err()
+    );
+    let decision = CandidateDecisionRequest {
+        request_id: "exact-apply".into(),
+        candidate_id: "candidate-first".into(),
+        reason: "retain the learned constraint".into(),
+        decision: CandidateDecision::Apply {
+            target: MemoryRevisionTarget::Create,
+            draft: MemoryDraft::active(
+                CandidateKind::Constraint,
+                "Keep committed memory",
+                "Applicable across failures",
+                "Prevents duplicate application",
+                "confirmed from candidate",
+            ),
+        },
+    };
+    let receipt = store
+        .decide_candidate(&subject.id, decision.clone())
+        .unwrap();
+    let replay = store.decide_candidate(&subject.id, decision).unwrap();
+    assert_eq!(
+        receipt.memory.as_ref().unwrap().id,
+        replay.memory.as_ref().unwrap().id
+    );
+    assert_eq!(receipt.store_revision, replay.store_revision);
+    let generation = store
+        .prepare_job_surface_generation(
+            &subject.id,
+            Some((&job.request.job_id, &attempt.attempt_id)),
+        )
+        .unwrap();
+    let Json(failed) = op(
+        &api,
+        worker,
+        server_api::SubjektivMemoryBackendOperation::FailSurface(
+            server_api::SubjektivSurfaceFailureRequest {
+                generation_id: generation.id.clone(),
+                reason_code: "editor_failed".into(),
+            },
+        ),
+    )
+    .await
+    .unwrap();
+    assert!(
+        matches!(failed, server_api::SubjektivMemoryBackendResponse::SurfaceFailed(output) if output.status == "failed_confirmed" && output.store_revision == generation.store_revision)
+    );
+    let result = serde_json::json!({"subject_id": subject.id, "candidate_ids": ["candidate-first"], "surface": {"availability":"failed", "generation_id":generation.id, "store_revision":generation.store_revision,"reason_code":"editor_failed"}});
+    let submission = BackendJobResultSubmission {
+        job_id: job.request.job_id.clone(),
+        attempt_id: attempt.attempt_id.clone(),
+        input_revision: job.request.input_revision.clone(),
+        result,
+    };
+    assert!(
+        api.accept_backend_job_result(&ordinary, &submission)
+            .is_err()
+    );
+    let mut forged = submission.clone();
+    forged.result["surface"]["reason_code"] = serde_json::json!("not_recorded");
+    assert!(api.accept_backend_job_result(worker, &forged).is_err());
+    seed_worker_session_for_cleanup(dir.path(), worker);
+    let accepted = api.accept_backend_job_result(worker, &submission).unwrap();
+    assert_eq!(accepted.job.state, BackendJobState::Completed);
+    assert_eq!(
+        store.resident_surface(&subject.id).unwrap().availability,
+        SurfaceAvailability::Failed
+    );
+    assert_eq!(store.pending_staging_backlog(&subject.id).unwrap().0, 1);
+    assert!(
+        store
+            .memory(&subject.id, &receipt.memory.unwrap().id)
+            .unwrap()
+            .is_some()
+    );
+    assert!(subjektiv_subject_scope(&api, TEST_WORKSPACE_ID, &context(worker)).is_err());
+    wait_for_backend_job_worker_cleanup(
+        &api,
+        &job.request.job_id,
+        &attempt.attempt_id,
+        BackendJobWorkerCleanupState::Completed,
+    )
+    .await;
+    assert!(
+        api.store
+            .get_backend_job(TEST_WORKSPACE_ID, &job.request.job_id)
+            .unwrap()
+            .unwrap()
+            .result
+            .is_some()
+    );
+    let next = start_subjektiv_staging_consolidation(
+        api.clone(),
+        &subject.id,
+        MemoryConsolidateStagingOperation { force: true },
+    )
+    .await
+    .unwrap();
+    assert_eq!(next.status, "started");
+    assert_eq!(
+        resource(&api, &subject.id)
+            .request
+            .grants
+            .subjektiv_consolidation
+            .unwrap()
+            .candidate_ids,
+        ["candidate-arrived-later"]
+    );
+}
+
+#[tokio::test]
+async fn subjektiv_job_surface_only_requires_attempt_owned_generation_and_real_outcome() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, _) = test_api_with_recording_backend(dir.path()).await;
+    let store = open_subjektiv_store(&api).unwrap();
+    let subject = store
+        .create_subject(SubjectRole::new("surface-only").unwrap())
+        .unwrap();
+    let old = store.prepare_surface_generation(&subject.id).unwrap();
+    assert_eq!(
+        start_subjektiv_staging_consolidation(
+            api.clone(),
+            &subject.id,
+            MemoryConsolidateStagingOperation { force: false }
+        )
+        .await
+        .unwrap()
+        .status,
+        "started"
+    );
+    let job = resource(&api, &subject.id);
+    let attempt = current_attempt(&api, &job);
+    let worker = attempt.worker.as_ref().unwrap();
+    assert!(
+        store
+            .require_job_surface_generation(
+                &subject.id,
+                &old.id,
+                &job.request.job_id,
+                &attempt.attempt_id
+            )
+            .is_err()
+    );
+    let generation = store
+        .prepare_job_surface_generation(
+            &subject.id,
+            Some((&job.request.job_id, &attempt.attempt_id)),
+        )
+        .unwrap();
+    assert!(
+        store
+            .require_job_surface_generation(
+                &subject.id,
+                &generation.id,
+                &job.request.job_id,
+                "wrong-attempt"
+            )
+            .is_err()
+    );
+    let snapshot = store
+        .publish_surface_generation(&subject.id, &generation.id, vec![])
+        .unwrap();
+    assert!(
+        op(
+            &api,
+            worker,
+            server_api::SubjektivMemoryBackendOperation::FailSurface(
+                server_api::SubjektivSurfaceFailureRequest {
+                    generation_id: generation.id.clone(),
+                    reason_code: "editor_failed".into()
+                }
+            )
+        )
+        .await
+        .is_err(),
+        "ready protection is not a confirmed failed outcome"
+    );
+    let Json(prepared) = op(
+        &api,
+        worker,
+        server_api::SubjektivMemoryBackendOperation::PrepareSurface(
+            server_api::SubjektivSurfacePrepareRequest {},
+        ),
+    )
+    .await
+    .unwrap();
+    assert!(
+        matches!(prepared, server_api::SubjektivMemoryBackendResponse::SurfacePrepared(output) if output.current_snapshot_id.as_deref() == Some(snapshot.id.as_str()))
+    );
+    seed_worker_session_for_cleanup(dir.path(), worker);
+    let submission = BackendJobResultSubmission {
+        job_id: job.request.job_id.clone(),
+        attempt_id: attempt.attempt_id.clone(),
+        input_revision: job.request.input_revision.clone(),
+        result: serde_json::json!({"subject_id":subject.id,"candidate_ids":[],"surface":{"availability":"ready","generation_id":generation.id,"store_revision":generation.store_revision,"snapshot_id":snapshot.id}}),
+    };
+    api.accept_backend_job_result(worker, &submission).unwrap();
+    wait_for_backend_job_worker_cleanup(
+        &api,
+        &job.request.job_id,
+        &attempt.attempt_id,
+        BackendJobWorkerCleanupState::Completed,
+    )
+    .await;
+    assert_eq!(
+        start_subjektiv_staging_consolidation(
+            api.clone(),
+            &subject.id,
+            MemoryConsolidateStagingOperation { force: true }
+        )
+        .await
+        .unwrap()
+        .status,
+        "skipped_empty"
+    );
+    assert!(
+        api.accept_backend_job_result(worker, &submission)
+            .unwrap()
+            .replayed
+    );
+}
+
+#[tokio::test]
+async fn backend_job_profile_resolution_rejects_ungranted_requirements_without_fallback() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(dir.path()).await;
+    let mut request = BackendJobRequest {
+        job_id: "profile-resolution".into(),
+        purpose: "any-purpose".into(),
+        input_revision: "1".into(),
+        input_ref: "test://profiles".into(),
+        input: serde_json::json!({}),
+        instruction: "return structured result".into(),
+        profile: "builtin:backend-job".into(),
+        grants: Default::default(),
+        serialization_key: None,
+        source_worker: None,
+        notification_target: None,
+        limits: Default::default(),
+    };
+    let (selector, bundle) = api.resolve_backend_job_profile(&request).unwrap();
+    assert_eq!(selector, ProfileSelector::Builtin(request.profile.clone()));
+    assert_eq!(
+        crate::profile_settings::resolve_profile_manifest_from_config_bundle(
+            &bundle,
+            &request.profile
+        )
+        .unwrap()
+        .engine
+        .instruction,
+        "internal.backend_job_system"
+    );
+    for profile in [
+        "path:/etc/profile",
+        "builtin:missing",
+        "project:missing",
+        "builtin:coder",
+        "builtin:companion",
+        "builtin:subjektiv-memory-consolidation",
+    ] {
+        request.profile = profile.into();
+        assert!(api.dispatch_backend_job(&request).is_err(), "{profile}");
+        assert!(
+            api.store
+                .get_backend_job(TEST_WORKSPACE_ID, &request.job_id)
+                .unwrap()
+                .is_none()
+        );
+    }
+    assert!(execution.take_inputs().is_empty());
+}
+
+#[tokio::test]
+async fn backend_job_project_recipe_is_resolved_and_delivered_without_system_override() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, _) = test_api_with_recording_backend(dir.path()).await;
+    let state = api
+        .config_store
+        .load_workspace_config(TEST_WORKSPACE_ID)
+        .unwrap()
+        .unwrap();
+    let entrypoint = state.contract.entrypoints.first().unwrap();
+    let root = state.snapshot.entries.get(entrypoint).unwrap();
+    let config = config_commit_request_from_api(server_api::ConfigCommitRequest {
+        base_revision: state.snapshot.revision, base_digest: state.snapshot.digest.clone(),
+        entrypoints: state.contract.entrypoints.iter().map(|p| p.as_str().into()).collect(),
+        changes: vec![server_api::ConfigTreeChange::Update {
+            path: entrypoint.as_str().into(), expected_digest: root.content_digest.clone(),
+            content: "{ profile = { entries = [{ selector = \"project:job-recipe\"; source = \"profiles/job-recipe.dcdl\"; }]; }; }".into(),
+        }, server_api::ConfigTreeChange::Create {
+            path: "profiles/job-recipe.dcdl".into(), content_type: server_api::ConfigContentType::Decodal,
+            content: "{ slug = \"job-recipe\"; scope = \"workspace_read\"; model = { id = \"fixture-model\"; }; engine = { instruction = \"role.default\"; }; feature = { task = { enabled = true; }; }; }".into(),
+        }],
+    }).unwrap();
+    commit_workspace_config_tree(&api, TEST_WORKSPACE_ID, &config).unwrap();
+    let request = BackendJobRequest {
+        job_id: "project-recipe-job".into(),
+        purpose: "no-profile-switch".into(),
+        input_revision: "r1".into(),
+        input_ref: "test://recipe".into(),
+        input: serde_json::json!({}),
+        instruction: "Submit the result".into(),
+        profile: "project:job-recipe".into(),
+        grants: Default::default(),
+        serialization_key: None,
+        source_worker: None,
+        notification_target: None,
+        limits: Default::default(),
+    };
+    let (selector, bundle) = api.resolve_backend_job_profile(&request).unwrap();
+    assert_eq!(
+        selector,
+        ProfileSelector::Named("project:job-recipe".into())
+    );
+    let manifest = crate::profile_settings::resolve_profile_manifest_from_config_bundle(
+        &bundle,
+        &request.profile,
+    )
+    .unwrap();
+    assert_eq!(manifest.engine.instruction, "role.default");
+    assert!(manifest.feature.task.enabled);
+    let dispatched = api.dispatch_backend_job(&request).unwrap();
+    let worker = dispatched.attempt.worker.unwrap();
+    assert_eq!(
+        api.runtime.worker(&worker).unwrap().profile.as_deref(),
+        Some("project:job-recipe")
+    );
+}
+
+#[tokio::test]
+async fn subjektiv_job_legacy_cutover_waits_foreground_then_removes_only_owned_consolidator() {
+    let dir = tempfile::tempdir().unwrap();
+    let (api, execution) = test_api_with_recording_backend(dir.path()).await;
+    let store = open_subjektiv_store(&api).unwrap();
+    let subject = store
+        .create_subject(SubjectRole::new("legacy-cutover").unwrap())
+        .unwrap();
+    let subject_body = spawn_ticket_check_source(&api, "subject-body-protected");
+    let key = format!("subjektiv-consolidation:{}", subject.id);
+    let legacy = api
+        .spawn_workspace_worker(
+            EMBEDDED_WORKER_RUNTIME_ID,
+            WorkerSpawnRequest {
+                requested_worker_name: Some("not-authority-display".into()),
+                singleton_key: Some(key.clone()),
+                intent: WorkerSpawnIntent::WorkspaceOrchestrator,
+                acceptance: WorkerSpawnAcceptanceRequirement::RunAccepted {
+                    expected_segments: 0,
+                },
+                profile: ProfileSelector::Builtin(SUBJEKTIV_MEMORY_CONSOLIDATION_PROFILE.into()),
+                ticket_assignment: None,
+                initial_submit: vec![],
+                workdir_attachment_requests: vec![],
+                resolved_workdir_attachment_requests: vec![],
+                resolved_workdir_attachments: vec![],
+                resolved_config_bundle: None,
+                resolved_workspace_api: None,
+                resolved_memory_settings: None,
+                resolved_subjektiv_attached: false,
+                resolved_worker_observation_enabled: false,
+                resolved_worker_observation_grants: vec![],
+                resolved_control_operation: None,
+            },
+        )
+        .unwrap()
+        .worker
+        .unwrap()
+        .worker;
+    seed_worker_session_for_cleanup(dir.path(), &legacy);
+    execution.publish_worker_state(
+        &legacy,
+        protocol::WorkerStateSnapshot::from(protocol::WorkerStatus::Running),
+    );
+    tokio::time::timeout(std::time::Duration::from_secs(2), async {
+        while api
+            .runtime
+            .worker(&legacy)
+            .unwrap()
+            .worker_state
+            .as_ref()
+            .is_none_or(|s| s.state == protocol::WorkerState::Idle)
+        {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(
+        start_subjektiv_staging_consolidation(
+            api.clone(),
+            &subject.id,
+            MemoryConsolidateStagingOperation { force: true }
+        )
+        .await
+        .unwrap()
+        .status,
+        "skipped_legacy_inflight"
+    );
+    assert!(
+        api.store
+            .find_active_backend_job_for_resource(TEST_WORKSPACE_ID, &key)
+            .unwrap()
+            .is_none()
+    );
+    assert!(api.runtime.worker(&legacy).is_ok());
+    execution.publish_worker_state(&legacy, protocol::WorkerStateSnapshot::initial());
+    tokio::time::timeout(std::time::Duration::from_secs(2), async {
+        while api
+            .runtime
+            .worker(&legacy)
+            .unwrap()
+            .worker_state
+            .as_ref()
+            .is_none_or(|s| s.state != protocol::WorkerState::Idle)
+        {
+            tokio::task::yield_now().await;
+        }
+    })
+    .await
+    .unwrap();
+    assert_eq!(
+        start_subjektiv_staging_consolidation(
+            api.clone(),
+            &subject.id,
+            MemoryConsolidateStagingOperation { force: true }
+        )
+        .await
+        .unwrap()
+        .status,
+        "started"
+    );
+    assert!(api.runtime.worker(&legacy).is_err());
+    assert!(
+        api.store
+            .current_worker_singleton_owner(TEST_WORKSPACE_ID, &key)
+            .unwrap()
+            .is_none()
+    );
+    assert!(api.runtime.worker(&subject_body).is_ok());
+    assert!(store.subject(&subject.id).unwrap().is_some());
+}

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

@@ -527,6 +527,8 @@ async fn browser_protocol_sender_rejects_job_input_and_all_live_methods() {
         input: serde_json::json!({"immutable": true}),
         instruction: "Return a structured result.".into(),
         profile: "builtin:backend-job".into(),
+        grants: Default::default(),
+        serialization_key: None,
         source_worker: None,
         notification_target: None,
         limits: crate::backend_job::BackendJobLimits::default(),

crates/workspace-server/src/subjektiv.rs

@@ -1021,6 +1021,11 @@ ALTER TABLE subjects ADD COLUMN behavior_revision INTEGER NOT NULL DEFAULT 0
     Ok(())
 }
 
+fn add_surface_job_provenance(transaction: &Transaction<'_>) -> crate::feature_storage::Result<()> {
+    transaction.execute_batch("ALTER TABLE surface_generation_runs ADD COLUMN job_id TEXT; ALTER TABLE surface_generation_runs ADD COLUMN attempt_id TEXT;")?;
+    Ok(())
+}
+
 static MIGRATIONS: &[FeatureMigration] = &[
     FeatureMigration::new(1, "create subjektiv subject memory store", create_schema),
     FeatureMigration::new(
@@ -1039,6 +1044,11 @@ static MIGRATIONS: &[FeatureMigration] = &[
         add_surface_generation_state,
     ),
     FeatureMigration::new(5, "add user-managed subject behavior", add_subject_behavior),
+    FeatureMigration::new(
+        6,
+        "bind surface generation provenance to Job attempts",
+        add_surface_job_provenance,
+    ),
 ];
 
 pub const REGISTRATION: FeatureRegistration =
@@ -1958,6 +1968,14 @@ impl SubjektivStore {
     /// active Memory revisions. Kind order is fixed and selection is round-robin
     /// across kinds; within each kind, updated_at desc then memory id asc.
     pub fn prepare_surface_generation(&self, subject_id: &str) -> Result<SurfaceGeneration> {
+        self.prepare_job_surface_generation(subject_id, None)
+    }
+
+    pub(crate) fn prepare_job_surface_generation(
+        &self,
+        subject_id: &str,
+        job_attempt: Option<(&str, &str)>,
+    ) -> Result<SurfaceGeneration> {
         self.database.try_transaction(|transaction| {
             let subject = require_active_subject(transaction, subject_id)?;
             let (active_memory_count, materials) =
@@ -1973,21 +1991,71 @@ impl SubjektivStore {
             transaction.execute(
                 "INSERT INTO surface_generation_runs (
                     subject_id, generation_id, store_revision, active_memory_count,
-                    materials_json, created_at
-                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
+                    materials_json, created_at, job_id, attempt_id
+                 ) VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8)",
                 params![
                     subject_id,
                     generation.id,
                     to_i64(generation.store_revision)?,
                     to_i64(generation.active_memory_count as u64)?,
                     serde_json::to_string(&generation.materials)?,
-                    generation.created_at
+                    generation.created_at,
+                    job_attempt.map(|(job_id, _)| job_id),
+                    job_attempt.map(|(_, attempt_id)| attempt_id),
                 ],
             )?;
             Ok(generation)
         })
     }
 
+    /// Validate a Job's structured surface claim against the existing generation
+    /// and publication/failure records. This does not create another receipt or
+    /// roll back Memory. The generation must have begun inside this attempt.
+    #[allow(clippy::too_many_arguments)]
+    pub(crate) fn validate_job_surface_outcome(
+        &self,
+        subject_id: &str,
+        generation_id: &str,
+        store_revision: u64,
+        availability: &str,
+        snapshot_id: Option<&str>,
+        reason_code: Option<&str>,
+        job_id: &str,
+        attempt_id: &str,
+    ) -> Result<()> {
+        self.database.try_with_connection(|connection| {
+            let subject = require_subject_in_connection(connection, subject_id)?;
+            let generation_revision = require_job_generation(connection, subject_id, generation_id, job_id, attempt_id)?;
+            if generation_revision != store_revision || subject.store_revision != store_revision {
+                return Err(SubjektivError::InvalidRecord("surface generation is stale".into()));
+            }
+            let state = connection.query_row(
+                "SELECT store_revision, status, snapshot_id, reason_code FROM surface_generation_state WHERE subject_id = ?1",
+                [subject_id],
+                |row| Ok((row.get::<_, i64>(0)?, row.get::<_, String>(1)?, row.get::<_, Option<String>>(2)?, row.get::<_, Option<String>>(3)?)),
+            ).optional()?.ok_or_else(|| SubjektivError::InvalidRecord("surface generation has no terminal outcome".into()))?;
+            if state.0 != to_i64(store_revision)? || state.1 != availability
+                || !matches!(availability, "ready" | "failed")
+                || state.2.as_deref() != snapshot_id || state.3.as_deref() != reason_code {
+                return Err(SubjektivError::InvalidRecord("surface claim differs from the persisted outcome".into()));
+            }
+            Ok(())
+        })
+    }
+
+    pub(crate) fn require_job_surface_generation(
+        &self,
+        subject_id: &str,
+        generation_id: &str,
+        job_id: &str,
+        attempt_id: &str,
+    ) -> Result<()> {
+        self.database.try_with_connection(|connection| {
+            require_job_generation(connection, subject_id, generation_id, job_id, attempt_id)?;
+            Ok(())
+        })
+    }
+
     /// Validates structured editor output against the exact material snapshot and
     /// atomically publishes it only while the subject store generation is current.
     pub fn publish_surface_generation(
@@ -3605,6 +3673,26 @@ fn reject_duplicate_memory_ids(kind: &str, values: &[MemoryRevisionRef]) -> Resu
     Ok(())
 }
 
+fn require_job_generation(
+    connection: &Connection,
+    subject_id: &str,
+    generation_id: &str,
+    job_id: &str,
+    attempt_id: &str,
+) -> Result<u64> {
+    let revision: Option<i64> = connection.query_row(
+        "SELECT store_revision FROM surface_generation_runs WHERE subject_id = ?1 AND generation_id = ?2 AND job_id = ?3 AND attempt_id = ?4",
+        params![subject_id, generation_id, job_id, attempt_id], |row| row.get(0),
+    ).optional()?;
+    revision
+        .and_then(|value| u64::try_from(value).ok())
+        .ok_or_else(|| {
+            SubjektivError::InvalidRecord(
+                "surface generation is not owned by this Job attempt".into(),
+            )
+        })
+}
+
 fn to_i64(value: u64) -> Result<i64> {
     i64::try_from(value).map_err(|_| {
         SubjektivError::InvalidRecord(format!("revision value {value} exceeds SQLite range"))

crates/workspace-server/src/ticket_item_checker.rs

@@ -129,6 +129,8 @@ pub fn request(
         })?,
         instruction,
         profile: "builtin:backend-job".to_string(),
+        grants: Default::default(),
+        serialization_key: None,
         source_worker: Some(source_worker.clone()),
         notification_target: Some(source_worker),
         limits: BackendJobLimits {