fix: authorize bound standalone Subject body and Job operations

crates/standalone/src/subjektiv.rs

@@ -803,6 +803,41 @@ impl LocalSubjectHost {
     }
 }
 impl SubjektivHost for LocalSubjectHost {
+    fn authorize_operation(
+        &self,
+        context: &SubjektivHostContext,
+        operation: &str,
+    ) -> Result<(), SubjektivHostError> {
+        // An explicit operator connection issues this local body capability.
+        // Revalidate its live writer on every access/supply, not Feature enablement.
+        // Job decisions and Backend conversations are not local body capabilities.
+        self.body(context)?;
+        match operation {
+            "session.record"
+            | "memory.resident.read"
+            | "memory.resident.supply"
+            | "memory.surface.supply"
+            | "memory.query"
+            | "memory.read"
+            | "memory.changes.list"
+            | "memory.proposal.validate"
+            | "memory.receipt.read"
+            | "memory.staging.write"
+            | "memory.extract.read"
+            | "memory.extract.supply"
+            | "memory.extract.commit"
+            | "memory.consolidation.request"
+            | "journal.startup.read"
+            | "journal.startup.supply"
+            | "journal.compaction.read"
+            | "journal.compaction.supply"
+            | "journal.recording.request" => Ok(()),
+            _ => Err(SubjektivHostError::PermissionDenied(format!(
+                "operation `{operation}` is not granted by the local Subject body capability"
+            ))),
+        }
+    }
+
     fn memory_journal(
         &self,
         context: &SubjektivHostContext,
@@ -1176,6 +1211,37 @@ impl LocalConsolidationHost {
     }
 }
 impl SubjektivHost for LocalConsolidationHost {
+    fn authorize_operation(
+        &self,
+        context: &SubjektivHostContext,
+        operation: &str,
+    ) -> Result<(), SubjektivHostError> {
+        self.live()?;
+        let attempt = format!("job:{}", self.attempt_id);
+        if context.worker_id != attempt || context.session_id != attempt {
+            return Err(SubjektivHostError::PermissionDenied(
+                "context is not bound to this local consolidation attempt".into(),
+            ));
+        }
+        // The Host-issued batch/attempt grant is independent of body permission.
+        // Domain operations still enforce candidate membership and saved receipts.
+        match operation {
+            "memory.query"
+            | "memory.read"
+            | "memory.changes.list"
+            | "memory.staging.list"
+            | "memory.staging.read"
+            | "memory.candidate.decide"
+            | "memory.surface.prepare"
+            | "memory.surface.supply"
+            | "memory.surface.publish"
+            | "memory.surface.fail" => Ok(()),
+            _ => Err(SubjektivHostError::PermissionDenied(format!(
+                "operation `{operation}` is not granted by the local consolidation attempt"
+            ))),
+        }
+    }
+
     fn settings(&self) -> SubjektivHostSettings {
         self.host.settings()
     }

crates/standalone/src/subjektiv/tests.rs

@@ -86,11 +86,56 @@ fn closed_capabilities_and_foreign_worker_context_are_rejected() {
         session_id: session_store::SessionId::now_v7().to_string(),
     };
     assert!(port.memory(&owner, read.clone()).is_ok());
+    let local_operations = [
+        "memory.resident.read",
+        "memory.resident.supply",
+        "memory.staging.write",
+        "journal.startup.read",
+        "journal.startup.supply",
+        "journal.compaction.read",
+        "journal.compaction.supply",
+        "journal.recording.request",
+    ];
+    for operation in local_operations {
+        assert!(
+            port.authorize_operation(&owner, operation).is_ok(),
+            "{operation}"
+        );
+        assert!(
+            port.authorize_operation(&foreign, operation).is_err(),
+            "{operation}"
+        );
+    }
+    for operation in [
+        "chat.read",
+        "chat.send",
+        "worker.messaging",
+        "memory.candidate.decide",
+        "memory.surface.publish",
+        "memory.staging.list",
+        "journal",
+        "journal.startup",
+        "journal.startup.read.future",
+    ] {
+        assert!(
+            matches!(
+                port.authorize_operation(&owner, operation),
+                Err(SubjektivHostError::PermissionDenied(_))
+            ),
+            "local body must not grant {operation}"
+        );
+    }
     assert!(
         port.record_session(&owner, false).is_err(),
         "unattributed Session is not read authority"
     );
     connection.close().unwrap();
+    for operation in local_operations {
+        assert!(
+            port.authorize_operation(&owner, operation).is_err(),
+            "released capability: {operation}"
+        );
+    }
     assert!(port.memory(&owner, read).is_err());
 }
 
@@ -134,3 +179,91 @@ fn different_local_roots_do_not_authorize_shared_subject_ids() {
             .is_err()
     );
 }
+
+#[test]
+fn consolidation_operation_policy_is_attempt_bound_and_does_not_inherit_body_rights() {
+    struct Fence(AtomicBool);
+    impl crate::jobs::JobAttemptFence for Fence {
+        fn ensure_live(&self) -> Result<(), String> {
+            self.0
+                .load(Ordering::Acquire)
+                .then_some(())
+                .ok_or("attempt revoked".into())
+        }
+    }
+    let temp = tempfile::tempdir().unwrap();
+    let catalog = StandaloneSubjects::open(temp.path().join("state")).unwrap();
+    let subject = catalog.create("owner", None).unwrap();
+    let mut body = catalog
+        .connect(&subject.id, protocol::WorkerId::now_v7())
+        .unwrap();
+    let fence = Arc::new(Fence(AtomicBool::new(true)));
+    let job = LocalConsolidationHost {
+        host: body.host.clone(),
+        grant: LocalConsolidationGrant {
+            binding: body.binding.clone(),
+            candidate_ids: vec![],
+        },
+        job_id: "job-1".into(),
+        attempt_id: "attempt-1".into(),
+        fence: fence.clone(),
+    };
+    let context = SubjektivHostContext {
+        worker_id: "job:attempt-1".into(),
+        session_id: "job:attempt-1".into(),
+    };
+    assert!(
+        job.authorize_operation(&context, "memory.staging.list")
+            .is_ok()
+    );
+    assert!(
+        job.authorize_operation(&context, "memory.surface.supply")
+            .is_ok()
+    );
+    let foreign = SubjektivHostContext {
+        worker_id: "job:old-attempt".into(),
+        session_id: "job:old-attempt".into(),
+    };
+    assert!(
+        job.authorize_operation(&foreign, "memory.staging.list")
+            .is_err()
+    );
+    for op in [
+        "session.record",
+        "memory.resident.read",
+        "memory.resident.supply",
+        "memory.extract.read",
+        "memory.staging.write",
+        "journal.startup.read",
+        "chat.send",
+        "memory",
+    ] {
+        assert!(
+            matches!(
+                job.authorize_operation(&context, op),
+                Err(SubjektivHostError::PermissionDenied(_))
+            ),
+            "Job must not gain {op}"
+        );
+    }
+    assert!(
+        job.memory(
+            &context,
+            SubjektivMemoryBackendOperation::ReadCandidate(SubjektivMemoryCandidateReadRequest {
+                candidate_id: "outside-batch".into()
+            })
+        )
+        .is_err()
+    );
+    fence.0.store(false, Ordering::Release);
+    assert!(
+        job.authorize_operation(&context, "memory.staging.list")
+            .is_err()
+    );
+    fence.0.store(true, Ordering::Release);
+    body.close().unwrap();
+    assert!(
+        job.authorize_operation(&context, "memory.staging.list")
+            .is_err()
+    );
+}