feat(task-plane): add isolated readonly Codex worker policy

active-runtime/app/task_plane/api.py

@@ -110,6 +110,7 @@ class TaskPlaneService:
             existing = None
         if existing is not None and str(body.get("identity") or "") != existing.identity:
             raise ConflictError("worker identity takeover rejected")
+        self.policy.assert_readonly_worker_registration(actor, body)
         self.policy.assert_allowed(
             "worker.admin" if is_admin else "worker.register", actor
         )
@@ -127,6 +128,7 @@ class TaskPlaneService:
     def worker_heartbeat(self, worker_id: str, payload: dict[str, Any]) -> dict[str, Any]:
         interval = payload.get("heartbeat_interval_seconds")
         actor = str(payload.get("actor_identity") or "")
+        self.policy.assert_readonly_worker_binding(actor, worker_id)
         self.policy.assert_allowed("worker.register", actor)
         return self.workers.heartbeat(
             worker_id,
@@ -162,6 +164,7 @@ class TaskPlaneService:
             raise ValidationError("worker_id required")
 
         task = self.store.get_task(task_id)
+        self.policy.assert_readonly_worker_binding(actor, str(worker_id))
         self.policy.assert_allowed("claim", actor, task=task)
         lease_seconds = _bounded_int(
             body.get("lease_seconds", 30), "lease_seconds", 1, 3600
@@ -181,10 +184,10 @@ class TaskPlaneService:
         if not claim_id:
             raise ValidationError("claim_id required")
         actor = body.get("actor_identity") or body.get("worker_identity") or "worker"
-        self.policy.assert_allowed("heartbeat", actor)
         task = self.store.get_task(task_id)
-        self.policy.assert_allowed("heartbeat", actor, task=task)
         claim = self.store.get_claim(claim_id)
+        self.policy.assert_readonly_worker_binding(actor, claim.worker_id)
+        self.policy.assert_allowed("heartbeat", actor, task=task)
         if claim.worker_identity != actor:
             raise ForbiddenError("claim identity mismatch")
         lease_seconds = _bounded_int(
@@ -198,13 +201,14 @@ class TaskPlaneService:
         if not claim_id:
             raise ValidationError("claim_id required")
         actor = body.get("actor_identity") or body.get("worker_identity") or "worker"
-        self.policy.assert_allowed("progress", actor)
         progress = body.get("progress")
         stage = body.get("stage", "RUNNING")
         message = body.get("message")
         if progress is None:
             raise ValidationError("progress required")
         task = self.store.get_task(task_id)
+        claim = self.store.get_claim(claim_id)
+        self.policy.assert_readonly_worker_binding(actor, claim.worker_id)
         self.policy.assert_allowed("progress", actor, task=task)
         if task.task_plan.data_classification == "credential-secret":
             if message not in {None, ""} or contains_secret_text(stage):
@@ -230,9 +234,7 @@ class TaskPlaneService:
         if not claim_id:
             raise ValidationError("claim_id required")
         actor = body.get("actor_identity") or body.get("worker_identity") or "worker"
-        self.policy.assert_allowed("result", actor)
         task = self.store.get_task(task_id)
-        self.policy.assert_allowed("result", actor, task=task)
 
         status = body.get("status", "COMPLETED")
         outputs = body.get("outputs", [])
@@ -244,6 +246,8 @@ class TaskPlaneService:
         if any(not isinstance(ev, dict) for ev in evidences):
             raise ValidationError("evidence items must be objects")
         claim = self.store.get_claim(claim_id)
+        self.policy.assert_readonly_worker_binding(actor, claim.worker_id)
+        self.policy.assert_allowed("result", actor, task=task)
         if claim.worker_identity != actor:
             raise ForbiddenError("claim identity mismatch")
         if task.task_plan.data_classification == "credential-secret":

active-runtime/app/task_plane/policy.py

@@ -10,6 +10,9 @@ from .models import ProbeEvidence, ResultEnvelope, TaskRecord
 
 
 RISK_RANK = {"low": 1, "medium": 2, "high": 3}
+READONLY_WORKER_ACTIONS = frozenset(
+    {"claim", "result", "fail", "progress", "heartbeat"}
+)
 ALL_ACTION_CAPABILITIES = frozenset(
     {
         "task.creator",
@@ -30,6 +33,26 @@ class PolicyDecision:
     reason: str = ""
 
 
+@dataclass(frozen=True)
+class ReadonlyWorkerProfile:
+    """A narrowly-scoped identity allowed to process safe Shadow tasks only."""
+
+    worker_id: str
+    identity: str
+    capabilities: frozenset[str]
+    task_types: frozenset[str] = frozenset({"code.review", "test.execute"})
+
+    def permits_task(self, task: TaskRecord) -> bool:
+        plan = task.task_plan
+        if plan.task_type not in self.task_types or plan.risk != "low":
+            return False
+        if plan.data_classification not in {"public", "internal"}:
+            return False
+        if plan.metadata.get("production_side_effects") is not False:
+            return False
+        return set(plan.required_capabilities).issubset(self.capabilities)
+
+
 class IdentityVerifier:
     def __init__(self, trusted_identities: set[str] | None = None):
         self.trusted_identities = trusted_identities
@@ -71,6 +94,8 @@ class PolicyEngine:
         probe_attestation_verifier: Callable[[ProbeEvidence], bool] | None = None,
         identity_capabilities: dict[str, set[str]] | None = None,
         admin_identities: set[str] | None = None,
+        readonly_worker_execution_enabled: bool = False,
+        readonly_worker_profiles: list[ReadonlyWorkerProfile] | None = None,
     ) -> None:
         if max_risk not in RISK_RANK:
             raise ValueError("invalid max_risk")
@@ -85,6 +110,40 @@ class PolicyEngine:
             for identity, capabilities in (identity_capabilities or {}).items()
         }
         self.admin_identities = set(admin_identities or ())
+        self.readonly_worker_execution_enabled = readonly_worker_execution_enabled
+        self.readonly_worker_profiles = {
+            profile.identity: profile for profile in (readonly_worker_profiles or [])
+        }
+
+    def _readonly_worker_allows(
+        self, action: str, actor: str, task: TaskRecord | None
+    ) -> bool:
+        profile = self.readonly_worker_profiles.get(actor)
+        return bool(
+            self.readonly_worker_execution_enabled
+            and action in READONLY_WORKER_ACTIONS
+            and profile is not None
+            and (task is None or profile.permits_task(task))
+        )
+
+    def assert_readonly_worker_binding(self, actor: str, worker_id: str) -> None:
+        profile = self.readonly_worker_profiles.get(actor)
+        if profile is not None and profile.worker_id != worker_id:
+            raise ForbiddenError("worker identity binding mismatch")
+
+    def assert_readonly_worker_registration(self, actor: str, payload: dict) -> None:
+        profile = self.readonly_worker_profiles.get(actor)
+        if profile is None:
+            return
+        if (
+            str(payload.get("worker_id") or "") != profile.worker_id
+            or str(payload.get("identity") or "") != profile.identity
+            or set(payload.get("capabilities") or []) != set(profile.capabilities)
+            or str(payload.get("trust_zone") or "") != "isolated-review"
+            or int(payload.get("max_concurrency") or 0) != 1
+            or int(payload.get("heartbeat_interval_seconds") or 0) != 15
+        ):
+            raise ForbiddenError("readonly worker registration mismatch")
 
     def verify_result(
         self, result: ResultEnvelope, task: TaskRecord | None = None
@@ -147,7 +206,8 @@ class PolicyEngine:
                 return PolicyDecision(False, f"action {action} not permitted")
 
         if action in {"claim", "result", "fail", "cancel", "progress", "delegate", "heartbeat"} and not self.execution_enabled:
-            return PolicyDecision(False, "execution disabled")
+            if not self._readonly_worker_allows(action, actor, task):
+                return PolicyDecision(False, "execution disabled")
 
         if action == "delegate" and task and task.status not in {"COMPLETED", "FAILED_FINAL", "CANCELLED"}:
             risk = str(context.get("risk", task.task_plan.risk))

tests/task_plane/public_interface/test_task_plane_api.py

@@ -7,7 +7,12 @@ if os.path.join(ROOT, "active-runtime") not in sys.path:
     sys.path.append(os.path.join(ROOT, "active-runtime"))
 
 from app.task_plane.api import TaskPlaneService
-from app.task_plane.policy import ALL_ACTION_CAPABILITIES, PolicyEngine
+from app.task_plane.policy import (
+    ALL_ACTION_CAPABILITIES,
+    IdentityVerifier,
+    PolicyEngine,
+    ReadonlyWorkerProfile,
+)
 
 
 def _policy(**kwargs):
@@ -81,6 +86,89 @@ class TaskPlaneAPIBehaviorTests(unittest.TestCase):
         self.assertEqual(service.get_children(task["task_id"]), [])
 
 
+class ReadonlyCodexWorkerPolicyTests(unittest.TestCase):
+    identity = "spiffe://aios/agent/codex/review-test-01"
+    worker_id = "codex-review-test-01"
+    capabilities = frozenset(
+        {
+            "code.read",
+            "git.diff.read",
+            "code.review",
+            "review.security",
+            "review.api-contract",
+            "test.python",
+            "test.shell.allowlisted",
+            "task.progress",
+            "task.result",
+            "task.evidence.submit",
+        }
+    )
+
+    def setUp(self):
+        profile = ReadonlyWorkerProfile(self.worker_id, self.identity, self.capabilities)
+        policy = PolicyEngine(
+            identity_verifier=IdentityVerifier({"anon", self.identity}),
+            execution_enabled=False,
+            readonly_worker_execution_enabled=True,
+            readonly_worker_profiles=[profile],
+            identity_capabilities={
+                "anon": {"task.creator"},
+                self.identity: {"worker.self_register", "task.claim"},
+            },
+        )
+        self.service = TaskPlaneService(TaskPlaneStore(":memory:"), policy=policy)
+        self.registration = {
+            "worker_id": self.worker_id,
+            "worker_type": "codex",
+            "identity": self.identity,
+            "capabilities": sorted(self.capabilities),
+            "trust_zone": "isolated-review",
+            "max_concurrency": 1,
+            "heartbeat_interval_seconds": 15,
+            "supported_protocols": ["aios-task-v1", "a2a"],
+            "actor_identity": self.identity,
+        }
+        self.service.register_worker(self.registration)
+
+    def _task(self, *, task_type="code.review", capabilities=None, risk="low"):
+        return self.service.create_task(
+            {
+                "task_type": task_type,
+                "required_capabilities": capabilities or ["code.read", "code.review"],
+                "risk": risk,
+                "metadata": {"production_side_effects": False},
+            }
+        )
+
+    def test_readonly_profile_can_claim_only_safe_task(self):
+        task = self._task()
+        claim = self.service.claim(
+            task["task_id"],
+            {"worker_id": self.worker_id, "lease_seconds": 120, "actor_identity": self.identity},
+        )
+        self.assertTrue(claim["claim_id"])
+
+    def test_readonly_profile_rejects_wrong_worker_or_write_task(self):
+        task = self._task()
+        with self.assertRaises(Exception):
+            self.service.claim(
+                task["task_id"],
+                {"worker_id": "other-worker", "lease_seconds": 120, "actor_identity": self.identity},
+            )
+        write_task = self._task(task_type="code.change", capabilities=["code.write"])
+        with self.assertRaises(Exception):
+            self.service.claim(
+                write_task["task_id"],
+                {"worker_id": self.worker_id, "lease_seconds": 120, "actor_identity": self.identity},
+            )
+
+    def test_readonly_registration_cannot_escalate_capabilities(self):
+        payload = dict(self.registration)
+        payload["capabilities"] = [*payload["capabilities"], "git.commit"]
+        with self.assertRaises(Exception):
+            self.service.register_worker(payload)
+
+
 class DelegationGuardTests(unittest.TestCase):
     def setUp(self):
         self.store = TaskPlaneStore(":memory:")