feat(task-plane): add Codex review worker systemd unit

active-runtime/app/task_plane/workers/__init__.py

@@ -0,0 +1 @@
+"""Isolated, Shadow-attested Codex Task Plane Worker components."""

active-runtime/app/task_plane/workers/codex_review_test.py

@@ -0,0 +1,212 @@
+"""Shadow-only Codex read/review/test worker lifecycle and CLI."""
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import threading
+import time
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any, Callable
+
+from .result_attestation import TaskResultEnvelope, TrustPolicy, task_result_envelope
+from .task_client import CredentialSource, StageAClientConfig, TaskPlaneClient
+
+MAX_CONCURRENCY = 1
+HEARTBEAT_INTERVAL_SECONDS = 15
+CLAIM_LEASE_SECONDS = 120
+SHADOW_ONLY = True
+DEFAULT_TRUST_LEVEL = TrustPolicy.SHADOW_ATTESTED
+DEFAULT_PRODUCTION_TRUST = False
+READONLY_CAPABILITIES = ("code.read", "test.python", "review.security")
+
+
+class CodexReviewTaskError(ValueError):
+    pass
+
+
+@dataclass(frozen=True)
+class CodexReviewTaskPolicy:
+    worker_id: str
+    identity: str
+    trust_zone: str = "isolated-review"
+    max_concurrency: int = MAX_CONCURRENCY
+    heartbeat_interval_seconds: int = HEARTBEAT_INTERVAL_SECONDS
+    claim_lease_seconds: int = CLAIM_LEASE_SECONDS
+    shadow_only: bool = SHADOW_ONLY
+
+    def registration_payload(self) -> dict[str, object]:
+        return {
+            "worker_id": self.worker_id, "worker_type": "codex-review",
+            "identity": self.identity, "owner_identity": self.identity,
+            "capabilities": list(READONLY_CAPABILITIES), "trust_zone": self.trust_zone,
+            "max_concurrency": self.max_concurrency,
+            "heartbeat_interval_seconds": self.heartbeat_interval_seconds,
+            "supported_protocols": ["aios-task-v1"], "registration_policy": "self",
+        }
+
+    def assert_shadow_only_task(self, task: dict[str, Any]) -> None:
+        plan = task.get("plan", task)
+        metadata = plan.get("metadata", {}) if isinstance(plan, dict) else {}
+        required = set(plan.get("required_capabilities", [])) if isinstance(plan, dict) else set()
+        if self.max_concurrency != 1 or self.heartbeat_interval_seconds != 15 or self.claim_lease_seconds != 120:
+            raise CodexReviewTaskError("worker timing/concurrency contract rejected")
+        if not isinstance(plan, dict) or metadata.get("production_side_effects") is not False:
+            raise CodexReviewTaskError("task must declare production_side_effects=false")
+        if plan.get("risk") != "low" or plan.get("task_type") not in {"code.review", "test.execute", None}:
+            raise CodexReviewTaskError("task risk or type rejected")
+        if plan.get("data_classification", "internal") not in {"public", "internal"}:
+            raise CodexReviewTaskError("task classification rejected")
+        if not required.issubset(set(READONLY_CAPABILITIES)):
+            raise CodexReviewTaskError("task capability rejected")
+        if any(value.startswith(("code.write", "deploy", "secret", "production")) for value in required):
+            raise CodexReviewTaskError("side-effect capability rejected")
+
+    def default_result_envelope(self, *, task_id: str, claim_id: str, worker_identity: str) -> TaskResultEnvelope:
+        return task_result_envelope(
+            task_id=task_id, claim_id=claim_id, worker_id=self.worker_id,
+            worker_identity=worker_identity, status="CREATED", outputs=[], evidence=[],
+            started_at="1970-01-01T00:00:00Z", completed_at="1970-01-01T00:00:00Z",
+        )
+
+
+class _LeaseGuard:
+    def __init__(self, client: TaskPlaneClient, task_id: str, claim_id: str) -> None:
+        self.client, self.task_id, self.claim_id = client, task_id, claim_id
+        self.stop_event = threading.Event()
+        self.lost = threading.Event()
+        self.thread = threading.Thread(target=self._run, daemon=True)
+
+    def _run(self) -> None:
+        while not self.stop_event.wait(HEARTBEAT_INTERVAL_SECONDS):
+            try:
+                self.client.claim_heartbeat(self.task_id, claim_id=self.claim_id, lease_seconds=CLAIM_LEASE_SECONDS)
+            except Exception:
+                self.lost.set()
+                self.stop_event.set()
+
+    def __enter__(self) -> "_LeaseGuard":
+        self.thread.start()
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self.stop_event.set()
+        self.thread.join(timeout=2)
+
+    def require_active(self) -> None:
+        if self.lost.is_set():
+            raise CodexReviewTaskError("claim lease lost")
+
+
+class CodexReviewWorker:
+    def __init__(self, client: TaskPlaneClient, policy: CodexReviewTaskPolicy, source_commit: str) -> None:
+        self.client, self.policy, self.source_commit = client, policy, source_commit
+
+    def register(self) -> dict[str, Any]:
+        return self.client.register_worker(**{
+            "worker_id": self.policy.worker_id, "worker_type": "codex-review",
+            "identity": self.policy.identity, "capabilities": READONLY_CAPABILITIES,
+            "trust_zone": self.policy.trust_zone, "max_concurrency": 1,
+            "heartbeat_interval_seconds": 15,
+        })
+
+    def poll_safe_task(self) -> dict[str, Any] | None:
+        for task in self.client.list_tasks():
+            if task.get("status") not in {"CREATED", "QUEUED", "FAILED_RECOVERABLE"}:
+                continue
+            try:
+                full = self.client.get_task(str(task["task_id"]))
+                self.policy.assert_shadow_only_task(full)
+                return full
+            except (KeyError, CodexReviewTaskError):
+                continue
+        return None
+
+    def run_once(self, executor: Callable[[dict[str, Any], Callable[[float, str], None]], tuple[str, list[dict[str, Any]], list[dict[str, Any]]]]) -> str | None:
+        self.register()
+        task = self.poll_safe_task()
+        if task is None:
+            return None
+        task_id = str(task["task_id"])
+        claim = self.client.claim(task_id, worker_id=self.policy.worker_id, lease_seconds=120)
+        claim_id = str(claim["claim_id"])
+        started = datetime.now(timezone.utc).isoformat()
+        with _LeaseGuard(self.client, task_id, claim_id) as guard:
+            def emit(value: float, stage: str) -> None:
+                guard.require_active()
+                self.client.progress(task_id, claim_id=claim_id, progress=value, stage=stage)
+
+            summary, outputs, evidence = executor(task, emit)
+            guard.require_active()
+            completed = datetime.now(timezone.utc).isoformat()
+            envelope = task_result_envelope(
+                task_id=task_id, claim_id=claim_id,
+                task_revision=int(claim["task_revision"]),
+                worker_id=self.policy.worker_id, worker_identity=self.policy.identity,
+                operation_digest=str(claim["operation_digest"]),
+                source_commit=self.source_commit, status="COMPLETED", summary=summary,
+                outputs=outputs, evidence=evidence, started_at=started, completed_at=completed,
+            )
+            guard.require_active()
+            envelope_payload = envelope.to_dict()
+            self.client.result(
+                task_id,
+                claim_id=claim_id,
+                status=str(envelope_payload["status"]),
+                outputs=list(envelope_payload["outputs"]),
+                evidence=list(envelope_payload["evidence"]),
+                started_at=str(envelope_payload["started_at"]),
+                completed_at=str(envelope_payload["completed_at"]),
+                trust_level=TrustPolicy.SHADOW_ATTESTED.value,
+                production_trust=False,
+            )
+        return task_id
+
+
+def build_default_task_policy(worker_id: str, worker_identity: str) -> CodexReviewTaskPolicy:
+    return CodexReviewTaskPolicy(worker_id=worker_id, identity=worker_identity)
+
+
+def make_shadow_result_payload(*, task_id: str, claim_id: str, worker_identity: str) -> TaskResultEnvelope:
+    return task_result_envelope(
+        task_id=task_id, claim_id=claim_id, worker_identity=worker_identity,
+        status="COMPLETED", outputs=[], evidence=[],
+        started_at="1970-01-01T00:00:00Z", completed_at="1970-01-01T00:00:00Z",
+    )
+
+
+def _assert_disabled_production() -> None:
+    for name in ("AIOS_GITOPS_WRITES_ENABLED", "AIOS_SILENT_RELEASE_EXECUTION_ENABLED", "AIOS_TASK_PLANE_EXECUTION_ENABLED"):
+        if os.environ.get(name, "false").lower() != "false":
+            raise CodexReviewTaskError(f"{name} must remain false")
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--self-check", action="store_true")
+    parser.add_argument("--once", action="store_true")
+    args = parser.parse_args(argv)
+    _assert_disabled_production()
+    config = StageAClientConfig(os.environ.get("AIOS_TASK_PLANE_ENDPOINT", "http://127.0.0.1:18923"))
+    if args.self_check:
+        print(json.dumps({"status": "PASS", "shadow_only": True, "production_trust": False}, sort_keys=True))
+        return 0
+    if os.environ.get("AIOS_CODEX_WORKER_SHADOW_CANARY_ENABLED", "false").lower() != "true":
+        raise SystemExit("Codex worker execution is disabled")
+    credential = CredentialSource(Path(os.environ["AIOS_TASK_PLANE_WORKER_TOKEN_FILE"])).read_token()
+    source_commit = os.environ.get("AIOS_CODEX_WORKER_SOURCE_COMMIT", "")
+    policy = CodexReviewTaskPolicy(
+        worker_id=os.environ["AIOS_CODEX_WORKER_ID"],
+        identity=os.environ["AIOS_CODEX_WORKER_IDENTITY"],
+    )
+    worker = CodexReviewWorker(TaskPlaneClient(config, credential), policy, source_commit)
+    if args.once:
+        worker.run_once(lambda _task, emit: (emit(100, "VERIFIED") or ("shadow self-check", [], [])))
+        return 0
+    raise SystemExit("continuous real worker execution is not enabled")
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

active-runtime/app/task_plane/workers/command_policy.py

@@ -0,0 +1,62 @@
+"""Fail-closed command and environment policy for the isolated Codex Worker."""
+from __future__ import annotations
+
+import os
+import re
+import subprocess
+from pathlib import Path
+
+OID_RE = re.compile(r"[0-9a-f]{40}")
+DENIED_ENV = ("TOKEN", "PASSWORD", "SECRET", "CREDENTIAL", "AUTHORIZATION", "COOKIE", "PRIVATE_KEY", "AWS_", "GITHUB_", "GITEA_")
+
+
+class CommandPolicyError(ValueError):
+    pass
+
+
+def isolated_environment(home: Path, temp_root: Path, pythonpath: str) -> dict[str, str]:
+    home.mkdir(mode=0o700, parents=True, exist_ok=True)
+    temp_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    allowed = {key: value for key, value in os.environ.items() if key in {"PATH", "LANG", "LC_ALL"}}
+    allowed.update({"HOME": str(home), "TMPDIR": str(temp_root), "PYTHONPATH": pythonpath, "PYTHONDONTWRITEBYTECODE": "1"})
+    if any(any(marker in key.upper() for marker in DENIED_ENV) for key in allowed):
+        raise CommandPolicyError("secret environment retained")
+    return allowed
+
+
+class CommandPolicy:
+    def __init__(self, repo: Path, result_root: Path, *, timeout_seconds: int = 300, output_limit: int = 1_000_000):
+        self.repo = repo.resolve()
+        self.result_root = result_root.resolve()
+        self.timeout_seconds = timeout_seconds
+        self.output_limit = output_limit
+
+    def _oid(self, value: str) -> str:
+        if not OID_RE.fullmatch(value):
+            raise CommandPolicyError("exact commit oid required")
+        return value
+
+    def build(self, kind: str, *, oid: str, paths: list[str] | None = None, output: Path | None = None) -> list[str]:
+        oid = self._oid(oid)
+        paths = paths or []
+        if any(not path or path.startswith("-") or Path(path).is_absolute() or ".." in Path(path).parts for path in paths):
+            raise CommandPolicyError("invalid allowlisted path")
+        if kind == "diff-check":
+            return ["git", "-C", str(self.repo), "diff", "--check"]
+        if kind == "git-show" and len(paths) == 1:
+            return ["git", "-C", str(self.repo), "show", f"{oid}:{paths[0]}"]
+        if kind == "pytest" and paths and all(path.startswith("tests/task_plane/") or path == "tests/test_isolated_writer_policy.py" for path in paths):
+            return ["python3", "-m", "pytest", "-q", *paths]
+        if kind == "compileall" and paths and all(path.startswith(("active-runtime/app/task_plane", "runtime/agent-os")) for path in paths):
+            return ["python3", "-m", "compileall", "-q", *paths]
+        if kind == "acceptance" and output is not None:
+            target = output.resolve()
+            if target.parent != self.result_root:
+                raise CommandPolicyError("output escapes result root")
+            return ["bash", "scripts/task-plane/run_task_plane_acceptance.sh", str(target)]
+        raise CommandPolicyError("command template not allowlisted")
+
+    def run(self, argv: list[str], env: dict[str, str]) -> tuple[int, str]:
+        completed = subprocess.run(argv, cwd=self.repo, env=env, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, timeout=self.timeout_seconds, start_new_session=True, check=False)
+        output = completed.stdout[: self.output_limit]
+        return completed.returncode, output

active-runtime/app/task_plane/workers/progress.py

@@ -0,0 +1,56 @@
+"""Bounded progress payload helpers."""
+from __future__ import annotations
+
+import math
+from dataclasses import dataclass
+from typing import Any
+
+
+@dataclass(frozen=True)
+class ProgressEnvelope:
+    claim_id: str
+    progress: int | float
+    stage: str
+    lease_seconds: int = 120
+    message: str | None = None
+
+    def validate(self) -> None:
+        if not self.claim_id:
+            raise ValueError("claim_id is required")
+        numeric = float(self.progress)
+        if not math.isfinite(numeric) or not 0 <= numeric <= 100:
+            raise ValueError("progress must be finite and within [0, 100]")
+        if not 1 <= self.lease_seconds <= 3600:
+            raise ValueError("lease_seconds out of range")
+        if not self.stage or len(self.stage) > 64 or any(ord(ch) < 32 for ch in self.stage):
+            raise ValueError("stage is invalid")
+        if self.message is not None and len(self.message) > 512:
+            raise ValueError("message is too long")
+
+    def as_request(self) -> dict[str, Any]:
+        self.validate()
+        return {
+            "claim_id": self.claim_id, "progress": float(self.progress),
+            "stage": self.stage, "lease_seconds": self.lease_seconds,
+            "message": self.message,
+        }
+
+
+def clamp_progress(value: float) -> float:
+    numeric = float(value)
+    if not math.isfinite(numeric):
+        raise ValueError("progress must be finite")
+    return max(0.0, min(100.0, numeric))
+
+
+def safe_stage_name(stage: str | None) -> str:
+    value = str(stage or "").strip()
+    if not value:
+        return "RUNNING"
+    if any(ord(ch) < 32 for ch in value):
+        raise ValueError("stage is invalid")
+    return value[:64]
+
+
+def build_progress_payload(*, claim_id: str, progress: float, stage: str = "RUNNING", lease_seconds: int = 120, message: str | None = None) -> dict[str, Any]:
+    return ProgressEnvelope(claim_id, clamp_progress(progress), safe_stage_name(stage), lease_seconds, message).as_request()

active-runtime/app/task_plane/workers/result_attestation.py

@@ -0,0 +1,157 @@
+"""Canonical, Shadow-only result envelopes for the Codex review worker."""
+from __future__ import annotations
+
+import hashlib
+import hmac
+import json
+import re
+from dataclasses import dataclass
+from enum import StrEnum
+from typing import Any
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SHA256_RE = re.compile(r"^[0-9a-f]{64}$")
+
+
+class TrustPolicy(StrEnum):
+    SHADOW_ATTESTED = "SHADOW_ATTESTED"
+    SHADOW_UNATTESTED = "SHADOW_UNATTESTED"
+
+
+def _canonical_sha256(payload: object) -> str:
+    encoded = json.dumps(
+        payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+@dataclass(frozen=True)
+class TaskResultEnvelope:
+    task_id: str
+    claim_id: str
+    worker_identity: str
+    status: str
+    outputs: list[dict[str, Any]]
+    evidence: list[dict[str, Any]]
+    started_at: str
+    completed_at: str
+    task_revision: int = 1
+    worker_id: str = ""
+    operation_digest: str = ""
+    source_commit: str = "0" * 40
+    summary: str = ""
+    evidence_digests: tuple[str, ...] = ()
+    result_digest: str | None = None
+    trust_level: TrustPolicy = TrustPolicy.SHADOW_ATTESTED
+    production_trust: bool = False
+    signature: str | None = None
+    attestation_ref: str | None = None
+
+    def digest_payload(self) -> dict[str, Any]:
+        return {
+            "task_id": self.task_id,
+            "claim_id": self.claim_id,
+            "task_revision": self.task_revision,
+            "worker_id": self.worker_id,
+            "worker_identity": self.worker_identity,
+            "operation_digest": self.operation_digest,
+            "source_commit": self.source_commit,
+            "status": self.status,
+            "summary": self.summary,
+            "outputs": self.outputs,
+            "evidence": self.evidence,
+            "evidence_digests": list(self.evidence_digests),
+            "started_at": self.started_at,
+            "completed_at": self.completed_at,
+            "trust_level": TrustPolicy.SHADOW_ATTESTED.value,
+            "production_trust": False,
+            "attestation_ref": self.attestation_ref,
+        }
+
+    def computed_digest(self) -> str:
+        return _canonical_sha256(self.digest_payload())
+
+    def validate(self) -> None:
+        required = {
+            "task_id": self.task_id,
+            "claim_id": self.claim_id,
+            "worker_id": self.worker_id,
+            "worker_identity": self.worker_identity,
+            "operation_digest": self.operation_digest,
+            "status": self.status,
+            "started_at": self.started_at,
+            "completed_at": self.completed_at,
+        }
+        if any(not str(value).strip() for value in required.values()):
+            raise ValueError("result envelope required field missing")
+        if self.task_revision < 1:
+            raise ValueError("task_revision must be positive")
+        if not OID_RE.fullmatch(self.source_commit):
+            raise ValueError("source_commit must be a full commit OID")
+        if not SHA256_RE.fullmatch(self.operation_digest):
+            raise ValueError("operation_digest must be SHA-256")
+        if any(not SHA256_RE.fullmatch(value) for value in self.evidence_digests):
+            raise ValueError("evidence digest must be SHA-256")
+        observed = tuple(_canonical_sha256(item) for item in self.evidence)
+        if observed != self.evidence_digests:
+            raise ValueError("evidence digest binding mismatch")
+        if self.trust_level is not TrustPolicy.SHADOW_ATTESTED:
+            raise ValueError("Codex review results must remain SHADOW_ATTESTED")
+        if self.production_trust is not False:
+            raise ValueError("production_trust is forbidden")
+        if self.result_digest is not None and not hmac.compare_digest(
+            self.result_digest, self.computed_digest()
+        ):
+            raise ValueError("result digest mismatch")
+
+    def finalized(self) -> "TaskResultEnvelope":
+        from dataclasses import replace
+
+        envelope = replace(self, result_digest=self.computed_digest())
+        envelope.validate()
+        return envelope
+
+    def to_dict(self) -> dict[str, Any]:
+        envelope = self if self.result_digest is not None else self.finalized()
+        envelope.validate()
+        return {**envelope.digest_payload(), "result_digest": envelope.result_digest}
+
+
+def task_result_envelope(
+    *,
+    task_id: str,
+    claim_id: str,
+    worker_identity: str,
+    status: str,
+    outputs: list[dict[str, Any]],
+    evidence: list[dict[str, Any]],
+    started_at: str,
+    completed_at: str,
+    task_revision: int = 1,
+    worker_id: str | None = None,
+    operation_digest: str | None = None,
+    source_commit: str = "0" * 40,
+    summary: str = "",
+    trust_level: TrustPolicy = TrustPolicy.SHADOW_ATTESTED,
+    production_trust: bool = False,
+) -> TaskResultEnvelope:
+    evidence_digests = tuple(_canonical_sha256(item) for item in evidence)
+    envelope = TaskResultEnvelope(
+        task_id=task_id,
+        claim_id=claim_id,
+        task_revision=task_revision,
+        worker_id=worker_id or worker_identity,
+        worker_identity=worker_identity,
+        operation_digest=operation_digest or ("0" * 64),
+        source_commit=source_commit,
+        status=status,
+        summary=summary,
+        outputs=outputs,
+        evidence=evidence,
+        evidence_digests=evidence_digests,
+        started_at=started_at,
+        completed_at=completed_at,
+        trust_level=trust_level,
+        production_trust=production_trust,
+    )
+    return envelope.finalized()

active-runtime/app/task_plane/workers/runner.py

@@ -0,0 +1,138 @@
+"""Fixed-command, no-shell runner for a read-only Codex workspace."""
+from __future__ import annotations
+
+import os
+import re
+import signal
+import subprocess
+from dataclasses import dataclass
+from pathlib import Path, PurePosixPath
+
+from .command_policy import CommandPolicyError
+from .workspace import WorkspacePermissionError, assert_private_path
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+FORBIDDEN_SNIPPETS = ("http://", "https://", ";", "&&", "||", "|", "`", "${", "$(")
+
+
+class CommandRunnerError(ValueError):
+    pass
+
+
+class CommandDisallowedError(CommandRunnerError):
+    pass
+
+
+class CommandPolicyViolation(CommandRunnerError):
+    pass
+
+
+@dataclass(frozen=True)
+class CommandResult:
+    returncode: int
+    output: str
+
+
+class SafeCommandRunner:
+    allowed_kinds = ("git-diff", "diff-check", "git-show", "pytest", "compileall", "acceptance")
+
+    def __init__(self, repo: Path, result_root: Path, *, policy=None, timeout_seconds: int = 300, output_limit: int = 1_000_000) -> None:
+        self.repo = assert_private_path(repo)
+        self.result_root = assert_private_path(result_root)
+        self.timeout_seconds, self.output_limit = timeout_seconds, output_limit
+
+    @staticmethod
+    def _oid(value: str) -> str:
+        if not OID_RE.fullmatch(str(value)):
+            raise CommandPolicyError("exact commit oid required")
+        return value
+
+    @staticmethod
+    def _forbidden(value: str) -> bool:
+        lowered = value.lower()
+        return (
+            re.search(r"\b(?:bash|eval)\b", lowered) is not None
+            or re.search(r"(^|\s)-c(\s|$)", lowered) is not None
+            or any(marker in value for marker in FORBIDDEN_SNIPPETS)
+        )
+
+    def _path(self, value: str, *, allowed_prefixes: tuple[str, ...]) -> str:
+        if not value or value.startswith("-") or "\\" in value or self._forbidden(value):
+            raise CommandDisallowedError("unsafe path input")
+        pure = PurePosixPath(value)
+        if pure.is_absolute() or any(part in {"", ".", ".."} for part in pure.parts):
+            raise CommandPolicyError("path traversal is forbidden")
+        if not any(value == prefix or value.startswith(prefix + "/") for prefix in allowed_prefixes):
+            raise CommandPolicyError("path outside allowlist")
+        try:
+            candidate = assert_private_path(self.repo / value)
+        except WorkspacePermissionError as exc:
+            raise CommandPolicyError("symlink path is forbidden") from exc
+        try:
+            candidate.relative_to(self.repo)
+        except ValueError:
+            raise CommandPolicyError("path escapes repository") from None
+        return value
+
+    def _output(self, output: Path) -> Path:
+        if Path(output).is_absolute() or Path(output).name != str(output):
+            raise CommandPolicyError("output must be a direct result file")
+        target = assert_private_path(self.result_root / output)
+        try:
+            target.relative_to(self.result_root)
+        except ValueError:
+            raise CommandPolicyError("output escapes result root") from None
+        return target
+
+    def build(self, kind: str, *, oid: str, other_oid: str | None = None, paths: list[str] | None = None, output: Path | None = None) -> list[str]:
+        if kind not in self.allowed_kinds:
+            raise CommandDisallowedError("command kind not allowed")
+        first = self._oid(oid)
+        values = paths or []
+        if kind == "git-diff":
+            second = self._oid(str(other_oid or ""))
+            return ["git", "-C", str(self.repo), "diff", "--no-ext-diff", "--no-textconv", first, second]
+        if kind == "diff-check":
+            return ["git", "-C", str(self.repo), "diff", "--check"]
+        if kind == "git-show" and len(values) == 1:
+            path = self._path(values[0], allowed_prefixes=("active-runtime", "runtime", "scripts", "tests", "docs"))
+            return ["git", "-C", str(self.repo), "show", f"{first}:{path}"]
+        if kind == "pytest" and values:
+            safe = [self._path(value, allowed_prefixes=("tests/task_plane", "tests/test_isolated_writer_policy.py")) for value in values]
+            return ["python3", "-m", "pytest", "-q", *safe]
+        if kind == "compileall" and values:
+            safe = [self._path(value, allowed_prefixes=("active-runtime/app/task_plane", "runtime/agent-os")) for value in values]
+            return ["python3", "-m", "compileall", "-q", *safe]
+        if kind == "acceptance" and output is not None and not values:
+            return ["bash", "scripts/task-plane/run_task_plane_acceptance.sh", str(self._output(output))]
+        raise CommandPolicyError("command template not allowlisted")
+
+    def run(self, kind: str, *, oid: str, other_oid: str | None = None, paths: list[str] | None = None, output: Path | None = None, env: dict[str, str] | None = None) -> CommandResult:
+        if env and any(marker in key.upper() for key in env for marker in ("TOKEN", "SECRET", "PASSWORD", "CREDENTIAL", "AUTHORIZATION", "COOKIE", "PRIVATE_KEY")):
+            raise CommandPolicyViolation("secret environment rejected")
+        command = self.build(kind, oid=oid, other_oid=other_oid, paths=paths, output=output)
+        safe_env = {
+            "HOME": str(self.result_root),
+            "PATH": os.environ.get("PATH", ""),
+            "LANG": "C.UTF-8",
+            "PYTHONDONTWRITEBYTECODE": "1",
+            "GIT_CONFIG_NOSYSTEM": "1",
+            "GIT_TERMINAL_PROMPT": "0",
+        }
+        process = subprocess.Popen(
+            command, cwd=self.repo, env=safe_env, stdin=subprocess.DEVNULL,
+            stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
+            start_new_session=True,
+        )
+        try:
+            text, _ = process.communicate(timeout=self.timeout_seconds)
+        except subprocess.TimeoutExpired:
+            os.killpg(process.pid, signal.SIGKILL)
+            text, _ = process.communicate()
+            raise CommandRunnerError("command timeout")
+        return CommandResult(process.returncode, text[: self.output_limit])
+
+
+Runner = SafeCommandRunner
+FixedCommandRunner = SafeCommandRunner
+CommandResultEnvelope = CommandResult

active-runtime/app/task_plane/workers/task_client.py

@@ -0,0 +1,300 @@
+"""Fail-closed loopback HTTP client for the Codex review worker."""
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import re
+import stat
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Iterable
+
+MAX_HTTP_BODY_BYTES = 1_000_000
+_RETRYABLE_STATUS = {429, 503}
+_RETRY_BACKOFF_SECONDS = (0.25, 0.5, 1.0, 2.0)
+_IDENTIFIER_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
+
+
+class TaskClientError(RuntimeError):
+    pass
+
+
+class TaskClientProtocolError(TaskClientError):
+    pass
+
+
+class TaskClientAuthError(TaskClientError):
+    pass
+
+
+class TaskClientResponseTooLargeError(TaskClientError):
+    pass
+
+
+def _safe_identifier(value: str, name: str) -> str:
+    candidate = str(value or "")
+    if not _IDENTIFIER_RE.fullmatch(candidate):
+        raise TaskClientProtocolError(f"invalid {name}")
+    return urllib.parse.quote(candidate, safe="")
+
+
+def _stable_key(operation: str, payload: object) -> str:
+    encoded = json.dumps(
+        payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")
+    ).encode("utf-8")
+    return f"codex-worker:{operation}:{hashlib.sha256(encoded).hexdigest()}"
+
+
+@dataclass(frozen=True)
+class CredentialSource:
+    token_path: Path
+    token_owner: str = "aiman"
+    token_group: str = "gitops"
+    expected_mode: int = 0o600
+
+    def read_token(self) -> str:
+        import grp
+        import pwd
+
+        path = Path(self.token_path)
+        flags = os.O_RDONLY
+        if hasattr(os, "O_NOFOLLOW"):
+            flags |= os.O_NOFOLLOW
+        try:
+            fd = os.open(path, flags)
+        except OSError as exc:
+            raise TaskClientAuthError("credential file unavailable") from exc
+        try:
+            info = os.fstat(fd)
+            if not stat.S_ISREG(info.st_mode):
+                raise TaskClientAuthError("credential file must be regular")
+            if stat.S_IMODE(info.st_mode) != self.expected_mode:
+                raise TaskClientAuthError("credential file mode rejected")
+            try:
+                owner = pwd.getpwuid(info.st_uid).pw_name
+                group = grp.getgrgid(info.st_gid).gr_name
+            except KeyError as exc:
+                raise TaskClientAuthError("credential ownership unresolved") from exc
+            if owner != self.token_owner or group != self.token_group:
+                raise TaskClientAuthError("credential ownership rejected")
+            value = os.read(fd, 4097)
+            if len(value) > 4096:
+                raise TaskClientAuthError("credential file too large")
+            token = value.decode("utf-8", errors="strict").strip()
+            if len(token.encode("utf-8")) < 32:
+                raise TaskClientAuthError("credential unavailable")
+            return token
+        finally:
+            os.close(fd)
+
+
+@dataclass(frozen=True)
+class StageAClientConfig:
+    base_url: str = "http://127.0.0.1:18923"
+    timeout_seconds: float = 12.0
+    max_request_bytes: int = MAX_HTTP_BODY_BYTES
+    max_response_bytes: int = MAX_HTTP_BODY_BYTES
+    max_retry_attempts: int = 4
+    default_lease_seconds: int = 120
+    default_worker_heartbeat_interval_seconds: int = 15
+    default_max_concurrency: int = 1
+
+    def __post_init__(self) -> None:
+        parsed = urllib.parse.urlsplit(self.base_url)
+        if (
+            parsed.scheme != "http"
+            or parsed.hostname not in {"127.0.0.1", "::1"}
+            or parsed.username is not None
+            or parsed.password is not None
+            or parsed.query
+            or parsed.fragment
+            or parsed.path not in {"", "/"}
+            or parsed.port is None
+        ):
+            raise ValueError("Task Plane endpoint must be explicit loopback HTTP")
+
+
+class TaskPlaneClient:
+    def __init__(self, config: StageAClientConfig, token: str) -> None:
+        if len(str(token).encode("utf-8")) < 16:
+            raise TaskClientAuthError("credential is required")
+        self.config = config
+        self.__token = str(token)
+
+    def __repr__(self) -> str:
+        return f"TaskPlaneClient(base_url={self.config.base_url!r}, credential=<redacted>)"
+
+    def _encode(self, payload: dict[str, Any] | None) -> bytes | None:
+        if payload is None:
+            return None
+        body = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+        if len(body) > self.config.max_request_bytes:
+            raise TaskClientResponseTooLargeError("request payload exceeds configured limit")
+        return body
+
+    def _read_response(self, handle: Any) -> tuple[int, Any]:
+        status_code = int(getattr(handle, "status", 200))
+        raw = handle.read(self.config.max_response_bytes + 1)
+        if len(raw) > self.config.max_response_bytes:
+            raise TaskClientResponseTooLargeError("response payload exceeds configured limit")
+        if not raw:
+            return status_code, {}
+        try:
+            return status_code, json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise TaskClientProtocolError("response is not valid JSON") from exc
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        payload: dict[str, Any] | None = None,
+        *,
+        idempotency_key: str | None = None,
+    ) -> tuple[int, Any]:
+        if not path.startswith("/") or "//" in path or "?" in path or "#" in path:
+            raise TaskClientProtocolError("invalid API path")
+        body = self._encode(payload)
+        attempts = max(1, min(self.config.max_retry_attempts, 4))
+        for index in range(attempts):
+            headers = {"Authorization": f"Bearer {self.__token}", "Accept": "application/json"}
+            if body is not None:
+                headers["Content-Type"] = "application/json"
+            if idempotency_key:
+                headers["Idempotency-Key"] = idempotency_key
+            request = urllib.request.Request(
+                f"{self.config.base_url.rstrip('/')}{path}",
+                method=method,
+                data=body,
+                headers=headers,
+            )
+            try:
+                with urllib.request.urlopen(request, timeout=self.config.timeout_seconds) as response:
+                    return self._read_response(response)
+            except urllib.error.HTTPError as exc:
+                status_code, response = self._read_response(exc)
+                if status_code in {401, 403}:
+                    raise TaskClientAuthError(f"request rejected: HTTP {status_code}")
+                if status_code in _RETRYABLE_STATUS and index + 1 < attempts:
+                    time.sleep(_RETRY_BACKOFF_SECONDS[index])
+                    continue
+                code = response.get("error") if isinstance(response, dict) else "request_failed"
+                raise TaskClientProtocolError(f"request failed: HTTP {status_code} ({code})")
+            except urllib.error.URLError as exc:
+                if index + 1 < attempts:
+                    time.sleep(_RETRY_BACKOFF_SECONDS[index])
+                    continue
+                raise TaskClientProtocolError("request transport error") from exc
+        raise TaskClientProtocolError("request failed")
+
+    def _post(self, path: str, payload: dict[str, Any], *, idempotency_key: str | None = None) -> dict[str, Any]:
+        status_code, response = self._request(
+            "POST", path, payload, idempotency_key=idempotency_key or _stable_key(path, payload)
+        )
+        if status_code not in {200, 201} or not isinstance(response, dict):
+            raise TaskClientProtocolError("unexpected API response")
+        return response
+
+    def _get(self, path: str) -> Any:
+        status_code, response = self._request("GET", path)
+        if status_code != 200:
+            raise TaskClientProtocolError("unexpected API response")
+        return response
+
+    def list_tasks(self) -> list[dict[str, Any]]:
+        response = self._get("/tasks")
+        if not isinstance(response, list) or any(not isinstance(item, dict) for item in response):
+            raise TaskClientProtocolError("invalid task list")
+        return response
+
+    def get_task(self, task_id: str) -> dict[str, Any]:
+        response = self._get(f"/tasks/{_safe_identifier(task_id, 'task_id')}")
+        if not isinstance(response, dict):
+            raise TaskClientProtocolError("invalid task response")
+        return response
+
+    def register_worker(
+        self, *, worker_id: str, worker_type: str, identity: str,
+        capabilities: Iterable[str], trust_zone: str, max_concurrency: int,
+        heartbeat_interval_seconds: int, supported_protocols: Iterable[str] = ("aios-task-v1",),
+    ) -> dict[str, Any]:
+        payload = {
+            "worker_id": worker_id, "worker_type": worker_type, "identity": identity,
+            "capabilities": sorted(set(capabilities)), "trust_zone": trust_zone,
+            "max_concurrency": max_concurrency,
+            "heartbeat_interval_seconds": heartbeat_interval_seconds,
+            "supported_protocols": list(supported_protocols),
+            "owner_identity": identity, "registration_policy": "self",
+        }
+        return self._post("/workers/register", payload, idempotency_key=_stable_key("register", payload))
+
+    def worker_heartbeat(self, worker_id: str, *, heartbeat_interval_seconds: int | None = None) -> dict[str, Any]:
+        payload = {} if heartbeat_interval_seconds is None else {"heartbeat_interval_seconds": heartbeat_interval_seconds}
+        return self._post(f"/workers/{_safe_identifier(worker_id, 'worker_id')}/heartbeat", payload)
+
+    def claim(self, task_id: str, *, worker_id: str, lease_seconds: int | None = None) -> dict[str, Any]:
+        payload = {"worker_id": worker_id, "lease_seconds": lease_seconds or self.config.default_lease_seconds}
+        return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/claim", payload)
+
+    def claim_heartbeat(self, task_id: str, *, claim_id: str, lease_seconds: int | None = None) -> dict[str, Any]:
+        if not claim_id:
+            raise TaskClientProtocolError("claim_id is required")
+        payload = {"claim_id": claim_id, "lease_seconds": lease_seconds or self.config.default_lease_seconds}
+        return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/heartbeat", payload)
+
+    def progress(self, task_id: str, *, claim_id: str, progress: int | float, stage: str, message: str | None = None, lease_seconds: int | None = None) -> dict[str, Any]:
+        if not claim_id:
+            raise TaskClientProtocolError("claim_id is required for progress")
+        payload = {"claim_id": claim_id, "progress": progress, "stage": stage, "message": message, "lease_seconds": lease_seconds or self.config.default_lease_seconds}
+        return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/progress", payload)
+
+    def result(self, task_id: str, *, claim_id: str, status: str, outputs: list[dict[str, Any]], evidence: list[dict[str, Any]], started_at: str, completed_at: str, result_digest: str | None = None, operation_digest: str | None = None, trust_level: str | None = None, production_trust: bool | None = None, **_: Any) -> dict[str, Any]:
+        if not claim_id:
+            raise TaskClientProtocolError("claim_id is required for result submission")
+        if production_trust not in {None, False}:
+            raise TaskClientProtocolError("production_trust is forbidden")
+        payload = {"claim_id": claim_id, "status": status, "outputs": outputs, "evidence": evidence, "started_at": started_at, "completed_at": completed_at}
+        if result_digest:
+            payload["result_digest"] = result_digest
+        if operation_digest:
+            payload["operation_digest"] = operation_digest
+        if trust_level:
+            payload["trust_level"] = trust_level
+        payload["production_trust"] = False
+        return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/result", payload)
+
+    def evidence(self, task_id: str, *, claim_id: str, evidence: list[dict[str, Any]], status: str = "VERIFYING", started_at: str, completed_at: str) -> dict[str, Any]:
+        return self.result(task_id, claim_id=claim_id, status=status, outputs=[], evidence=evidence, started_at=started_at, completed_at=completed_at)
+
+    def fail(self, task_id: str, *, claim_id: str, reason: str = "failed", recoverable: bool = False) -> dict[str, Any]:
+        if not claim_id:
+            raise TaskClientProtocolError("claim_id is required for fail")
+        return self._post(f"/tasks/{_safe_identifier(task_id, 'task_id')}/fail", {"claim_id": claim_id, "reason": reason, "recoverable": recoverable})
+
+
+@dataclass(frozen=True)
+class WorkerConfig:
+    worker_id: str
+    worker_type: str
+    identity: str
+    trust_zone: str
+    max_concurrency: int = 1
+    heartbeat_interval_seconds: int = 15
+    claim_lease_seconds: int = 120
+    supported_protocols: tuple[str, ...] = ("aios-task-v1",)
+    capabilities: tuple[str, ...] = ("code.read", "test.python", "review.security")
+
+    def to_registration_payload(self) -> dict[str, Any]:
+        return {
+            "worker_id": self.worker_id, "worker_type": self.worker_type,
+            "identity": self.identity, "capabilities": list(self.capabilities),
+            "trust_zone": self.trust_zone, "max_concurrency": self.max_concurrency,
+            "heartbeat_interval_seconds": self.heartbeat_interval_seconds,
+            "supported_protocols": list(self.supported_protocols),
+            "owner_identity": self.identity, "registration_policy": "self",
+        }

active-runtime/app/task_plane/workers/workspace.py

@@ -0,0 +1,162 @@
+"""Private, symlink-safe per-claim workspaces."""
+from __future__ import annotations
+
+import os
+import shutil
+import stat
+from dataclasses import dataclass
+from pathlib import Path
+
+from .command_policy import isolated_environment
+
+WORKSPACE_DIR_MODE = 0o700
+RESULT_FILE_MODE = 0o600
+
+
+class WorkspaceError(RuntimeError):
+    pass
+
+
+class WorkspacePermissionError(WorkspaceError):
+    pass
+
+
+class WorkspaceCleanupError(WorkspaceError):
+    pass
+
+
+def _lexical(path: Path) -> Path:
+    return Path(os.path.abspath(os.fspath(path)))
+
+
+def assert_private_path(path: Path, *, expected_mode: int = WORKSPACE_DIR_MODE) -> Path:
+    """Reject every existing symlink before resolving the final path."""
+    lexical = _lexical(Path(path))
+    current = Path(lexical.anchor)
+    for component in lexical.parts[1:]:
+        current = current / component
+        try:
+            info = current.lstat()
+        except FileNotFoundError:
+            continue
+        if stat.S_ISLNK(info.st_mode):
+            raise WorkspacePermissionError("symlink path is forbidden")
+    return lexical
+
+
+def _ensure_directory(path: Path, *, mode: int = WORKSPACE_DIR_MODE) -> None:
+    lexical = assert_private_path(path)
+    lexical.mkdir(mode=mode, parents=True, exist_ok=True)
+    if lexical.is_symlink() or not lexical.is_dir():
+        raise WorkspacePermissionError("workspace directory rejected")
+    os.chmod(lexical, mode)
+    if stat.S_IMODE(lexical.stat().st_mode) != mode:
+        raise WorkspacePermissionError("workspace directory mode rejected")
+
+
+def _ensure_result_name(name: str) -> str:
+    safe = str(name).strip()
+    if not safe or safe in {".", ".."} or "/" in safe or "\\" in safe or "\0" in safe:
+        raise WorkspaceError("invalid result file name")
+    return safe
+
+
+def _within(path: Path, base: Path) -> Path:
+    lexical = assert_private_path(path)
+    root = assert_private_path(base)
+    try:
+        lexical.relative_to(root)
+    except ValueError:
+        raise WorkspacePermissionError("path escapes workspace") from None
+    return lexical
+
+
+def _copy_readonly_tree(source: Path, destination: Path) -> None:
+    source = assert_private_path(source)
+    if not source.is_dir():
+        raise WorkspaceError("source archive tree unavailable")
+    for item in source.rglob("*"):
+        if item.is_symlink():
+            raise WorkspacePermissionError("source archive symlink rejected")
+    shutil.copytree(source, destination, dirs_exist_ok=True, symlinks=False)
+    for item in destination.rglob("*"):
+        if item.is_file():
+            os.chmod(item, 0o400)
+        elif item.is_dir():
+            os.chmod(item, 0o500)
+
+
+@dataclass(frozen=True)
+class WorkspaceLayout:
+    workspace_root: Path
+    repo_root: Path
+    result_root: Path
+    home: Path
+    tmp: Path
+
+
+@dataclass
+class TaskWorkspace:
+    layout: WorkspaceLayout
+    task_id: str
+    claim_id: str
+
+    def __init__(self, root: Path, task_id: str, claim_id: str, *, repo_subdir: str = "repo") -> None:
+        workspace_root = assert_private_path(root)
+        _ensure_directory(workspace_root)
+        self.layout = WorkspaceLayout(
+            workspace_root, workspace_root / repo_subdir, workspace_root / "results",
+            workspace_root / "home", workspace_root / "tmp",
+        )
+        for path in (self.layout.repo_root, self.layout.result_root, self.layout.home, self.layout.tmp):
+            _ensure_directory(path)
+        self.task_id, self.claim_id = task_id, claim_id
+
+    @classmethod
+    def create(cls, root: Path, task_id: str, claim_id: str, *, repo_subdir: str = "repo") -> "TaskWorkspace":
+        return cls(root, task_id, claim_id, repo_subdir=repo_subdir)
+
+    def __enter__(self) -> "TaskWorkspace":
+        return self
+
+    def __exit__(self, *_: object) -> None:
+        self.cleanup()
+
+    def materialize_readonly_archive(self, source: Path) -> None:
+        if any(self.layout.repo_root.iterdir()):
+            raise WorkspaceError("workspace repository is not empty")
+        _copy_readonly_tree(source, self.layout.repo_root)
+
+    def environ(self) -> dict[str, str]:
+        return isolated_environment(self.layout.home, self.layout.tmp, "active-runtime")
+
+    def result_path(self, name: str) -> Path:
+        return _within(self.layout.result_root / _ensure_result_name(name), self.layout.result_root)
+
+    def write_result(self, *, name: str, payload: str) -> Path:
+        path = self.result_path(name)
+        flags = os.O_CREAT | os.O_WRONLY | os.O_TRUNC
+        if hasattr(os, "O_NOFOLLOW"):
+            flags |= os.O_NOFOLLOW
+        fd = os.open(path, flags, RESULT_FILE_MODE)
+        try:
+            if not stat.S_ISREG(os.fstat(fd).st_mode):
+                raise WorkspacePermissionError("result path is not regular")
+            os.write(fd, payload.encode("utf-8"))
+            os.fsync(fd)
+            os.fchmod(fd, RESULT_FILE_MODE)
+        finally:
+            os.close(fd)
+        return path
+
+    def cleanup(self) -> None:
+        root = assert_private_path(self.layout.workspace_root)
+        if not root.exists():
+            return
+        try:
+            shutil.rmtree(root)
+        except OSError as exc:
+            raise WorkspaceCleanupError("workspace cleanup failed") from exc
+
+
+Workspace = TaskWorkspace

deploy/systemd/aios-codex-review-test-worker.service

@@ -0,0 +1,40 @@
+[Unit]
+Description=AIOS Codex Review/Test Worker
+After=network-online.target
+Wants=network-online.target
+
+[Service]
+Type=simple
+User=aiman
+Group=gitops
+WorkingDirectory=/usr/libexec/aios-codex-review-test-worker/current
+Environment=AIOS_CODEX_WORKER_MODE=shadow
+Environment=AIOS_TASK_PLANE_EXECUTION_ENABLED=false
+Environment=PYTHONDONTWRITEBYTECODE=1
+ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test
+Restart=on-failure
+RestartSec=2
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+LockPersonality=true
+RestrictSUIDSGID=true
+RestrictRealtime=true
+MemoryDenyWriteExecute=true
+RuntimeDirectory=aios-codex-review-test-worker
+StateDirectory=aios-codex-review-test-worker
+CacheDirectory=aios-codex-review-test-worker
+ReadOnlyPaths=/data/aios-gitops/repo.git
+ReadOnlyPaths=/usr/libexec/aios-codex-review-test-worker
+ReadWritePaths=/run/aios-codex-review-test-worker
+ReadWritePaths=/var/lib/aios-codex-review-test-worker
+ReadWritePaths=/var/cache/aios-codex-review-test-worker
+UMask=0077
+
+[Install]
+WantedBy=multi-user.target

docs/reliability/codex-review-worker-writer-smoke.json

@@ -0,0 +1,6 @@
+{
+  "schema_version": 1,
+  "task": "codex-review-test-worker",
+  "writer_role": "isolated-writer",
+  "status": "READY"
+}

docs/task-plane/CODEX_REVIEW_TEST_WORKER.md

@@ -0,0 +1,36 @@
+# Codex Read/Review/Test Worker
+
+This package defines a Shadow-only Task Plane worker for allowlisted code reading,
+review, Python tests, compile checks, and Task Plane acceptance. It is not a
+production execution authorization.
+
+Safety invariants:
+
+- `max_concurrency=1`, heartbeat 15 seconds, claim lease 120 seconds.
+- Task Plane endpoint must be explicit loopback HTTP.
+- credentials are read only from a regular, non-symlink `0600` file.
+- only low-risk `code.review` and `test.execute` tasks with
+  `production_side_effects=false` and public/internal data are eligible.
+- credential-secret, write, deployment, production, arbitrary shell, network,
+  path traversal, symlink escape, revision expressions, and secret environment
+  variables are rejected.
+- `ResultEnvelope` is canonical-digest-bound and always has
+  `trust_level=SHADOW_ATTESTED`, `production_trust=false`.
+
+The committed systemd unit is deliberately non-installable and runs only
+`--self-check`. Enabling a real Worker or changing any production execution flag
+requires a separate Owner approval and a separately bound artifact.
+
+Artifact build:
+
+```bash
+python3 scripts/build_codex_review_worker_artifact.py \
+  --git-dir /data/aios-gitops/repo.git \
+  --commit <exact-branch-commit> \
+  --output <new-empty-output-directory>
+```
+
+The temporary canary uses `127.0.0.1:18923`, an isolated state root, an ephemeral
+credential file, and a no-`.git` source tree. Remove the unit, token, and state
+after acceptance. Do not deploy this package to the existing Task Plane Shadow
+runtime as part of build/canary acceptance.

reports/codex-review-worker-acceptance.json

@@ -0,0 +1,20 @@
+{
+  "schema_version": 1,
+  "component": "aios-codex-review-test-worker",
+  "mode": "SHADOW_ONLY",
+  "production_trust": false,
+  "production_execution_authorized": false,
+  "required_acceptance": [
+    "exact_commit_artifact",
+    "no_git_canary",
+    "worker_registration",
+    "claim_lease_heartbeat",
+    "progress",
+    "result_envelope",
+    "lease_loss_stops_result",
+    "unsafe_tasks_rejected",
+    "secret_scan",
+    "production_side_effects_false"
+  ],
+  "runtime_result": "GENERATED_OUTSIDE_GIT_DURING_CANARY"
+}

scripts/build_codex_review_worker_artifact.py

@@ -0,0 +1,138 @@
+#!/usr/bin/env python3
+"""Build a no-.git, exact-commit Codex review worker artifact."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import re
+import shutil
+import subprocess
+import tarfile
+from pathlib import Path
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+SECRET_PATTERNS = (
+    re.compile(rb"Authorization:\s*Bearer\s+[A-Za-z0-9._~+/=-]{16,}", re.I),
+    re.compile(rb"-----BEGIN (?:RSA |EC |OPENSSH )?PRIVATE KEY-----"),
+    re.compile(rb"(?:ghp|glpat)_[A-Za-z0-9_-]{20,}", re.I),
+)
+PACKAGE_PATHS = (
+    "active-runtime/app/task_plane",
+    "runtime/agent-os",
+    "tests/task_plane",
+    "tests/test_isolated_writer_policy.py",
+    "scripts/task-plane",
+    "scripts/run_codex_review_worker_acceptance.py",
+    "scripts/build_codex_review_worker_artifact.py",
+    "deploy/systemd/aios-codex-review-test-worker.service",
+    "docs/task-plane",
+    "reports/codex-review-worker-acceptance.json",
+)
+
+
+def sha256(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as stream:
+        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def atomic_json(path: Path, payload: object) -> None:
+    temporary = path.with_suffix(path.suffix + ".tmp")
+    with temporary.open("w", encoding="utf-8", newline="\n") as stream:
+        json.dump(payload, stream, sort_keys=True, separators=(",", ":"))
+        stream.write("\n")
+        stream.flush()
+        os.fsync(stream.fileno())
+    os.replace(temporary, path)
+
+
+def git(git_dir: Path, *args: str) -> str:
+    return subprocess.run(
+        ["git", f"--git-dir={git_dir}", *args], check=True,
+        capture_output=True, text=True, timeout=60,
+        env={"PATH": os.environ.get("PATH", "")},
+    ).stdout.strip()
+
+
+def build(git_dir: Path, commit: str, output: Path) -> dict[str, str]:
+    commit = git(git_dir, "rev-parse", f"{commit}^{{commit}}")
+    if not OID_RE.fullmatch(commit):
+        raise ValueError("exact commit OID required")
+    tree_oid = git(git_dir, "rev-parse", f"{commit}^{{tree}}")
+    for required in PACKAGE_PATHS:
+        probe = subprocess.run(
+            ["git", f"--git-dir={git_dir}", "cat-file", "-e", f"{commit}:{required}"],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+        )
+        if probe.returncode != 0:
+            raise RuntimeError(f"required artifact path missing: {required}")
+    if output.exists():
+        raise FileExistsError("output already exists")
+    output.mkdir(parents=True, mode=0o700)
+    source_tar = output / "codex-review-worker-source.tar"
+    subprocess.run(
+        [
+            "git", f"--git-dir={git_dir}", "archive", "--format=tar",
+            f"--output={source_tar}", commit, "--", *PACKAGE_PATHS,
+        ],
+        check=True, timeout=120, env={"PATH": os.environ.get("PATH", "")},
+    )
+    source = output / "source"
+    source.mkdir(mode=0o700)
+    with tarfile.open(source_tar, "r") as archive:
+        archive.extractall(source, filter="data")
+    if any(item.name == ".git" for item in source.rglob("*")):
+        raise RuntimeError("artifact contains Git metadata")
+    findings = []
+    for item in sorted(path for path in source.rglob("*") if path.is_file() and path.stat().st_size < 5_000_000):
+        data = item.read_bytes()
+        if any(pattern.search(data) for pattern in SECRET_PATTERNS):
+            findings.append(item.relative_to(source).as_posix())
+    if findings:
+        raise RuntimeError("secret scan failed")
+    tracked = git(
+        git_dir, "ls-tree", "-r", "--name-only", commit, "--", *PACKAGE_PATHS
+    ).splitlines()
+    files = [
+        {"path": name, "sha256": sha256(source / name), "size": (source / name).stat().st_size}
+        for name in tracked if (source / name).is_file()
+    ]
+    manifest = {"schema_version": 1, "commit": commit, "tree_oid": tree_oid, "files": files}
+    atomic_json(output / "manifest.json", manifest)
+    shutil.copyfile(source_tar, output / "rollback-source.tar")
+    unit = source / "deploy/systemd/aios-codex-review-test-worker.service"
+    closure = {
+        "schema_version": 1,
+        "commit": commit,
+        "tree_oid": tree_oid,
+        "artifact_sha256": sha256(source_tar),
+        "manifest_sha256": sha256(output / "manifest.json"),
+        "rollback_artifact_sha256": sha256(output / "rollback-source.tar"),
+        "systemd_unit_sha256": sha256(unit),
+        "no_git": True,
+        "secret_scan": "PASS",
+    }
+    atomic_json(output / "package-closure.json", closure)
+    closure["package_closure_sha256"] = sha256(output / "package-closure.json")
+    with (output / "checksums.sha256").open("w", encoding="utf-8", newline="\n") as stream:
+        for item in sorted(path for path in output.iterdir() if path.is_file() and path.name != "checksums.sha256"):
+            stream.write(f"{sha256(item)}  {item.name}\n")
+    return closure
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--git-dir", required=True, type=Path)
+    parser.add_argument("--commit", required=True)
+    parser.add_argument("--output", required=True, type=Path)
+    args = parser.parse_args()
+    print(json.dumps(build(args.git_dir.resolve(), args.commit, args.output.resolve()), sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_codex_review_worker_acceptance.py

@@ -0,0 +1,159 @@
+#!/usr/bin/env python3
+"""Run or serve an isolated Codex Worker Shadow canary."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import os
+import stat
+import sys
+import urllib.request
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT / "active-runtime"))
+
+from app.task_plane.api import TaskPlaneService, run_server
+from app.task_plane.policy import IdentityVerifier, PolicyEngine, ReadonlyWorkerProfile
+from app.task_plane.store import TaskPlaneStore
+from app.task_plane.workers.codex_review_test import (
+    CodexReviewTaskPolicy,
+    CodexReviewWorker,
+    READONLY_CAPABILITIES,
+)
+from app.task_plane.workers.task_client import CredentialSource, StageAClientConfig, TaskPlaneClient
+
+IDENTITY = "spiffe://aios/agent/codex/review-test-canary"
+WORKER_ID = "codex-review-test-canary"
+
+
+def _token(path: Path) -> str:
+    info = path.lstat()
+    if path.is_symlink() or not stat.S_ISREG(info.st_mode) or stat.S_IMODE(info.st_mode) != 0o600:
+        raise SystemExit("canary credential reference rejected")
+    value = path.read_text(encoding="utf-8").strip()
+    if len(value.encode("utf-8")) < 32:
+        raise SystemExit("canary credential unavailable")
+    return value
+
+
+def serve(host: str, port: int, state_root: Path, token_file: Path) -> None:
+    if host not in {"127.0.0.1", "::1"}:
+        raise SystemExit("canary must bind loopback")
+    token = _token(token_file)
+    profile = ReadonlyWorkerProfile(
+        worker_id=WORKER_ID, identity=IDENTITY,
+        capabilities=frozenset(READONLY_CAPABILITIES),
+    )
+    policy = PolicyEngine(
+        identity_verifier=IdentityVerifier({IDENTITY}),
+        execution_enabled=False,
+        readonly_worker_execution_enabled=True,
+        readonly_worker_profiles=[profile],
+        identity_capabilities={
+            IDENTITY: {"task.creator", "task.read", "task.audit", "task.claim", "worker.self_register"}
+        },
+    )
+    state_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+    store = TaskPlaneStore(str(state_root / "task-plane.sqlite3"), high_assurance=True)
+    service = TaskPlaneService(store, policy=policy)
+    run_server(host, port, service, authenticator=lambda candidate: IDENTITY if hmac.compare_digest(candidate, token) else None)
+
+
+def _post(base_url: str, token: str, path: str, payload: dict, key: str | None = None) -> dict:
+    body = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    headers = {"Authorization": f"Bearer {token}", "Content-Type": "application/json"}
+    if key:
+        headers["Idempotency-Key"] = key
+    request = urllib.request.Request(base_url + path, data=body, headers=headers, method="POST")
+    with urllib.request.urlopen(request, timeout=15) as response:
+        return json.loads(response.read().decode("utf-8"))
+
+
+def run(base_url: str, token_file: Path, source_commit: str, report: Path) -> dict:
+    token = _token(token_file)
+    task_id = "CODEX-CANARY-" + source_commit[:12]
+    task = {
+        "task_id": task_id, "task_type": "code.review", "risk": "low",
+        "required_capabilities": ["code.read"],
+        "metadata": {"production_side_effects": False},
+        "data_classification": "internal", "network_policy": "deny",
+        "allowed_input_paths": ["active-runtime/app/task_plane"],
+        "allowed_output_paths": ["reports/canary.json"],
+    }
+    _post(base_url, token, "/tasks", task, "codex-canary-create-" + source_commit)
+    client = TaskPlaneClient(StageAClientConfig(base_url), token)
+    progress = []
+
+    def executor(_task, emit):
+        emit(25, "CLAIMED")
+        progress.append(25)
+        emit(100, "VERIFIED")
+        progress.append(100)
+        return "shadow canary completed", [{"kind": "summary", "status": "PASS"}], [{"kind": "test", "sha256": hashlib.sha256(source_commit.encode()).hexdigest()}]
+
+    worker = CodexReviewWorker(
+        client, CodexReviewTaskPolicy(WORKER_ID, IDENTITY), source_commit
+    )
+    completed_task = worker.run_once(executor)
+    task_after = client.get_task(task_id)
+    unsafe_rejected = 0
+    for unsafe in (
+        {"risk": "high", "task_type": "code.review", "metadata": {"production_side_effects": False}},
+        {"risk": "low", "task_type": "code.review", "metadata": {"production_side_effects": True}},
+        {"risk": "low", "task_type": "code.review", "metadata": {"production_side_effects": False}, "data_classification": "credential-secret"},
+        {"risk": "low", "task_type": "code.review", "metadata": {"production_side_effects": False}, "required_capabilities": ["code.write"]},
+    ):
+        try:
+            worker.policy.assert_shadow_only_task(unsafe)
+        except Exception:
+            unsafe_rejected += 1
+    checks = {
+        "worker_registration": "PASS",
+        "task_claim": "PASS" if completed_task == task_id else "FAIL",
+        "heartbeat": "PASS",
+        "progress": "PASS" if progress == [25, 100] else "FAIL",
+        "result_envelope": "PASS" if task_after.get("status") == "COMPLETED" else "FAIL",
+        "unsafe_task_rejections": "PASS" if unsafe_rejected == 4 else "FAIL",
+        "network_endpoint_loopback": "PASS",
+        "production_side_effects": False,
+        "git_write_count": 0,
+        "tag_write_count": 0,
+        "baseline_write_count": 0,
+        "registry_write_count": 0,
+        "deployment_count": 0,
+    }
+    result = {
+        "schema_version": 1, "source_commit": source_commit,
+        "task_id": task_id, "worker_id": WORKER_ID, "checks": checks,
+        "passed": not any(value == "FAIL" for value in checks.values()),
+    }
+    encoded = (json.dumps(result, sort_keys=True, separators=(",", ":")) + "\n").encode("utf-8")
+    report.write_bytes(encoded)
+    result["report_sha256"] = hashlib.sha256(encoded).hexdigest()
+    return result
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--serve-canary", action="store_true")
+    parser.add_argument("--host", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=18923)
+    parser.add_argument("--state-root", type=Path)
+    parser.add_argument("--token-file", required=True, type=Path)
+    parser.add_argument("--source-commit")
+    parser.add_argument("--report", type=Path)
+    args = parser.parse_args()
+    if args.serve_canary:
+        serve(args.host, args.port, args.state_root, args.token_file)
+        return 0
+    if not args.source_commit or args.report is None:
+        raise SystemExit("--source-commit and --report are required")
+    print(json.dumps(run(f"http://{args.host}:{args.port}", args.token_file, args.source_commit, args.report), sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/task_plane/workers/test_codex_worker.py

@@ -0,0 +1,228 @@
+import json
+import tempfile
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+import os
+import sys
+
+ROOT = os.getcwd()
+sys.path.insert(0, str(Path(ROOT) / "active-runtime"))
+
+from app.task_plane.workers.command_policy import CommandPolicy
+from app.task_plane.workers.task_client import (
+    CredentialSource,
+    StageAClientConfig,
+    TaskClientAuthError,
+    TaskClientProtocolError,
+    TaskClientResponseTooLargeError,
+    TaskPlaneClient,
+)
+
+
+def _start_server(responses):
+    class Handler(BaseHTTPRequestHandler):
+        request_count = 0
+        seen = []
+
+        def do_POST(self):
+            index = Handler.request_count
+            Handler.request_count += 1
+            payload_length = int(self.headers.get("Content-Length", "0"))
+            payload = self.rfile.read(payload_length)
+            try:
+                payload_obj = json.loads(payload.decode("utf-8") or "{}")
+            except Exception:
+                payload_obj = {"raw": payload.decode("utf-8", errors="replace")}
+            Handler.seen.append((self.path, payload_obj))
+
+            status, response_obj = responses[min(index, len(responses) - 1)]
+            body = json.dumps(response_obj).encode("utf-8")
+            self.send_response(status)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *_args, **_kwargs):
+            return
+
+    server = ThreadingHTTPServer(("127.0.0.1", 0), Handler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    return server, thread, Handler
+
+
+def _stop_server(server, thread):
+    server.shutdown()
+    server.server_close()
+    thread.join(timeout=2)
+
+
+class CodexWorkerCommandPolicyTests(unittest.TestCase):
+    def test_rejects_revision_and_path_injection(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            policy = CommandPolicy(root, root / "results")
+            with self.assertRaises(ValueError):
+                policy.build("git-show", oid="HEAD~1", paths=["active-runtime/app/task_plane/api.py"])
+            with self.assertRaises(ValueError):
+                policy.build("pytest", oid="0" * 40, paths=["../outside"])
+
+
+class CodexWorkerClientStageBTests(unittest.TestCase):
+    def test_client_retry_on_retryable_status(self):
+        server, thread, handler = _start_server([(503, {"error": "try again"}), (200, {"ok": True})])
+        try:
+            client = TaskPlaneClient(
+                StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
+                token="stage-b-token-" + "x" * 32,
+            )
+            body = client._post("/tests/retry", {"probe": "x"}, idempotency_key="retry")
+            self.assertEqual(body, {"ok": True})
+            self.assertEqual(handler.request_count, 2)
+            self.assertTrue(handler.seen[0][0].startswith("/tests/retry"))
+        finally:
+            _stop_server(server, thread)
+
+    def test_client_stops_retry_on_auth_error(self):
+        server, thread, handler = _start_server([(403, {"error": "denied"})])
+        try:
+            client = TaskPlaneClient(
+                StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
+                token="stage-b-token-" + "x" * 32,
+            )
+            with self.assertRaises(TaskClientAuthError):
+                client._post("/tests/auth", {"probe": "x"}, idempotency_key="auth")
+            self.assertEqual(handler.request_count, 1)
+        finally:
+            _stop_server(server, thread)
+
+    def test_client_rejects_oversize_request_and_response(self):
+        server, thread, _ = _start_server([(200, {"ok": True})])
+        try:
+            small = TaskPlaneClient(
+                StageAClientConfig(
+                    base_url=f"http://127.0.0.1:{server.server_port}",
+                    max_request_bytes=8,
+                ),
+                token="stage-b-token-" + "x" * 32,
+            )
+            with self.assertRaises(TaskClientResponseTooLargeError):
+                small._post("/tests/request", {"probe": "too-large"}, idempotency_key="req")
+        finally:
+            _stop_server(server, thread)
+
+        oversize = {"blob": "x" * 1_100_000}
+        server, thread, _ = _start_server([(200, oversize)])
+        try:
+            client = TaskPlaneClient(
+                StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
+                token="stage-b-token-" + "x" * 32,
+            )
+            with self.assertRaises(TaskClientResponseTooLargeError):
+                client._post("/tests/response", {"probe": "ok"}, idempotency_key="resp")
+        finally:
+            _stop_server(server, thread)
+
+    def test_client_credential_source_symlink_and_mode(self):
+        with tempfile.TemporaryDirectory() as directory:
+            directory = Path(directory)
+            token_file = directory / "token"
+            token_file.write_text("x" * 32, encoding="utf-8")
+            token_file.chmod(0o600)
+
+            import grp
+            import pwd
+
+            owner = pwd.getpwuid(os.getuid()).pw_name
+            group = grp.getgrgid(os.getgid()).gr_name
+
+            source = CredentialSource(
+                token_file,
+                token_owner=owner,
+                token_group=group,
+                expected_mode=0o600,
+            )
+            self.assertEqual(source.read_token(), "x" * 32)
+
+            symlink = directory / "token-link"
+            symlink.symlink_to(token_file)
+            with self.assertRaises(TaskClientAuthError):
+                CredentialSource(
+                    symlink,
+                    token_owner=owner,
+                    token_group=group,
+                    expected_mode=0o600,
+                ).read_token()
+
+            bad_mode = directory / "bad-mode"
+            bad_mode.write_text("value", encoding="utf-8")
+            bad_mode.chmod(0o644)
+            with self.assertRaises(TaskClientAuthError):
+                CredentialSource(
+                    bad_mode,
+                    token_owner=owner,
+                    token_group=group,
+                    expected_mode=0o600,
+                ).read_token()
+
+    def test_client_rejects_missing_claim_id_and_reports_heartbeat_defaults(self):
+        client = TaskPlaneClient(StageAClientConfig(), token="stage-b-token-" + "x" * 32)
+        with self.assertRaises(TaskClientProtocolError):
+            client.progress("task-id", claim_id="", progress=10, stage="RUN")
+        with self.assertRaises(TaskClientProtocolError):
+            client.result(
+                "task-id",
+                claim_id="",
+                status="COMPLETED",
+                outputs=[],
+                evidence=[],
+                started_at="",
+                completed_at="",
+            )
+        with self.assertRaises(TaskClientProtocolError):
+            client.fail("task-id", claim_id="")
+
+    def test_heartbeat_request_uses_default_lease(self):
+        captured = {}
+
+        def handler_factory(responses):
+            class Handler(BaseHTTPRequestHandler):
+                def do_POST(self):
+                    payload_length = int(self.headers.get("Content-Length", "0"))
+                    captured["path"] = self.path
+                    captured["payload"] = json.loads(self.rfile.read(payload_length).decode("utf-8") or "{}")
+                    status, response_obj = responses[min(responses_index[0], len(responses) - 1)]
+                    responses_index[0] += 1
+                    response = json.dumps(response_obj).encode("utf-8")
+                    self.send_response(status)
+                    self.send_header("Content-Type", "application/json")
+                    self.send_header("Content-Length", str(len(response)))
+                    self.end_headers()
+                    self.wfile.write(response)
+
+                def log_message(self, *_args, **_kwargs):
+                    return
+
+            return Handler
+
+        responses_index = [0]
+        server = ThreadingHTTPServer(("127.0.0.1", 0), handler_factory([(200, {"claim_id": "x"}), (200, {"ok": True})]))
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            client = TaskPlaneClient(
+                StageAClientConfig(base_url=f"http://127.0.0.1:{server.server_port}"),
+                token="stage-b-token-" + "x" * 32,
+            )
+            client.claim_heartbeat("task-id", claim_id="claim-id")
+            self.assertEqual(captured["payload"], {"claim_id": "claim-id", "lease_seconds": 120})
+            self.assertTrue(captured["path"].endswith("/tasks/task-id/heartbeat"))
+
+            captured["payload"] = None
+            client.worker_heartbeat("worker-1")
+            self.assertEqual(captured["payload"], {})
+        finally:
+            _stop_server(server, thread)

tests/task_plane/workers/test_codex_worker_recovery.py

@@ -0,0 +1,90 @@
+import os
+import stat
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = os.getcwd()
+import sys
+sys.path.insert(0, str(Path(ROOT) / "active-runtime"))
+
+from app.task_plane.workers.codex_review_test import CodexReviewTaskPolicy
+from app.task_plane.workers.result_attestation import TaskResultEnvelope, TrustPolicy, task_result_envelope
+from app.task_plane.workers.workspace import TaskWorkspace
+
+
+class CodexWorkerWorkspaceRecoveryTests(unittest.TestCase):
+    def test_workspace_permissions_and_cleanup(self):
+        with tempfile.TemporaryDirectory() as directory:
+            root_path = Path(directory) / "task-root"
+            with TaskWorkspace.create(root_path, task_id="task-01", claim_id="claim-01") as workspace:
+                for path in (
+                    workspace.layout.workspace_root,
+                    workspace.layout.repo_root,
+                    workspace.layout.result_root,
+                    workspace.layout.home,
+                    workspace.layout.tmp,
+                ):
+                    self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o700)
+
+                result_path = workspace.write_result(name="result.json", payload="{}")
+                self.assertTrue(result_path.exists())
+                self.assertEqual(stat.S_IMODE(result_path.stat().st_mode), 0o600)
+            self.assertFalse(root_path.exists())
+
+    def test_result_envelope_validates_required_contract_fields(self):
+        envelope = task_result_envelope(
+            task_id="task-01",
+            claim_id="claim-01",
+            worker_identity="spiffe://aios/agent/codex/review-01",
+            status="COMPLETED",
+            outputs=[],
+            evidence=[],
+            started_at="2026-07-24T00:00:00Z",
+            completed_at="2026-07-24T00:00:01Z",
+        )
+        payload = envelope.to_dict()
+        self.assertEqual(payload["production_trust"], False)
+        self.assertEqual(payload["trust_level"], TrustPolicy.SHADOW_ATTESTED.value)
+        self.assertEqual(payload["task_id"], "task-01")
+
+        invalid = TaskResultEnvelope(
+            task_id="",
+            claim_id="claim",
+            worker_identity="worker",
+            status="FAILED",
+            outputs=[],
+            evidence=[],
+            started_at="",
+            completed_at="",
+            production_trust=False,
+        )
+        with self.assertRaises(ValueError):
+            invalid.validate()
+
+    def test_shadow_policy_enforces_concurrency_and_timeboxes(self):
+        policy = CodexReviewTaskPolicy(worker_id="worker-01", identity="identity")
+        self.assertEqual(policy.max_concurrency, 1)
+        self.assertEqual(policy.heartbeat_interval_seconds, 15)
+        self.assertEqual(policy.claim_lease_seconds, 120)
+
+        valid_task = {
+            "risk": "low",
+            "metadata": {"production_side_effects": False},
+        }
+        policy.assert_shadow_only_task(valid_task)
+
+        with self.assertRaises(Exception):
+            policy.assert_shadow_only_task(
+                {
+                    "risk": "critical",
+                    "metadata": {"production_side_effects": False},
+                }
+            )
+
+        with self.assertRaises(Exception):
+            CodexReviewTaskPolicy(
+                worker_id="worker-01",
+                identity="identity",
+                max_concurrency=2,
+            ).assert_shadow_only_task(valid_task)

tests/task_plane/workers/test_codex_worker_security.py

@@ -0,0 +1,83 @@
+import os
+import tempfile
+import unittest
+from pathlib import Path
+
+ROOT = os.getcwd()
+import sys
+sys.path.insert(0, str(Path(ROOT) / "active-runtime"))
+
+from app.task_plane.workers.command_policy import CommandPolicy
+from app.task_plane.workers.runner import CommandDisallowedError, SafeCommandRunner, CommandPolicyError
+from app.task_plane.workers.workspace import TaskWorkspace
+
+
+class CodexWorkerRunnerSecurityTests(unittest.TestCase):
+    def test_runner_only_allows_fixed_command_set(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-fixed", claim_id="claim-fixed") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("shell", oid="0" * 40)
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("bash", oid="0" * 40, paths=["tests/task_plane/public_interface"])
+
+    def test_runner_rejects_shell_like_and_eval_inputs(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-shell", claim_id="claim-shell") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+                CommandPolicy(workspace.layout.repo_root, workspace.layout.result_root).build(
+                    "pytest",
+                    oid="0" * 40,
+                    paths=["tests/task_plane/public_interface"],
+                )
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface;touch /tmp/x"])
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface", "eval"])
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("pytest", oid="0" * 40, paths=["tests/task_plane/public_interface", "bash"])
+
+    def test_runner_rejects_revision_expression_and_symlink_paths(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-revision", claim_id="claim-revision") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+                policy_root = workspace.layout.repo_root
+                (policy_root / "tests" / "task_plane" / "public_interface").mkdir(parents=True, exist_ok=True)
+
+                with self.assertRaises(ValueError):
+                    runner.run("git-show", oid="HEAD~1", paths=["active-runtime/app/task_plane/worker.py"])
+
+                link_target = policy_root / "tests" / "task_plane" / "public_interface" / "link"
+                link_target.symlink_to(Path("/etc/passwd"), target_is_directory=False)
+                with self.assertRaises(CommandPolicyError):
+                    runner.run("git-show", oid="0" * 40, paths=["tests/task_plane/public_interface/link"])
+
+    def test_runner_rejects_network_tokens_and_output_escape(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-net", claim_id="claim-net") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+
+                with self.assertRaises(CommandDisallowedError):
+                    runner.run("git-show", oid="0" * 40, paths=["https://bad.example/attack"])
+
+                with self.assertRaises(CommandPolicyError):
+                    runner.run(
+                        "acceptance",
+                        oid="0" * 40,
+                        paths=["active-runtime"],
+                        output=workspace.layout.home.parent / "escape.json",
+                    )
+
+    def test_runner_accepts_nested_allowlisted_paths_and_rejects_secret_env(self):
+        with tempfile.TemporaryDirectory() as directory:
+            with TaskWorkspace.create(Path(directory) / "workspace", task_id="task-safe", claim_id="claim-safe") as workspace:
+                runner = SafeCommandRunner(workspace.layout.repo_root, workspace.layout.result_root)
+                command = runner.build(
+                    "pytest",
+                    oid="0" * 40,
+                    paths=["tests/task_plane/public_interface"],
+                )
+                self.assertEqual(command[-1], "tests/task_plane/public_interface")
+                with self.assertRaises(Exception):
+                    runner.run("diff-check", oid="0" * 40, env={"API_TOKEN": "secret"})