feat(openbao): add autonomy runtime v1 candidate

Source: patch/isolated-writer-token-policy-v2-20260724
Source commit: b1e240dda3
Target base: df099420d2
Production runtime changes: not authorized
Tag/Baseline/Registry operations: not authorized

active-runtime/app/aios_isolated_writer_api.py

@@ -5,13 +5,148 @@ import json
 import os
 import subprocess
 from pathlib import Path
+import urllib.parse
 
 import aios_chatgpt_actions_api as api
 import aios_isolated_writer_policy as policy
+from isolated_writer_authz import AuthorizationError, authorize
+from isolated_writer_commit_file import AtomicCommitFile, CommitFileError, OperationLedger
+from isolated_writer_token_registry import TokenRegistry
 
 _reject = api.reject_non_read_http
 _write_file = api.write_file
 _sync_branch = api.sync_branch
+_original_do_get = api.H.do_GET
+api.OPENAPI = Path(os.getenv("AIOS_GITOPS_ISOLATED_WRITER_OPENAPI", str(api.OPENAPI)))
+
+
+def _registry() -> TokenRegistry:
+    return TokenRegistry(Path(os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"]))
+
+
+def _atomic_writer() -> AtomicCommitFile:
+    return AtomicCommitFile(
+        workspace=api.WS,
+        bare_repo=api.BARE_REPO,
+        ledger=OperationLedger(Path(os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"])),
+        remote=os.getenv("AIOS_GITOPS_GITEA_PUSH_REMOTE", ""),
+        audit=policy.append_audit_record,
+    )
+
+
+def _bearer(handler) -> str:
+    value = handler.headers.get("Authorization", "")
+    return value[7:].strip() if value.startswith("Bearer ") else ""
+
+
+def _authorize(handler, *, method: str, path: str, branch=None, target_path=None):
+    grant = authorize(
+        _registry(),
+        _bearer(handler),
+        method=method,
+        path=path,
+        source_ip=str(handler.client_address[0]),
+        workspace=api.WS,
+        branch=branch,
+        target_path=target_path,
+    )
+    grant.record["_source_ip"] = str(handler.client_address[0])
+    return grant
+
+
+def _authorization_error(handler, exc: AuthorizationError):
+    return handler.sendj({"error": exc.code}, exc.status)
+
+
+def _isolated_runtime_status(handler):
+    branch = api.sh(["rev-parse", "--abbrev-ref", "HEAD"], True)["stdout"]
+    head = api.sh(["rev-parse", "HEAD"], True)["stdout"]
+    return handler.sendj({
+        "service": os.getenv(
+            "AIOS_GITOPS_SERVICE_AUDIENCE",
+            "aios-isolated-writer.service",
+        ),
+        "status": "active",
+        "runtime_role": os.getenv("AIOS_GITOPS_RUNTIME_ROLE"),
+        "workspace_role": os.getenv("AIOS_GITOPS_WORKSPACE_ROLE"),
+        "writes_enabled": os.getenv("AIOS_GITOPS_WRITES_ENABLED", "").lower()
+        in {"1", "true", "yes", "on"},
+        "allowed_branch": policy.allowed_branch(),
+        "approved_base_oid": policy.allowed_base_oid(),
+        "workspace_branch": branch,
+        "workspace_head_oid": head,
+        "repository_identity": os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY"),
+        "token_policy_version": 2,
+        "atomic_commit_file": True,
+        "path_policy_sha256": policy.configured_path_policy_sha256(),
+    })
+
+
+def _isolated_do_get(self):
+    path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+    if path in {"/health", "/api/health"}:
+        return self.sendj({
+            "status": "ok",
+            "service": os.getenv(
+                "AIOS_GITOPS_SERVICE_AUDIENCE",
+                "aios-isolated-writer.service",
+            ),
+        })
+    if path not in {"/api/runtime-status", "/api/baseline", "/api/patches"}:
+        return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
+    try:
+        _authorize(self, method="GET", path=path)
+    except AuthorizationError as exc:
+        return _authorization_error(self, exc)
+    if path == "/api/runtime-status":
+        return _isolated_runtime_status(self)
+    return _original_do_get(self)
+
+
+def _read_commit_file_body(handler):
+    try:
+        length = int(handler.headers.get("Content-Length", "0"))
+    except ValueError:
+        raise CommitFileError("writer_content_length_rejected") from None
+    if length <= 0:
+        raise CommitFileError("writer_request_object_required")
+    if length > 1_100_000:
+        raise CommitFileError("writer_request_too_large", 413)
+    raw = handler.rfile.read(length)
+    try:
+        value = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise CommitFileError("writer_request_json_rejected") from None
+    if not isinstance(value, dict):
+        raise CommitFileError("writer_request_object_required")
+    return value
+
+
+def _isolated_do_post(self):
+    path = urllib.parse.urlparse(self.path).path.rstrip("/") or "/"
+    if path != "/api/commit-file":
+        return self.sendj({"error": "isolated_writer_operation_forbidden"}, 403)
+    try:
+        data = _read_commit_file_body(self)
+        branch = str(data.get("branch") or "")
+        target_path = str(data.get("path") or "")
+        grant = _authorize(
+            self,
+            method="POST",
+            path=path,
+            branch=branch,
+            target_path=target_path,
+        )
+        if api.content_has_secret(str(data.get("content") or "")):
+            raise CommitFileError("writer_secret_content_rejected", 403)
+        result = _atomic_writer().execute(grant, data)
+        return self.sendj(result)
+    except AuthorizationError as exc:
+        return _authorization_error(self, exc)
+    except CommitFileError as exc:
+        return self.sendj({"error": exc.code}, exc.status)
+    except PermissionError:
+        return self.sendj({"error": "isolated_writer_policy_rejected"}, 403)
 
 
 def reject_non_read_http(handler, method: str, path: str) -> bool:
@@ -173,6 +308,8 @@ api.H.branch, api.H.file, api.H.commit, api.H.patch = (
     _checked_commit,
     _checked_patch,
 )
+api.H.do_GET = _isolated_do_get
+api.H.do_POST = _isolated_do_post
 
 if __name__ == "__main__":
     policy.require_service_invariants(api.WS, api.BARE_REPO)

active-runtime/app/aios_isolated_writer_policy.py

@@ -9,10 +9,9 @@ import re
 from pathlib import Path, PurePosixPath, PureWindowsPath
 from urllib.parse import unquote
 
-WRITE_ENDPOINTS = frozenset({
+import isolated_writer_path_policy as path_policy
-    "/api/branch", "/api/create-branch", "/api/file", "/api/commit-file",
+
-    "/api/commit", "/api/patch", "/api/create-patch",
+WRITE_ENDPOINTS = frozenset({"/api/commit-file"})
-})
 ALLOWED_ROOTS = (
     "active-runtime",
     "docs",
@@ -69,6 +68,15 @@ def require_service_invariants(workspace: Path, bare_repo: Path) -> None:
         raise PermissionError("isolated_writer_role_required")
     if os.getenv("AIOS_GITOPS_WORKSPACE_ROLE", "") != "isolated-writer":
         raise PermissionError("isolated_writer_workspace_role_required")
+    if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
+        raise PermissionError("isolated_writer_token_policy_v2_required")
+    if not os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", "").strip():
+        raise PermissionError("isolated_writer_service_audience_required")
+    if not os.getenv("AIOS_GITOPS_TOKEN_REGISTRY_V2", "").strip():
+        raise PermissionError("isolated_writer_token_registry_required")
+    if not os.getenv("AIOS_GITOPS_WRITER_OPERATION_LEDGER", "").strip():
+        raise PermissionError("isolated_writer_operation_ledger_required")
+    path_policy.configured_path_policy()
     for name in _DENY_FLAGS:
         if os.getenv(name, "").strip().lower() not in _FALSE:
             raise PermissionError("isolated_writer_forbidden_capability_flag")
@@ -182,6 +190,7 @@ def _audit_sanitize(value):
                         "private_key",
                     )
                 )
+                and str(key).lower() != "token_id"
                 else _audit_sanitize(inner)
             )
             for key, inner in value.items()
@@ -239,9 +248,14 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
     ):
         raise PermissionError("isolated_writer_path_rejected")
     first = parts[0]
+    service_exact_path = False
     if first in DENIED_ROOTS:
-        raise PermissionError("isolated_writer_protected_path_rejected")
+        try:
-    if value != "README.md" and first not in ALLOWED_ROOTS:
+            path_policy.require_service_path("commit_file.atomic", value)
+            service_exact_path = True
+        except path_policy.PathPolicyError as exc:
+            raise PermissionError(str(exc)) from None
+    if value != "README.md" and first not in ALLOWED_ROOTS and not service_exact_path:
         raise PermissionError("isolated_writer_path_outside_allowed_roots")
     if workspace is not None:
         lexical_root = Path(os.path.abspath(str(workspace)))
@@ -261,6 +275,22 @@ def validate_path(path: str, workspace: Path | None = None) -> str:
     return value
 
 
+def validate_content_for_path(path: str, content: str) -> None:
+    if path.split("/", 1)[0] not in DENIED_ROOTS:
+        return
+    try:
+        path_policy.validate_content("commit_file.atomic", path, content)
+    except path_policy.PathPolicyError as exc:
+        raise PermissionError(str(exc)) from None
+
+
+def configured_path_policy_sha256() -> str | None:
+    try:
+        return path_policy.configured_path_policy_sha256()
+    except path_policy.PathPolicyError as exc:
+        raise PermissionError(str(exc)) from None
+
+
 def validate_commit_paths(paths: list[str], workspace: Path | None = None) -> None:
     if enforcement_required():
         for path in paths:

active-runtime/app/isolated_writer_authz.py

@@ -0,0 +1,167 @@
+"""Fail-closed operation authorization for an isolated Git writer."""
+from __future__ import annotations
+
+import ipaddress
+import os
+import re
+import subprocess
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+from isolated_writer_token_registry import TokenRegistry
+
+
+ROUTE_OPERATIONS = {
+    ("GET", "/api/runtime-status"): "runtime.status.read",
+    ("GET", "/api/baseline"): "baseline.read",
+    ("GET", "/api/patches"): "patches.read",
+    ("POST", "/api/commit-file"): "commit_file.atomic",
+}
+OID_RE = re.compile(r"[0-9a-f]{40}")
+
+
+class AuthorizationError(PermissionError):
+    def __init__(self, code: str, status: int = 403):
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+@dataclass(frozen=True)
+class AuthorizedGrant:
+    record: dict[str, Any]
+    operation: str
+    use_count: int
+
+    @property
+    def token_id(self) -> str:
+        return str(self.record["token_id"])
+
+    @property
+    def task_id(self) -> str:
+        return str(self.record["task_id"])
+
+
+def operation_for(method: str, path: str) -> str:
+    operation = ROUTE_OPERATIONS.get((method.upper(), path))
+    if not operation:
+        raise AuthorizationError("isolated_writer_operation_forbidden")
+    return operation
+
+
+def _parse_time(value: Any) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(str(value).replace("Z", "+00:00"))
+    except (TypeError, ValueError):
+        raise AuthorizationError("writer_token_time_invalid") from None
+    if parsed.tzinfo is None:
+        raise AuthorizationError("writer_token_time_invalid")
+    return parsed.astimezone(timezone.utc)
+
+
+def _git_is_ancestor(workspace: Path, base: str, branch: str) -> bool:
+    result = subprocess.run(
+        ["git", "merge-base", "--is-ancestor", base, branch],
+        cwd=workspace,
+        stdout=subprocess.DEVNULL,
+        stderr=subprocess.DEVNULL,
+    )
+    return result.returncode == 0
+
+
+def _validate_grant(
+    grant: dict[str, Any],
+    *,
+    audience: str,
+    source_ip: str,
+    operation: str,
+    branch: str | None,
+    path: str | None,
+    workspace: Path,
+    current_time: datetime,
+) -> None:
+    if int(grant.get("schema_version", 0)) != 2:
+        raise AuthorizationError("writer_token_policy_v2_required", 401)
+    if str(grant.get("audience") or "") != audience:
+        raise AuthorizationError("writer_token_audience_rejected")
+    try:
+        peer = ipaddress.ip_address(source_ip)
+        networks = [ipaddress.ip_network(str(value), strict=True) for value in grant.get("source_cidrs", [])]
+    except ValueError:
+        raise AuthorizationError("writer_token_source_rejected") from None
+    if not networks or not any(peer in network for network in networks):
+        raise AuthorizationError("writer_token_source_rejected")
+    created = _parse_time(grant.get("created_at"))
+    not_before = _parse_time(grant.get("not_before"))
+    expires = _parse_time(grant.get("expires_at"))
+    if expires <= not_before or (expires - created).total_seconds() > 1800:
+        raise AuthorizationError("writer_token_lifetime_rejected")
+    if current_time < not_before:
+        raise AuthorizationError("writer_token_not_yet_valid", 401)
+    if current_time >= expires:
+        raise AuthorizationError("writer_token_expired", 401)
+    if bool(grant.get("revoked")):
+        raise AuthorizationError("writer_token_revoked", 401)
+    operations = grant.get("allowed_operations")
+    if not isinstance(operations, list) or operation not in operations:
+        raise AuthorizationError("writer_token_operation_rejected")
+    expected_branch = str(grant.get("allowed_branch") or "")
+    configured_branch = os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", "")
+    if expected_branch != configured_branch:
+        raise AuthorizationError("writer_token_branch_rejected")
+    if branch is not None and branch != expected_branch:
+        raise AuthorizationError("writer_token_branch_rejected")
+    if path is not None:
+        paths = grant.get("allowed_paths")
+        if not isinstance(paths, list) or path not in paths:
+            raise AuthorizationError("writer_token_path_rejected")
+    base = str(grant.get("approved_base_oid") or "").lower()
+    if not OID_RE.fullmatch(base):
+        raise AuthorizationError("writer_token_base_rejected")
+    if not _git_is_ancestor(workspace, base, expected_branch):
+        raise AuthorizationError("writer_token_base_not_ancestor")
+
+
+def authorize(
+    registry: TokenRegistry,
+    token: str,
+    *,
+    method: str,
+    path: str,
+    source_ip: str,
+    workspace: Path,
+    branch: str | None = None,
+    target_path: str | None = None,
+    current_time: datetime | None = None,
+) -> AuthorizedGrant:
+    if os.getenv("AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED", "") != "2":
+        raise AuthorizationError("writer_token_policy_v2_not_required")
+    if not token:
+        raise AuthorizationError("unauthorized", 401)
+    grant = registry.find(token)
+    if grant is None:
+        raise AuthorizationError("unauthorized", 401)
+    operation = operation_for(method, path)
+    _validate_grant(
+        grant,
+        audience=os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""),
+        source_ip=source_ip,
+        operation=operation,
+        branch=branch,
+        path=target_path,
+        workspace=workspace,
+        current_time=current_time or datetime.now(timezone.utc),
+    )
+    limits = grant.get("max_uses")
+    if not isinstance(limits, dict):
+        raise AuthorizationError("writer_token_use_limit_missing")
+    try:
+        used = registry.consume(str(grant["token_id"]), operation, int(limits[operation]))
+    except (KeyError, TypeError, ValueError):
+        raise AuthorizationError("writer_token_use_limit_missing") from None
+    except PermissionError as exc:
+        code = str(exc)
+        raise AuthorizationError(code, 401 if code == "writer_token_revoked" else 403) from None
+    return AuthorizedGrant(grant, operation, used)

active-runtime/app/isolated_writer_commit_file.py

@@ -0,0 +1,403 @@
+"""Atomic, idempotent commit-file Saga for the isolated writer."""
+from __future__ import annotations
+
+import fcntl
+import hashlib
+import json
+import os
+import re
+import sqlite3
+import subprocess
+import tempfile
+import threading
+from pathlib import Path
+from typing import Any
+
+import aios_isolated_writer_policy as policy
+
+
+OID_RE = re.compile(r"[0-9a-f]{40}")
+IDEMPOTENCY_RE = re.compile(r"[A-Za-z0-9][A-Za-z0-9._:-]{7,127}")
+STATES = frozenset({
+    "PENDING", "FILE_WRITTEN", "COMMITTED", "PUSHED", "VERIFIED", "COMPLETED",
+    "FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED",
+})
+_LOCKS: dict[str, threading.Lock] = {}
+_LOCKS_GUARD = threading.Lock()
+_LEDGER_INIT_LOCK = threading.Lock()
+
+
+class CommitFileError(RuntimeError):
+    def __init__(self, code: str, status: int = 400):
+        super().__init__(code)
+        self.code = code
+        self.status = status
+
+
+class SimulatedCrash(RuntimeError):
+    pass
+
+
+def canonical_digest(data: dict[str, Any]) -> str:
+    payload = {
+        "branch": data["branch"],
+        "path": data["path"],
+        "content_sha256": hashlib.sha256(data["content"].encode("utf-8")).hexdigest(),
+        "message": data["message"],
+        "expected_head_oid": data["expected_head_oid"],
+    }
+    encoded = json.dumps(payload, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+class OperationLedger:
+    def __init__(self, path: Path | str):
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        if self.path.parent.is_symlink() or self.path.is_symlink():
+            raise PermissionError("writer_operation_ledger_symlink_rejected")
+        with _LEDGER_INIT_LOCK:
+            with self.connect() as db:
+                db.execute(
+                    """
+                    CREATE TABLE IF NOT EXISTS writer_operations (
+                        operation_id TEXT PRIMARY KEY,
+                        token_id TEXT NOT NULL,
+                        task_id TEXT NOT NULL,
+                        idempotency_key TEXT NOT NULL UNIQUE,
+                        operation_digest TEXT NOT NULL,
+                        branch TEXT NOT NULL,
+                        path TEXT NOT NULL,
+                        content_sha256 TEXT NOT NULL,
+                        pre_head_oid TEXT NOT NULL,
+                        commit_oid TEXT,
+                        push_status TEXT,
+                        verification_status TEXT,
+                        state TEXT NOT NULL,
+                        created_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                        updated_at TEXT NOT NULL DEFAULT CURRENT_TIMESTAMP,
+                        error_code TEXT
+                    )
+                    """
+                )
+        os.chmod(self.path, 0o600)
+
+    def connect(self) -> sqlite3.Connection:
+        db = sqlite3.connect(self.path, timeout=30)
+        db.row_factory = sqlite3.Row
+        db.execute("PRAGMA journal_mode=WAL")
+        db.execute("PRAGMA synchronous=FULL")
+        return db
+
+    def get_or_create(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
+        key = data["idempotency_key"]
+        with self.connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            row = db.execute(
+                "SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
+            ).fetchone()
+            if row:
+                db.commit()
+                record = dict(row)
+                if record["operation_digest"] != digest:
+                    raise CommitFileError("writer_idempotency_conflict", 409)
+                if (
+                    record["token_id"] != grant.token_id
+                    or record["task_id"] != grant.task_id
+                ):
+                    raise CommitFileError("writer_idempotency_owner_conflict", 409)
+                return record
+            operation_id = hashlib.sha256(
+                f"{grant.task_id}\0{key}".encode("utf-8")
+            ).hexdigest()
+            content_sha = hashlib.sha256(data["content"].encode("utf-8")).hexdigest()
+            db.execute(
+                """
+                INSERT INTO writer_operations(
+                    operation_id,token_id,task_id,idempotency_key,operation_digest,
+                    branch,path,content_sha256,pre_head_oid,state
+                ) VALUES(?,?,?,?,?,?,?,?,?,?)
+                """,
+                (
+                    operation_id, grant.token_id, grant.task_id, key, digest,
+                    data["branch"], data["path"], content_sha,
+                    data["expected_head_oid"], "PENDING",
+                ),
+            )
+            db.commit()
+        return self.get(key)
+
+    def get(self, key: str) -> dict[str, Any]:
+        with self.connect() as db:
+            row = db.execute(
+                "SELECT * FROM writer_operations WHERE idempotency_key=?", (key,)
+            ).fetchone()
+        if row is None:
+            raise KeyError(key)
+        return dict(row)
+
+    def update(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
+        if state not in STATES:
+            raise ValueError("invalid writer operation state")
+        allowed = {"commit_oid", "push_status", "verification_status", "error_code"}
+        assignments = ["state=?", "updated_at=CURRENT_TIMESTAMP"]
+        values: list[Any] = [state]
+        for name, value in fields.items():
+            if name not in allowed:
+                raise ValueError("invalid ledger field")
+            assignments.append(f"{name}=?")
+            values.append(value)
+        values.append(key)
+        with self.connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            db.execute(
+                f"UPDATE writer_operations SET {','.join(assignments)} WHERE idempotency_key=?",
+                values,
+            )
+            db.commit()
+        return self.get(key)
+
+
+class AtomicCommitFile:
+    def __init__(
+        self,
+        *,
+        workspace: Path,
+        bare_repo: Path,
+        ledger: OperationLedger,
+        remote: str,
+        audit,
+        crash_after_state: str | None = None,
+    ):
+        self.workspace = workspace
+        self.bare_repo = bare_repo
+        self.ledger = ledger
+        self.remote = remote
+        self.audit = audit
+        self.crash_after_state = crash_after_state
+
+    def _git(self, *args: str, check: bool = True) -> str:
+        result = subprocess.run(
+            ["git", *args], cwd=self.workspace, text=True, capture_output=True
+        )
+        if check and result.returncode != 0:
+            raise CommitFileError("writer_git_operation_failed", 500)
+        return result.stdout.strip()
+
+    def _transition(self, key: str, state: str, **fields: Any) -> dict[str, Any]:
+        record = self.ledger.update(key, state, **fields)
+        if self.crash_after_state == state:
+            raise SimulatedCrash(state)
+        return record
+
+    def _validate(self, data: Any) -> dict[str, Any]:
+        if not isinstance(data, dict):
+            raise CommitFileError("writer_request_object_required")
+        required = ("branch", "path", "content", "message", "expected_head_oid", "idempotency_key")
+        if any(field not in data for field in required):
+            raise CommitFileError("writer_request_incomplete")
+        if data.get("force") or data.get("force_push"):
+            raise CommitFileError("isolated_writer_force_rejected", 403)
+        if set(data) != set(required):
+            raise CommitFileError("writer_request_fields_rejected")
+        item = {field: data[field] for field in required}
+        if item["branch"] != policy.allowed_branch():
+            raise CommitFileError("isolated_writer_branch_rejected", 403)
+        try:
+            item["path"] = policy.validate_path(item["path"], self.workspace)
+        except PermissionError as exc:
+            raise CommitFileError(str(exc), 403) from None
+        if not isinstance(item["content"], str) or "\x00" in item["content"]:
+            raise CommitFileError("writer_content_rejected")
+        try:
+            policy.validate_content_for_path(item["path"], item["content"])
+        except PermissionError as exc:
+            raise CommitFileError(str(exc), 403) from None
+        if len(item["content"].encode("utf-8")) > 1_048_576:
+            raise CommitFileError("writer_content_too_large", 413)
+        message = item["message"]
+        if (
+            not isinstance(message, str)
+            or not 1 <= len(message) <= 200
+            or "\x00" in message
+            or "\n" in message
+            or "\r" in message
+        ):
+            raise CommitFileError("writer_message_rejected")
+        if not isinstance(item["expected_head_oid"], str) or not OID_RE.fullmatch(item["expected_head_oid"]):
+            raise CommitFileError("writer_expected_head_rejected")
+        if not isinstance(item["idempotency_key"], str) or not IDEMPOTENCY_RE.fullmatch(item["idempotency_key"]):
+            raise CommitFileError("writer_idempotency_key_rejected")
+        return item
+
+    def _atomic_write(self, relative: str, content: str) -> None:
+        target = self.workspace / relative
+        target.parent.mkdir(parents=True, exist_ok=True)
+        fd, temporary = tempfile.mkstemp(prefix=".writer-", dir=target.parent)
+        try:
+            with os.fdopen(fd, "wb", closefd=True) as stream:
+                stream.write(content.encode("utf-8"))
+                stream.flush()
+                os.fsync(stream.fileno())
+            os.replace(temporary, target)
+            parent_fd = os.open(target.parent, os.O_RDONLY | getattr(os, "O_DIRECTORY", 0))
+            try:
+                os.fsync(parent_fd)
+            finally:
+                os.close(parent_fd)
+        finally:
+            if os.path.exists(temporary):
+                os.unlink(temporary)
+
+    def _verify_remote(self, branch: str, oid: str) -> None:
+        bare = subprocess.run(
+            ["git", "--git-dir", str(self.bare_repo), "rev-parse", f"refs/heads/{branch}"],
+            text=True, capture_output=True,
+        )
+        if bare.returncode != 0 or bare.stdout.strip() != oid:
+            raise CommitFileError("writer_authoritative_bare_oid_mismatch", 500)
+        remote = subprocess.run(
+            ["git", "ls-remote", "--heads", self.remote, f"refs/heads/{branch}"],
+            cwd=self.workspace, text=True, capture_output=True,
+        )
+        if remote.returncode != 0:
+            raise CommitFileError("writer_gitea_verification_failed", 503)
+        fields = remote.stdout.strip().split()
+        if len(fields) != 2:
+            raise CommitFileError("writer_gitea_ref_missing", 503)
+        if fields[0] != oid:
+            raise CommitFileError("writer_gitea_oid_conflict", 409)
+
+    def execute(self, grant: Any, raw_data: Any) -> dict[str, Any]:
+        data = self._validate(raw_data)
+        digest = canonical_digest(data)
+        branch = data["branch"]
+        with _LOCKS_GUARD:
+            lock = _LOCKS.setdefault(branch, threading.Lock())
+        lock_path = self.ledger.path.parent / f"{hashlib.sha256(branch.encode()).hexdigest()}.lock"
+        with lock, open(lock_path, "a+b") as branch_lock:
+            fcntl.flock(branch_lock.fileno(), fcntl.LOCK_EX)
+            return self._execute_locked(grant, data, digest)
+
+    def _execute_locked(self, grant: Any, data: dict[str, Any], digest: str) -> dict[str, Any]:
+        key = data["idempotency_key"]
+        record = self.ledger.get_or_create(grant, data, digest)
+        if record["state"] == "COMPLETED":
+            return self._result(record, idempotent=True)
+        if self._git("rev-parse", "--abbrev-ref", "HEAD") != data["branch"]:
+            record = self._transition(
+                key, "MANUAL_REVIEW_REQUIRED", error_code="writer_current_branch_conflict"
+            )
+            self._audit(grant, data, digest, record, "manual_review_required")
+            raise CommitFileError("writer_current_branch_conflict", 409)
+        head = self._git("rev-parse", "HEAD")
+        commit_oid = record.get("commit_oid")
+        if head != record["pre_head_oid"] and head != commit_oid:
+            record = self._transition(
+                key, "MANUAL_REVIEW_REQUIRED", error_code="writer_branch_head_conflict"
+            )
+            self._audit(grant, data, digest, record, "manual_review_required")
+            raise CommitFileError("writer_branch_head_conflict", 409)
+        try:
+            state = record["state"]
+            if state == "MANUAL_REVIEW_REQUIRED":
+                raise CommitFileError("writer_manual_review_required", 409)
+            if state == "FAILED_RECOVERABLE":
+                if record.get("push_status") == "confirmed":
+                    state = "PUSHED"
+                else:
+                    state = "COMMITTED" if commit_oid else "FILE_WRITTEN"
+            if state == "PENDING":
+                self._atomic_write(data["path"], data["content"])
+                record = self._transition(key, "FILE_WRITTEN")
+                state = record["state"]
+            if state == "FILE_WRITTEN":
+                actual_sha = hashlib.sha256(
+                    (self.workspace / data["path"]).read_bytes()
+                ).hexdigest()
+                if actual_sha != record["content_sha256"]:
+                    raise CommitFileError("writer_file_content_conflict", 409)
+                staged_before = self._git("diff", "--cached", "--name-only", check=False).splitlines()
+                if staged_before:
+                    raise CommitFileError("writer_extra_staged_files_rejected", 409)
+                self._git("add", "--", data["path"])
+                staged = self._git("diff", "--cached", "--name-only").splitlines()
+                if staged != [data["path"]]:
+                    raise CommitFileError("writer_extra_staged_files_rejected", 409)
+                self._git("commit", "-m", data["message"])
+                commit_oid = self._git("rev-parse", "HEAD")
+                record = self._transition(key, "COMMITTED", commit_oid=commit_oid)
+                state = record["state"]
+            if state == "COMMITTED":
+                commit_oid = str(record["commit_oid"])
+                pushed = subprocess.run(
+                    ["git", "push", "--porcelain", self.remote, f"HEAD:refs/heads/{data['branch']}"],
+                    cwd=self.workspace, text=True, capture_output=True,
+                )
+                if pushed.returncode != 0:
+                    self._transition(
+                        key, "FAILED_RECOVERABLE",
+                        push_status="failed", error_code="writer_gitea_push_failed",
+                    )
+                    raise CommitFileError("writer_gitea_push_failed", 503)
+                record = self._transition(key, "PUSHED", push_status="confirmed")
+                state = record["state"]
+            if state == "PUSHED":
+                commit_oid = str(record["commit_oid"])
+                self._verify_remote(data["branch"], commit_oid)
+                record = self._transition(
+                    key, "VERIFIED", verification_status="bare+gitea"
+                )
+                state = record["state"]
+            if state == "VERIFIED":
+                record = self._transition(key, "COMPLETED", error_code=None)
+            self._audit(grant, data, digest, record, "completed")
+            return self._result(record, idempotent=False)
+        except SimulatedCrash:
+            raise
+        except CommitFileError as exc:
+            latest = self.ledger.get(key)
+            if latest["state"] not in {"FAILED_RECOVERABLE", "MANUAL_REVIEW_REQUIRED"}:
+                failure_state = (
+                    "MANUAL_REVIEW_REQUIRED"
+                    if exc.code in {
+                        "writer_gitea_oid_conflict",
+                        "writer_file_content_conflict",
+                        "writer_current_branch_conflict",
+                    }
+                    else "FAILED_RECOVERABLE"
+                )
+                latest = self._transition(key, failure_state, error_code=exc.code)
+            self._audit(grant, data, digest, latest, "failed")
+            raise
+
+    def _audit(self, grant: Any, data: dict[str, Any], digest: str, record: dict[str, Any], result: str) -> None:
+        self.audit({
+            "token_id": grant.token_id,
+            "task_id": grant.task_id,
+            "audience": grant.record["audience"],
+            "source_ip": grant.record.get("_source_ip", "socket-peer"),
+            "operation": grant.operation,
+            "branch": data["branch"],
+            "path": data["path"],
+            "idempotency_key": data["idempotency_key"],
+            "operation_digest": digest,
+            "path_policy_sha256": policy.configured_path_policy_sha256(),
+            "pre_head_oid": record["pre_head_oid"],
+            "commit_oid": record.get("commit_oid"),
+            "result": result,
+        })
+
+    @staticmethod
+    def _result(record: dict[str, Any], *, idempotent: bool) -> dict[str, Any]:
+        return {
+            "status": "completed",
+            "state": record["state"],
+            "operation_id": record["operation_id"],
+            "branch": record["branch"],
+            "path": record["path"],
+            "commit_oid": record.get("commit_oid"),
+            "push_status": record.get("push_status"),
+            "verification_status": record.get("verification_status"),
+            "idempotent": idempotent,
+        }

active-runtime/app/isolated_writer_path_policy.py

@@ -0,0 +1,422 @@
+"""Strict service-level exact-path policy for isolated Git writers."""
+from __future__ import annotations
+
+import grp
+import hashlib
+import json
+import os
+import re
+import stat
+from dataclasses import dataclass
+from functools import lru_cache
+from pathlib import Path, PurePosixPath, PureWindowsPath
+from typing import Any
+from urllib.parse import unquote
+
+
+MAX_POLICY_BYTES = 65_536
+POLICY_FIELDS = frozenset(
+    {
+        "schema_version",
+        "policy_id",
+        "audience",
+        "repository_identity",
+        "allowed_branch",
+        "allowed_operations",
+    }
+)
+OPERATION_FIELDS = frozenset({"exact_paths", "max_content_bytes", "content_profile"})
+SUPPORTED_OPERATIONS = frozenset({"commit_file.atomic"})
+SUPPORTED_CONTENT_PROFILES = frozenset({"systemd-codex-review-worker-v1"})
+_SECRET_RE = re.compile(
+    r"(?i)(authorization\s*:|bearer\s+|(?:token|password|secret)\s*[:=])"
+)
+_SYSTEMD_NAME_RE = re.compile(r"[A-Za-z][A-Za-z0-9]*\Z")
+_SYSTEMD_MAX_LINE_BYTES = 4096
+_SYSTEMD_ALLOWED_DIRECTIVES = {
+    "Unit": frozenset({"Description", "After", "Wants"}),
+    "Service": frozenset(
+        {
+            "Type",
+            "User",
+            "Group",
+            "WorkingDirectory",
+            "Environment",
+            "ExecStart",
+            "Restart",
+            "RestartSec",
+            "NoNewPrivileges",
+            "PrivateTmp",
+            "PrivateDevices",
+            "ProtectSystem",
+            "ProtectHome",
+            "ProtectKernelTunables",
+            "ProtectKernelModules",
+            "ProtectControlGroups",
+            "LockPersonality",
+            "RestrictSUIDSGID",
+            "RestrictRealtime",
+            "MemoryDenyWriteExecute",
+            "RuntimeDirectory",
+            "StateDirectory",
+            "CacheDirectory",
+            "ReadOnlyPaths",
+            "ReadWritePaths",
+            "UMask",
+        }
+    ),
+    "Install": frozenset({"WantedBy"}),
+}
+_SYSTEMD_MULTI_DIRECTIVES = frozenset(
+    {
+        ("Service", "Environment"),
+        ("Service", "ReadOnlyPaths"),
+        ("Service", "ReadWritePaths"),
+    }
+)
+_SYSTEMD_EXACT_SINGLE_VALUES = {
+    ("Unit", "Description"): "AIOS Codex Review/Test Worker",
+    ("Unit", "After"): "network-online.target",
+    ("Unit", "Wants"): "network-online.target",
+    ("Service", "Type"): "simple",
+    ("Service", "User"): "aiman",
+    ("Service", "Group"): "gitops",
+    (
+        "Service",
+        "WorkingDirectory",
+    ): "/usr/libexec/aios-codex-review-test-worker/current",
+    (
+        "Service",
+        "ExecStart",
+    ): "/usr/bin/python3 -m task_plane.workers.codex_review_test",
+    ("Service", "Restart"): "on-failure",
+    ("Service", "RestartSec"): "2",
+    ("Service", "NoNewPrivileges"): "true",
+    ("Service", "PrivateTmp"): "true",
+    ("Service", "PrivateDevices"): "true",
+    ("Service", "ProtectSystem"): "strict",
+    ("Service", "ProtectHome"): "true",
+    ("Service", "ProtectKernelTunables"): "true",
+    ("Service", "ProtectKernelModules"): "true",
+    ("Service", "ProtectControlGroups"): "true",
+    ("Service", "LockPersonality"): "true",
+    ("Service", "RestrictSUIDSGID"): "true",
+    ("Service", "RestrictRealtime"): "true",
+    ("Service", "MemoryDenyWriteExecute"): "true",
+    ("Service", "RuntimeDirectory"): "aios-codex-review-test-worker",
+    ("Service", "StateDirectory"): "aios-codex-review-test-worker",
+    ("Service", "CacheDirectory"): "aios-codex-review-test-worker",
+    ("Service", "UMask"): "0077",
+    ("Install", "WantedBy"): "multi-user.target",
+}
+_SYSTEMD_EXACT_MULTI_VALUES = {
+    ("Service", "Environment"): (
+        "AIOS_CODEX_WORKER_MODE=shadow",
+        "AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
+        "PYTHONDONTWRITEBYTECODE=1",
+    ),
+    ("Service", "ReadOnlyPaths"): (
+        "/data/aios-gitops/repo.git",
+        "/usr/libexec/aios-codex-review-test-worker",
+    ),
+    ("Service", "ReadWritePaths"): (
+        "/run/aios-codex-review-test-worker",
+        "/var/lib/aios-codex-review-test-worker",
+        "/var/cache/aios-codex-review-test-worker",
+    ),
+}
+
+
+class PathPolicyError(PermissionError):
+    pass
+
+
+@dataclass(frozen=True)
+class OperationPolicy:
+    exact_paths: tuple[str, ...]
+    max_content_bytes: int
+    content_profile: str
+
+
+@dataclass(frozen=True)
+class ServicePathPolicy:
+    schema_version: int
+    policy_id: str
+    audience: str
+    repository_identity: str
+    allowed_branch: str
+    allowed_operations: dict[str, OperationPolicy]
+    sha256: str
+
+    def operation(self, name: str) -> OperationPolicy:
+        try:
+            return self.allowed_operations[name]
+        except KeyError:
+            raise PathPolicyError("isolated_writer_service_operation_rejected") from None
+
+    def require_exact_path(self, operation: str, path: str) -> OperationPolicy:
+        selected = self.operation(operation)
+        if path not in selected.exact_paths:
+            raise PathPolicyError("isolated_writer_service_path_rejected")
+        return selected
+
+
+def _canonical_json(value: Any) -> bytes:
+    return (
+        json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+        + "\n"
+    ).encode("utf-8")
+
+
+def validate_relative_path_syntax(path: str) -> str:
+    if not isinstance(path, str):
+        raise PathPolicyError("isolated_writer_path_rejected")
+    if not path or path != path.strip() or "\x00" in path or "\\" in path:
+        raise PathPolicyError("isolated_writer_path_rejected")
+    try:
+        decoded = unquote(path, errors="strict")
+    except (UnicodeDecodeError, ValueError):
+        raise PathPolicyError("isolated_writer_path_rejected") from None
+    if decoded != path:
+        raise PathPolicyError("isolated_writer_path_rejected")
+    posix = PurePosixPath(path)
+    windows = PureWindowsPath(path)
+    parts = path.split("/")
+    if (
+        posix.is_absolute()
+        or windows.is_absolute()
+        or bool(windows.drive)
+        or any(part in {"", ".", ".."} for part in parts)
+    ):
+        raise PathPolicyError("isolated_writer_path_rejected")
+    return path
+
+
+def _require_fields(value: dict[str, Any], allowed: frozenset[str], code: str) -> None:
+    if set(value) != allowed:
+        raise PathPolicyError(code)
+
+
+def _parse_policy(document: Any, canonical: bytes) -> ServicePathPolicy:
+    if not isinstance(document, dict):
+        raise PathPolicyError("isolated_writer_path_policy_object_required")
+    _require_fields(document, POLICY_FIELDS, "isolated_writer_path_policy_fields_rejected")
+    if document["schema_version"] != 1:
+        raise PathPolicyError("isolated_writer_path_policy_schema_rejected")
+    text_fields = (
+        "policy_id",
+        "audience",
+        "repository_identity",
+        "allowed_branch",
+    )
+    if any(
+        not isinstance(document[field], str) or not document[field].strip()
+        for field in text_fields
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_value_rejected")
+    raw_operations = document["allowed_operations"]
+    if not isinstance(raw_operations, dict) or not raw_operations:
+        raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
+    if not set(raw_operations).issubset(SUPPORTED_OPERATIONS):
+        raise PathPolicyError("isolated_writer_path_policy_operations_rejected")
+    operations: dict[str, OperationPolicy] = {}
+    for name, raw in raw_operations.items():
+        if not isinstance(raw, dict):
+            raise PathPolicyError("isolated_writer_path_policy_operation_rejected")
+        _require_fields(
+            raw,
+            OPERATION_FIELDS,
+            "isolated_writer_path_policy_operation_fields_rejected",
+        )
+        paths = raw["exact_paths"]
+        if (
+            not isinstance(paths, list)
+            or not paths
+            or any(not isinstance(item, str) for item in paths)
+        ):
+            raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
+        validated = tuple(validate_relative_path_syntax(item) for item in paths)
+        if len(set(validated)) != len(validated):
+            raise PathPolicyError("isolated_writer_path_policy_paths_rejected")
+        maximum = raw["max_content_bytes"]
+        if not isinstance(maximum, int) or isinstance(maximum, bool) or not 1 <= maximum <= 1_048_576:
+            raise PathPolicyError("isolated_writer_path_policy_size_rejected")
+        profile = raw["content_profile"]
+        if profile not in SUPPORTED_CONTENT_PROFILES:
+            raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
+        operations[name] = OperationPolicy(validated, maximum, profile)
+    return ServicePathPolicy(
+        schema_version=1,
+        policy_id=document["policy_id"],
+        audience=document["audience"],
+        repository_identity=document["repository_identity"],
+        allowed_branch=document["allowed_branch"],
+        allowed_operations=operations,
+        sha256=hashlib.sha256(canonical).hexdigest(),
+    )
+
+
+def load_path_policy(
+    path: Path,
+    *,
+    expected_owner_uid: int = 0,
+    expected_group_gid: int | None = None,
+    expected_parent_mode: int = 0o750,
+    expected_file_mode: int = 0o640,
+) -> ServicePathPolicy:
+    lexical = Path(os.path.abspath(str(path)))
+    if lexical.is_symlink():
+        raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
+    try:
+        parent_info = lexical.parent.stat(follow_symlinks=False)
+        file_info = lexical.stat(follow_symlinks=False)
+    except FileNotFoundError:
+        raise PathPolicyError("isolated_writer_path_policy_missing") from None
+    if stat.S_ISLNK(parent_info.st_mode) or not stat.S_ISDIR(parent_info.st_mode):
+        raise PathPolicyError("isolated_writer_path_policy_parent_rejected")
+    if stat.S_ISLNK(file_info.st_mode) or not stat.S_ISREG(file_info.st_mode):
+        raise PathPolicyError("isolated_writer_path_policy_file_rejected")
+    if lexical.resolve(strict=True) != lexical:
+        raise PathPolicyError("isolated_writer_path_policy_symlink_rejected")
+    if (
+        parent_info.st_uid != expected_owner_uid
+        or stat.S_IMODE(parent_info.st_mode) != expected_parent_mode
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_parent_permissions_rejected")
+    if (
+        file_info.st_uid != expected_owner_uid
+        or stat.S_IMODE(file_info.st_mode) != expected_file_mode
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_permissions_rejected")
+    if expected_group_gid is not None and (
+        parent_info.st_gid != expected_group_gid or file_info.st_gid != expected_group_gid
+    ):
+        raise PathPolicyError("isolated_writer_path_policy_group_rejected")
+    if file_info.st_size <= 0 or file_info.st_size > MAX_POLICY_BYTES:
+        raise PathPolicyError("isolated_writer_path_policy_file_size_rejected")
+    raw = lexical.read_bytes()
+    try:
+        document = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise PathPolicyError("isolated_writer_path_policy_json_rejected") from None
+    canonical = _canonical_json(document)
+    if raw != canonical:
+        raise PathPolicyError("isolated_writer_path_policy_not_canonical")
+    return _parse_policy(document, canonical)
+
+
+@lru_cache(maxsize=1)
+def configured_path_policy() -> ServicePathPolicy | None:
+    configured = os.getenv("AIOS_GITOPS_PATH_POLICY_FILE", "").strip()
+    if not configured:
+        return None
+    if not Path(configured).is_absolute():
+        raise PathPolicyError("isolated_writer_path_policy_path_rejected")
+    try:
+        gitops_gid = grp.getgrnam("gitops").gr_gid
+    except KeyError:
+        raise PathPolicyError("isolated_writer_path_policy_group_rejected") from None
+    loaded = load_path_policy(Path(configured), expected_group_gid=gitops_gid)
+    if loaded.audience != os.getenv("AIOS_GITOPS_SERVICE_AUDIENCE", ""):
+        raise PathPolicyError("isolated_writer_path_policy_audience_rejected")
+    if loaded.repository_identity != os.getenv("AIOS_GITOPS_REPOSITORY_IDENTITY", ""):
+        raise PathPolicyError("isolated_writer_path_policy_repository_rejected")
+    if loaded.allowed_branch != os.getenv("AIOS_GITOPS_ALLOWED_BRANCH", ""):
+        raise PathPolicyError("isolated_writer_path_policy_branch_rejected")
+    return loaded
+
+
+def clear_path_policy_cache() -> None:
+    configured_path_policy.cache_clear()
+
+
+def require_service_path(operation: str, path: str) -> OperationPolicy:
+    loaded = configured_path_policy()
+    if loaded is None:
+        raise PathPolicyError("isolated_writer_service_path_policy_required")
+    return loaded.require_exact_path(operation, path)
+
+
+def configured_path_policy_sha256() -> str | None:
+    loaded = configured_path_policy()
+    return None if loaded is None else loaded.sha256
+
+
+def validate_content(operation: str, path: str, content: str) -> None:
+    selected = require_service_path(operation, path)
+    try:
+        encoded = content.encode("utf-8", errors="strict")
+    except (AttributeError, UnicodeEncodeError):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
+    if len(encoded) > selected.max_content_bytes:
+        raise PathPolicyError("isolated_writer_service_content_too_large")
+    if selected.content_profile == "systemd-codex-review-worker-v1":
+        _validate_systemd_codex_review_worker(content)
+        return
+    raise PathPolicyError("isolated_writer_path_policy_profile_rejected")
+
+
+def _validate_systemd_codex_review_worker(content: str) -> None:
+    if not isinstance(content, str):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    try:
+        content.encode("utf-8", errors="strict")
+    except UnicodeEncodeError:
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected") from None
+    if (
+        "\x00" in content
+        or "\r" in content
+        or _SECRET_RE.search(content)
+        or not content.endswith("\n")
+    ):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+
+    sections: set[str] = set()
+    values: dict[tuple[str, str], list[str]] = {}
+    section: str | None = None
+    for line in content.splitlines():
+        if len(line.encode("utf-8")) > _SYSTEMD_MAX_LINE_BYTES:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        if not line:
+            continue
+        if line[0].isspace() or line.startswith(("#", ";")):
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        if line.startswith("[") and line.endswith("]"):
+            candidate = line[1:-1]
+            if (
+                not _SYSTEMD_NAME_RE.fullmatch(candidate)
+                or candidate not in _SYSTEMD_ALLOWED_DIRECTIVES
+                or candidate in sections
+            ):
+                raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+            sections.add(candidate)
+            section = candidate
+            continue
+        if section is None or "=" not in line:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        directive, value = line.split("=", 1)
+        if (
+            not _SYSTEMD_NAME_RE.fullmatch(directive)
+            or directive not in _SYSTEMD_ALLOWED_DIRECTIVES[section]
+            or not value
+            or value != value.strip()
+        ):
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        key = (section, directive)
+        if key in values and key not in _SYSTEMD_MULTI_DIRECTIVES:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+        values.setdefault(key, []).append(value)
+
+    if sections != set(_SYSTEMD_ALLOWED_DIRECTIVES):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    if set(values) != set(_SYSTEMD_EXACT_SINGLE_VALUES) | set(
+        _SYSTEMD_EXACT_MULTI_VALUES
+    ):
+        raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    for key, expected in _SYSTEMD_EXACT_SINGLE_VALUES.items():
+        if values.get(key) != [expected]:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")
+    for key, expected in _SYSTEMD_EXACT_MULTI_VALUES.items():
+        observed = values.get(key, [])
+        if len(observed) != len(set(observed)) or tuple(observed) != expected:
+            raise PathPolicyError("isolated_writer_systemd_profile_rejected")

active-runtime/app/isolated_writer_token_registry.py

@@ -0,0 +1,241 @@
+"""Digest-only, task-scoped token registry for isolated writers."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import os
+import grp
+import pwd
+import secrets
+import sqlite3
+import stat
+import threading
+from datetime import datetime, timezone
+from pathlib import Path
+from typing import Any
+
+_REGISTRY_INIT_LOCK = threading.Lock()
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+
+
+def token_digest(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+class TokenRegistry:
+    def __init__(self, path: Path | str):
+        self.path = Path(path)
+        self._prepare()
+
+    def _prepare(self) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        if self.path.parent.is_symlink() or self.path.is_symlink():
+            raise PermissionError("writer_token_registry_symlink_rejected")
+        with _REGISTRY_INIT_LOCK:
+            with self._connect() as db:
+                db.executescript(
+                    """
+                    CREATE TABLE IF NOT EXISTS token_grants (
+                        token_id TEXT PRIMARY KEY,
+                        token_digest TEXT NOT NULL UNIQUE,
+                        grant_json TEXT NOT NULL,
+                        revoked INTEGER NOT NULL DEFAULT 0,
+                        created_at TEXT NOT NULL
+                    );
+                    CREATE TABLE IF NOT EXISTS token_uses (
+                        token_id TEXT NOT NULL,
+                        operation TEXT NOT NULL,
+                        use_count INTEGER NOT NULL DEFAULT 0,
+                        PRIMARY KEY(token_id, operation),
+                        FOREIGN KEY(token_id) REFERENCES token_grants(token_id)
+                    );
+                    """
+                )
+        os.chmod(self.path, 0o600)
+
+    def _connect(self) -> sqlite3.Connection:
+        db = sqlite3.connect(self.path, timeout=30, isolation_level=None)
+        db.row_factory = sqlite3.Row
+        db.execute("PRAGMA journal_mode=WAL")
+        db.execute("PRAGMA synchronous=FULL")
+        db.execute("PRAGMA foreign_keys=ON")
+        return db
+
+    def issue(self, grant: dict[str, Any], token: str | None = None) -> tuple[str, dict[str, Any]]:
+        value = token or secrets.token_urlsafe(32)
+        if len(value.encode("utf-8")) < 32:
+            raise ValueError("writer_token_entropy_insufficient")
+        record = dict(grant)
+        record["schema_version"] = 2
+        record["token_digest"] = token_digest(value)
+        record["revoked"] = False
+        record.setdefault("created_at", utc_now())
+        required = (
+            "token_id",
+            "audience",
+            "task_id",
+            "source_cidrs",
+            "allowed_branch",
+            "approved_base_oid",
+            "allowed_operations",
+            "allowed_paths",
+            "not_before",
+            "expires_at",
+            "max_uses",
+            "issued_by",
+        )
+        if any(not record.get(field) for field in required):
+            raise ValueError("writer_token_grant_incomplete")
+        encoded = json.dumps(record, sort_keys=True, separators=(",", ":"))
+        with self._connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            try:
+                db.execute(
+                    """
+                    INSERT INTO token_grants(token_id, token_digest, grant_json, revoked, created_at)
+                    VALUES(?,?,?,?,?)
+                    """,
+                    (
+                        str(record["token_id"]),
+                        str(record["token_digest"]),
+                        encoded,
+                        0,
+                        str(record["created_at"]),
+                    ),
+                )
+                db.execute("COMMIT")
+            except Exception:
+                db.execute("ROLLBACK")
+                raise
+        return value, record
+
+    def find(self, token: str) -> dict[str, Any] | None:
+        candidate = token_digest(token)
+        with self._connect() as db:
+            rows = db.execute(
+                "SELECT token_digest, grant_json, revoked FROM token_grants"
+            ).fetchall()
+        match = None
+        for row in rows:
+            if hmac.compare_digest(str(row["token_digest"]), candidate):
+                match = dict(json.loads(str(row["grant_json"])))
+                match["revoked"] = bool(row["revoked"])
+        return match
+
+    def consume(self, token_id: str, operation: str, maximum: int) -> int:
+        if maximum < 1:
+            raise PermissionError("writer_token_operation_exhausted")
+        with self._connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            try:
+                row = db.execute(
+                    "SELECT revoked FROM token_grants WHERE token_id=?",
+                    (token_id,),
+                ).fetchone()
+                if row is None or bool(row["revoked"]):
+                    raise PermissionError("writer_token_revoked")
+                used_row = db.execute(
+                    "SELECT use_count FROM token_uses WHERE token_id=? AND operation=?",
+                    (token_id, operation),
+                ).fetchone()
+                used = int(used_row["use_count"]) if used_row else 0
+                if used >= maximum:
+                    raise PermissionError("writer_token_operation_exhausted")
+                current = used + 1
+                db.execute(
+                    """
+                    INSERT INTO token_uses(token_id, operation, use_count)
+                    VALUES(?,?,?)
+                    ON CONFLICT(token_id, operation)
+                    DO UPDATE SET use_count=excluded.use_count
+                    """,
+                    (token_id, operation, current),
+                )
+                db.execute("COMMIT")
+                return current
+            except Exception:
+                db.execute("ROLLBACK")
+                raise
+
+    def revoke(self, token_id: str) -> bool:
+        with self._connect() as db:
+            db.execute("BEGIN IMMEDIATE")
+            changed = db.execute(
+                "UPDATE token_grants SET revoked=1 WHERE token_id=? AND revoked=0",
+                (token_id,),
+            ).rowcount
+            db.execute("COMMIT")
+        return bool(changed)
+
+    def usage(self, token_id: str, operation: str) -> int:
+        with self._connect() as db:
+            row = db.execute(
+                "SELECT use_count FROM token_uses WHERE token_id=? AND operation=?",
+                (token_id, operation),
+            ).fetchone()
+        return int(row["use_count"]) if row else 0
+
+
+def _write_secret_file(
+    path: Path,
+    token: str,
+    *,
+    owner: str | None = None,
+    group: str | None = None,
+) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    if path.parent.is_symlink() or path.is_symlink():
+        raise PermissionError("writer_token_file_symlink_rejected")
+    flags = os.O_WRONLY | os.O_CREAT | os.O_EXCL
+    if hasattr(os, "O_NOFOLLOW"):
+        flags |= os.O_NOFOLLOW
+    fd = os.open(path, flags, 0o600)
+    try:
+        if not stat.S_ISREG(os.fstat(fd).st_mode):
+            raise PermissionError("writer_token_file_not_regular")
+        if owner is not None or group is not None:
+            uid = pwd.getpwnam(owner).pw_uid if owner is not None else -1
+            gid = grp.getgrnam(group).gr_gid if group is not None else -1
+            os.fchown(fd, uid, gid)
+        os.write(fd, token.encode("utf-8"))
+        os.fsync(fd)
+    finally:
+        os.close(fd)
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--registry", required=True)
+    sub = parser.add_subparsers(dest="command", required=True)
+    issue = sub.add_parser("issue")
+    issue.add_argument("--grant-file", required=True)
+    issue.add_argument("--token-file", required=True)
+    issue.add_argument("--owner")
+    issue.add_argument("--group")
+    revoke = sub.add_parser("revoke")
+    revoke.add_argument("--token-id", required=True)
+    args = parser.parse_args(argv)
+    registry = TokenRegistry(args.registry)
+    if args.command == "issue":
+        grant = json.loads(Path(args.grant_file).read_text(encoding="utf-8"))
+        token, record = registry.issue(grant)
+        _write_secret_file(
+            Path(args.token_file),
+            token,
+            owner=args.owner,
+            group=args.group,
+        )
+        print(json.dumps({"token_id": record["token_id"], "created": True}, sort_keys=True))
+        return 0
+    changed = registry.revoke(args.token_id)
+    print(json.dumps({"token_id": args.token_id, "revoked": changed}, sort_keys=True))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

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

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

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

@@ -1,212 +0,0 @@
-"""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

@@ -1,62 +0,0 @@
-"""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

@@ -1,56 +0,0 @@
-"""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

@@ -1,157 +0,0 @@
-"""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

@@ -1,138 +0,0 @@
-"""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

@@ -1,300 +0,0 @@
-"""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

@@ -1,162 +0,0 @@
-"""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/openbao-autonomy/admin-bootstrap-openbao-autonomy-v1-v3.self-test.json

@@ -0,0 +1 @@
+{"checks": {"72h_acceptance_gate": true, "approle_implemented": true, "audit_implemented": true, "backup_implemented": true, "branch_protection_gate_not_bypassable": true, "charter_store_sign_verify": true, "continuous_optimization_implemented": true, "declarative_self_initialization": true, "embedded_shell_syntax": true, "exact_checksum_duplicate_rejected": true, "exact_checksum_good_fixture": true, "exact_checksum_missing_rejected": true, "fault_injection_implemented": true, "gitea_1_22_6_rejected": true, "gitea_1_23_0_accepted": true, "gitea_tag_api_version_gate": true, "isolated_restore_implemented": true, "kv_v2_implemented": true, "machine_credential_persisted": true, "no_git_force_push": true, "no_git_tag_mutation": true, "openbao_ready_gate_not_bypassable": true, "outer_bash_syntax": true, "policy_implemented": true, "ready_not_predeclared": true, "root_token_not_persisted": true, "self_contained_no_sibling_preflight": true, "self_init_secret_removal_implemented": true, "tag_protection_gate_implemented": true, "transit_charter_key_implemented": true, "transit_runtime_key_implemented": true}, "passed": 31, "runtime_claims": false, "schema": "AIOS_OPENBAO_BOOTSTRAP_STATIC_SELF_TEST_V3", "scope": "static_and_fixture_only", "script_sha256": "4bece5edc8be5d2024b5b81ce569549625c2bf318fdaed1066773b379b76cc66", "total": 31}

deploy/openbao-autonomy/admin-bootstrap-openbao-autonomy-v1-v3.sha256

@@ -0,0 +1 @@
+4bece5edc8be5d2024b5b81ce569549625c2bf318fdaed1066773b379b76cc66  admin-bootstrap-openbao-autonomy-v1-v3.sh

deploy/openbao-autonomy/checksums.sha256

@@ -0,0 +1,4 @@
+aa5a8eb14e80d23c2e3f8a36cdeea4b3cd026877eb4aa424637a85665388e6ec  admin-bootstrap-openbao-autonomy-v1-v3.self-test.json
+4bece5edc8be5d2024b5b81ce569549625c2bf318fdaed1066773b379b76cc66  admin-bootstrap-openbao-autonomy-v1-v3.sh
+b400b9427cd37cb7bbba85313081aeb93f0f3181cbfbab99948c793f30d3c53e  admin-bootstrap-openbao-autonomy-v1-v3.sha256
+c355df01c83c2801ae3a0df43806c562ee70f97c9aa558e3f67dd097efa5ac99  test-admin-bootstrap-openbao-autonomy-v1-v3.sh

deploy/openbao-autonomy/test-admin-bootstrap-openbao-autonomy-v1-v3.sh

@@ -0,0 +1,39 @@
+#!/usr/bin/env bash
+set -Eeuo pipefail
+umask 077
+
+script="${1:-$(CDPATH= cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)/admin-bootstrap-openbao-autonomy-v1-v3.sh}"
+report="${2:-$(CDPATH= cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd -P)/admin-bootstrap-openbao-autonomy-v1-v3.self-test.json}"
+
+[[ -f "$script" ]] || {
+  printf 'missing script: %s\n' "$script" >&2
+  exit 66
+}
+
+bash -n "$script"
+tmp="$(mktemp)"
+trap 'rm -f "$tmp"' EXIT
+bash "$script" --self-test >"$tmp"
+
+python3 - "$script" "$tmp" <<'PY'
+import hashlib
+import json
+import pathlib
+import sys
+
+script, report = map(pathlib.Path, sys.argv[1:])
+data = json.loads(report.read_text())
+actual_hash = hashlib.sha256(script.read_bytes()).hexdigest()
+
+assert data["schema"] == "AIOS_OPENBAO_BOOTSTRAP_STATIC_SELF_TEST_V3"
+assert data["scope"] == "static_and_fixture_only"
+assert data["runtime_claims"] is False
+assert data["script_sha256"] == actual_hash
+assert data["passed"] == data["total"]
+assert all(data["checks"].values())
+PY
+
+install -m 0600 "$tmp" "$report"
+printf 'PASS bash-n + embedded static/fixture tests\n'
+printf 'report=%s\n' "$report"
+sha256sum "$script"

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

@@ -1,40 +0,0 @@
-[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/gitops/isolated-writer-path-policy-v1.md

@@ -0,0 +1,68 @@
+# Isolated Writer Exact-Path Policy V1
+
+This policy is an additional, service-owned authorization layer for
+`POST /api/commit-file`. It never expands a Token Policy V2 grant.
+
+An atomic write is permitted only when all of the following agree:
+
+1. socket peer, audience, operation, branch, base and exact path in the
+   Token Policy V2 grant;
+2. service audience, repository identity, branch, operation and exact path
+   in `AIOS_GITOPS_PATH_POLICY_FILE`;
+3. lexical path, resolved containment and symlink checks;
+4. the configured content-size and content-profile checks;
+5. operation digest, expected branch HEAD and durable operation ledger.
+
+Any failed layer returns a fail-closed policy error. Coarse `/api/file` and
+`/api/commit` routes remain forbidden.
+
+## File contract
+
+The configured path must be absolute. The parent directory must be
+`root:gitops 0750`; the regular, non-symlink file must be
+`root:gitops 0640`. JSON must use UTF-8 canonical encoding with sorted keys,
+compact separators and one trailing newline. Unknown fields, unsupported
+operations or profiles, duplicate paths, oversized policies and
+group/world-writable files are rejected during Writer startup.
+
+Example:
+
+```json
+{"allowed_branch":"patch/codex-review-test-worker-20260724","allowed_operations":{"commit_file.atomic":{"content_profile":"systemd-codex-review-worker-v1","exact_paths":["deploy/systemd/aios-codex-review-test-worker.service"],"max_content_bytes":32768}},"audience":"aios-codex-review-worker-writer.service","policy_id":"codex-review-worker-path-policy-v1","repository_identity":"aios/aios","schema_version":1}
+```
+
+The SHA-256 of the canonical policy is exposed in the isolated runtime
+status and written to the operation audit record. The request body cannot
+select or override the policy file.
+
+## Content profile
+
+`systemd-codex-review-worker-v1` is an exact, fail-closed unit grammar. The
+only sections are `[Unit]`, `[Service]` and `[Install]`. Each section has a
+directive allowlist; every single-value directive and every repeated
+`Environment`, `ReadOnlyPaths` and `ReadWritePaths` entry must match the
+governed Shadow unit contract exactly.
+
+The profile fixes the Python module
+`task_plane.workers.codex_review_test`, `User=aiman`, `Group=gitops`,
+the controlled `/usr/libexec/aios-codex-review-test-worker/current`
+working directory, all hardening switches and the three permitted
+environment variables. It grants read access only to the authoritative
+bare repository and the versioned worker artifact, and write access only
+to the worker runtime, state and cache directories.
+
+Unknown sections or directives are rejected, including lifecycle hooks,
+credential injection, environment imports, root images, bind mounts,
+capability expansion, devices, sockets and top-level writable paths. The
+parser also rejects duplicate sections, duplicate single-value directives,
+empty resets, continuation lines, CRLF, NUL, non-UTF-8 surrogate content,
+overlong lines and Unicode lookalike directive names. It does not use
+`ConfigParser(strict=False)` or silently merge repeated values.
+
+Production credentials, if later authorized, must be supplied by a
+separately governed deployment drop-in. They are not permitted in the
+source unit or this content profile.
+
+Policy changes require a separate maintenance branch and Owner-reviewed
+ordinary merge commit. A task-scoped Writer must not edit its own policy
+file.

docs/gitops/openapi-chatgpt-actions.yaml

@@ -51,6 +51,31 @@ components:
         path: {type: string, example: docs/gitops/chatgpt-change.md}
         content: {type: string}
         message: {type: string}
+    AtomicFileCommitRequest:
+      type: object
+      additionalProperties: false
+      required: [branch, path, content, message, expected_head_oid, idempotency_key]
+      properties:
+        branch:
+          type: string
+          pattern: '^patch/[A-Za-z0-9._/-]+$'
+        path:
+          type: string
+          minLength: 1
+          maxLength: 512
+        content:
+          type: string
+          maxLength: 1048576
+        message:
+          type: string
+          minLength: 1
+          maxLength: 200
+        expected_head_oid:
+          type: string
+          pattern: '^[0-9a-f]{40}$'
+        idempotency_key:
+          type: string
+          pattern: '^[A-Za-z0-9][A-Za-z0-9._:-]{7,127}$'
     CommitRequest:
       type: object
       properties:
@@ -428,6 +453,25 @@ paths:
             schema: {$ref: '#/components/schemas/FileCommitRequest'}
       responses: {'200': {description: File committed}, '400': {description: Secret rejected or invalid path}, '401': {description: Unauthorized}}
       x-openai-isConsequential: true
+  /api/commit-file:
+    post:
+      operationId: commitFileAtomic
+      summary: Atomically write one allowlisted file, commit, push, and verify using Token Policy V2
+      security: [{bearerAuth: []}]
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema: {$ref: '#/components/schemas/AtomicFileCommitRequest'}
+      responses:
+        '200': {description: Completed or idempotently replayed}
+        '400': {description: Invalid canonical request}
+        '401': {description: Missing, invalid, expired, or revoked Token Policy V2 grant}
+        '403': {description: Audience, source, branch, operation, path, or policy denied}
+        '409': {description: Idempotency conflict or manual review required}
+        '413': {description: Request too large}
+        '503': {description: Push or remote verification unavailable; recoverable}
+      x-openai-isConsequential: true
   /api/commit:
     post:
       operationId: commitChanges

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

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

docs/task-plane/CODEX_REVIEW_TEST_WORKER.md

@@ -1,36 +0,0 @@
-# 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

@@ -1,20 +0,0 @@
-{
-  "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

@@ -1,138 +0,0 @@
-#!/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

@@ -1,159 +0,0 @@
-#!/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

@@ -1,228 +0,0 @@
-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

@@ -1,90 +0,0 @@
-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

@@ -1,83 +0,0 @@
-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"})

tests/test_isolated_writer_exact_path_policy.py

@@ -0,0 +1,417 @@
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest import mock
+
+import aios_isolated_writer_policy as writer_policy
+import isolated_writer_path_policy as path_policy
+from isolated_writer_authz import AuthorizationError, _validate_grant
+
+
+TARGET = "deploy/systemd/aios-codex-review-test-worker.service"
+BRANCH = "patch/codex-review-test-worker-20260724"
+AUDIENCE = "aios-codex-review-worker-writer.service"
+
+VALID_UNIT = """\
+[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
+"""
+
+
+def document(*, paths=None, **updates):
+    value = {
+        "schema_version": 1,
+        "policy_id": "codex-review-worker-path-policy-v1",
+        "audience": AUDIENCE,
+        "repository_identity": "aios/aios",
+        "allowed_branch": BRANCH,
+        "allowed_operations": {
+            "commit_file.atomic": {
+                "exact_paths": paths or [TARGET],
+                "max_content_bytes": 32768,
+                "content_profile": "systemd-codex-review-worker-v1",
+            }
+        },
+    }
+    value.update(updates)
+    return value
+
+
+class ExactPathPolicyTests(unittest.TestCase):
+    def setUp(self):
+        self.saved = dict(os.environ)
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.policy_dir = self.root / "policies"
+        self.policy_dir.mkdir(mode=0o750)
+        self.policy_file = self.policy_dir / "writer.json"
+        self.workspace = self.root / "workspace"
+        self.workspace.mkdir(mode=0o700)
+        os.environ.update(
+            {
+                "AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT": "required",
+                "AIOS_GITOPS_SERVICE_AUDIENCE": AUDIENCE,
+                "AIOS_GITOPS_REPOSITORY_IDENTITY": "aios/aios",
+                "AIOS_GITOPS_ALLOWED_BRANCH": BRANCH,
+            }
+        )
+
+    def tearDown(self):
+        path_policy.clear_path_policy_cache()
+        os.environ.clear()
+        os.environ.update(self.saved)
+        self.temp.cleanup()
+
+    def write_policy(self, value=None):
+        payload = document() if value is None else value
+        raw = (
+            json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+            + "\n"
+        )
+        self.policy_file.write_text(raw, encoding="utf-8")
+        self.policy_file.chmod(0o640)
+        return path_policy.load_path_policy(
+            self.policy_file,
+            expected_owner_uid=os.getuid(),
+            expected_group_gid=os.getgid(),
+        )
+
+    def configured(self, loaded):
+        return mock.patch.object(
+            path_policy,
+            "configured_path_policy",
+            return_value=loaded,
+        )
+
+    def test_exact_systemd_path_and_exact_token_path_are_allowed(self):
+        loaded = self.write_policy()
+        with self.configured(loaded):
+            self.assertEqual(TARGET, writer_policy.validate_path(TARGET, self.workspace))
+            writer_policy.validate_content_for_path(TARGET, VALID_UNIT)
+
+    def test_token_path_cannot_expand_service_policy(self):
+        loaded = self.write_policy(document(paths=["deploy/systemd/other.service"]))
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_service_path_rejected"
+        ):
+            writer_policy.validate_path(TARGET, self.workspace)
+
+    def test_service_policy_cannot_expand_token_path(self):
+        now = datetime.now(timezone.utc)
+        grant = {
+            "schema_version": 2,
+            "audience": AUDIENCE,
+            "source_cidrs": ["127.0.0.1/32"],
+            "created_at": (now - timedelta(seconds=5)).isoformat(),
+            "not_before": (now - timedelta(seconds=2)).isoformat(),
+            "expires_at": (now + timedelta(minutes=5)).isoformat(),
+            "revoked": False,
+            "allowed_operations": ["commit_file.atomic"],
+            "allowed_branch": BRANCH,
+            "allowed_paths": ["deploy/systemd/other.service"],
+            "approved_base_oid": "a" * 40,
+        }
+        with mock.patch(
+            "isolated_writer_authz._git_is_ancestor", return_value=True
+        ), self.assertRaisesRegex(AuthorizationError, "writer_token_path_rejected"):
+            _validate_grant(
+                grant,
+                audience=AUDIENCE,
+                source_ip="127.0.0.1",
+                operation="commit_file.atomic",
+                branch=BRANCH,
+                path=TARGET,
+                workspace=self.workspace,
+                current_time=now,
+            )
+
+    def test_other_service_file_is_rejected(self):
+        loaded = self.write_policy()
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_service_path_rejected"
+        ):
+            writer_policy.validate_path(
+                "deploy/systemd/other.service", self.workspace
+            )
+
+    def test_path_traversal_is_rejected(self):
+        for value in ("deploy/systemd/../x", "../deploy/systemd/x"):
+            with self.subTest(value=value), self.assertRaises(PermissionError):
+                writer_policy.validate_path(value, self.workspace)
+
+    def test_url_encoded_traversal_is_rejected(self):
+        with self.assertRaises(PermissionError):
+            writer_policy.validate_path(
+                "deploy/systemd/%2e%2e/x", self.workspace
+            )
+
+    def test_symlink_policy_file_is_rejected(self):
+        self.write_policy()
+        link = self.policy_dir / "link.json"
+        link.symlink_to(self.policy_file)
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError, "isolated_writer_path_policy_symlink_rejected"
+        ):
+            path_policy.load_path_policy(
+                link,
+                expected_owner_uid=os.getuid(),
+                expected_group_gid=os.getgid(),
+            )
+
+    def test_world_writable_policy_file_is_rejected(self):
+        self.write_policy()
+        self.policy_file.chmod(0o646)
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_permissions_rejected",
+        ):
+            path_policy.load_path_policy(
+                self.policy_file,
+                expected_owner_uid=os.getuid(),
+                expected_group_gid=os.getgid(),
+            )
+
+    def test_unknown_policy_field_is_rejected(self):
+        value = document()
+        value["unexpected"] = True
+        with self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_fields_rejected",
+        ):
+            self.write_policy(value)
+
+    def test_audience_mismatch_is_rejected(self):
+        loaded = self.write_policy()
+        os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
+        os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "other.service"
+        path_policy.clear_path_policy_cache()
+        with mock.patch.object(
+            path_policy, "load_path_policy", return_value=loaded
+        ), self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_audience_rejected",
+        ):
+            path_policy.configured_path_policy()
+
+    def test_branch_mismatch_is_rejected(self):
+        loaded = self.write_policy()
+        os.environ["AIOS_GITOPS_PATH_POLICY_FILE"] = str(self.policy_file)
+        os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/other"
+        path_policy.clear_path_policy_cache()
+        with mock.patch.object(
+            path_policy, "load_path_policy", return_value=loaded
+        ), self.assertRaisesRegex(
+            path_policy.PathPolicyError,
+            "isolated_writer_path_policy_branch_rejected",
+        ):
+            path_policy.configured_path_policy()
+
+    def assert_profile_rejected(self, content):
+        loaded = self.write_policy()
+        with self.configured(loaded), self.assertRaisesRegex(
+            PermissionError, "isolated_writer_systemd_profile_rejected"
+        ):
+            writer_policy.validate_content_for_path(TARGET, content)
+
+    def test_disallowed_execution_and_credential_directives_are_rejected(self):
+        directives = (
+            "ExecStartPre=/usr/bin/id",
+            "ExecStartPost=/usr/bin/id",
+            "ExecReload=/usr/bin/id",
+            "ExecStop=/usr/bin/id",
+            "ExecStopPost=/usr/bin/id",
+            "ExecCondition=/usr/bin/true",
+            "EnvironmentFile=/run/worker.env",
+            "PassEnvironment=PATH",
+            "LoadCredential=worker:/run/worker",
+            "SetCredential=worker:value",
+            "ImportCredential=worker",
+            "StandardInput=socket",
+            "StandardOutput=socket",
+        )
+        for directive in directives:
+            with self.subTest(directive=directive):
+                self.assert_profile_rejected(
+                    VALID_UNIT.replace(
+                        "Restart=on-failure",
+                        f"Restart=on-failure\n{directive}",
+                    )
+                )
+
+    def test_disallowed_namespace_and_capability_directives_are_rejected(self):
+        directives = (
+            "RootDirectory=/srv/worker",
+            "RootImage=/srv/worker.raw",
+            "ReadWritePaths=/",
+            "BindPaths=/etc:/etc",
+            "BindReadOnlyPaths=/etc:/etc",
+            "TemporaryFileSystem=/etc",
+            "CapabilityBoundingSet=CAP_SYS_ADMIN",
+            "AmbientCapabilities=CAP_SYS_ADMIN",
+            "DeviceAllow=/dev/sda rw",
+            "SupplementaryGroups=docker",
+            "Sockets=unsafe.socket",
+        )
+        for directive in directives:
+            with self.subTest(directive=directive):
+                self.assert_profile_rejected(
+                    VALID_UNIT.replace(
+                        "Restart=on-failure",
+                        f"Restart=on-failure\n{directive}",
+                    )
+                )
+
+    def test_duplicate_single_directives_and_sections_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace("User=aiman", "User=aiman\nUser=aiman"),
+            VALID_UNIT.replace(
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test\n"
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
+            ),
+            VALID_UNIT + "\n[Service]\nType=simple\n",
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_empty_reset_continuation_crlf_nul_and_long_lines_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace(
+                "ExecStart=/usr/bin/python3 -m task_plane.workers.codex_review_test",
+                "ExecStart=",
+            ),
+            VALID_UNIT.replace(
+                "Restart=on-failure",
+                "Restart=on-failure\n continuation-injection",
+            ),
+            VALID_UNIT.replace("\n", "\r\n"),
+            VALID_UNIT.replace("Restart=on-failure", "Restart=on-failure\x00"),
+            VALID_UNIT.replace(
+                "Description=AIOS Codex Review/Test Worker",
+                "Description=" + ("A" * 4097),
+            ),
+        )
+        for value in bad_values:
+            with self.subTest(kind=repr(value[:80])):
+                self.assert_profile_rejected(value)
+
+    def test_unknown_environment_and_production_execution_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace(
+                "Environment=PYTHONDONTWRITEBYTECODE=1",
+                "Environment=UNKNOWN_KEY=value",
+            ),
+            VALID_UNIT.replace(
+                "AIOS_TASK_PLANE_EXECUTION_ENABLED=false",
+                "AIOS_TASK_PLANE_EXECUTION_ENABLED=true",
+            ),
+            VALID_UNIT.replace(
+                "Environment=PYTHONDONTWRITEBYTECODE=1",
+                "Environment=Authorization: Bearer placeholder",
+            ),
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_identity_working_directory_and_path_prefix_bypasses_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace("User=aiman", "User=root"),
+            VALID_UNIT.replace(
+                "/usr/libexec/aios-codex-review-test-worker/current",
+                "/data/aios-gitops/worktrees/worker",
+            ),
+            VALID_UNIT.replace(
+                "ReadWritePaths=/run/aios-codex-review-test-worker",
+                "ReadWritePaths=/etc",
+            ),
+            VALID_UNIT.replace(
+                "ReadOnlyPaths=/data/aios-gitops/repo.git",
+                "ReadOnlyPaths=/data",
+            ),
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_unknown_section_directive_and_unicode_lookalike_are_rejected(self):
+        bad_values = (
+            VALID_UNIT.replace("[Install]", "[Socket]"),
+            VALID_UNIT.replace(
+                "Restart=on-failure",
+                "Restart=on-failure\nProtectClock=true",
+            ),
+            VALID_UNIT.replace("ExecStart=", "\u0415xecStart="),
+        )
+        for value in bad_values:
+            with self.subTest(value=value):
+                self.assert_profile_rejected(value)
+
+    def test_invalid_unicode_surrogate_and_missing_final_newline_are_rejected(self):
+        for value in (VALID_UNIT + "\ud800", VALID_UNIT.rstrip("\n")):
+            with self.subTest(value=repr(value[-30:])):
+                self.assert_profile_rejected(value)
+
+    def test_writer_request_cannot_modify_policy_file(self):
+        with self.assertRaises(PermissionError):
+            writer_policy.validate_path(
+                "etc/aios-gitops/writer-path-policies/codex-review-worker.json",
+                self.workspace,
+            )
+
+    def test_coarse_file_and_commit_endpoints_remain_forbidden(self):
+        for endpoint in ("/api/file", "/api/commit"):
+            with self.subTest(endpoint=endpoint), self.assertRaisesRegex(
+                PermissionError, "isolated_writer_operation_forbidden"
+            ):
+                writer_policy.validate_endpoint(
+                    "POST", endpoint, self.workspace, self.root
+                )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_isolated_writer_policy.py

@@ -15,6 +15,10 @@ class IsolatedWriterPolicyTests(unittest.TestCase):
         os.environ["AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT"] = "required"
         os.environ["AIOS_GITOPS_ALLOWED_BRANCH"] = "patch/aios-agent-task-control-plane-20260724"
         os.environ["AIOS_GITOPS_ALLOWED_BASE_OID"] = "a8e97868976b0dce17b7cb8487eadc1d9c687622"
+        os.environ["AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED"] = "2"
+        os.environ["AIOS_GITOPS_SERVICE_AUDIENCE"] = "test-isolated-writer.service"
+        os.environ["AIOS_GITOPS_TOKEN_REGISTRY_V2"] = "/tmp/test-writer-token-registry.sqlite3"
+        os.environ["AIOS_GITOPS_WRITER_OPERATION_LEDGER"] = "/tmp/test-writer-operations.sqlite3"
 
     def tearDown(self):
         os.environ.clear()

tests/test_isolated_writer_token_policy_v2.py

@@ -0,0 +1,471 @@
+import hashlib
+import json
+import os
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import threading
+import time
+import unittest
+from dataclasses import dataclass
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "active-runtime" / "app"))
+
+import aios_isolated_writer_policy as policy
+from isolated_writer_authz import AuthorizationError, authorize, operation_for
+from isolated_writer_commit_file import (
+    AtomicCommitFile,
+    CommitFileError,
+    OperationLedger,
+    SimulatedCrash,
+)
+from isolated_writer_token_registry import TokenRegistry, token_digest
+
+
+BRANCH = "patch/codex-review-test-worker-20260724"
+AUDIENCE = "aios-codex-review-worker-writer.service"
+TASK_ID = "AIOS-CODEX-WRITER-SMOKE-20260724"
+TARGET_PATH = "docs/reliability/codex-review-worker-writer-smoke.json"
+
+
+def run(cwd, *args):
+    return subprocess.check_output(args, cwd=cwd, text=True).strip()
+
+
+@dataclass
+class DummyGrant:
+    token_id: str = "test-token-id"
+    task_id: str = TASK_ID
+    operation: str = "commit_file.atomic"
+    record: dict = None
+
+    def __post_init__(self):
+        if self.record is None:
+            self.record = {"audience": AUDIENCE, "_source_ip": "127.0.0.1"}
+
+
+class RepositoryCase(unittest.TestCase):
+    def setUp(self):
+        self.saved = dict(os.environ)
+        self.temp = tempfile.TemporaryDirectory()
+        self.root = Path(self.temp.name)
+        self.authoritative = self.root / "authoritative.git"
+        self.gitea = self.root / "gitea.git"
+        self.seed = self.root / "seed"
+        self.workspace = self.root / "writer"
+        subprocess.run(["git", "init", "--bare", str(self.authoritative)], check=True, capture_output=True)
+        subprocess.run(["git", "init", str(self.seed)], check=True, capture_output=True)
+        subprocess.run(["git", "config", "user.email", "writer@example.invalid"], cwd=self.seed, check=True)
+        subprocess.run(["git", "config", "user.name", "Writer Test"], cwd=self.seed, check=True)
+        (self.seed / "README.md").write_text("base\n", encoding="utf-8")
+        subprocess.run(["git", "add", "README.md"], cwd=self.seed, check=True)
+        subprocess.run(["git", "commit", "-m", "base"], cwd=self.seed, check=True, capture_output=True)
+        subprocess.run(["git", "branch", "-M", "main"], cwd=self.seed, check=True)
+        subprocess.run(["git", "remote", "add", "origin", str(self.authoritative)], cwd=self.seed, check=True)
+        subprocess.run(["git", "push", "origin", "main"], cwd=self.seed, check=True, capture_output=True)
+        self.base = run(self.seed, "git", "rev-parse", "HEAD")
+        subprocess.run(["git", "checkout", "-b", BRANCH], cwd=self.seed, check=True, capture_output=True)
+        subprocess.run(["git", "push", "origin", BRANCH], cwd=self.seed, check=True, capture_output=True)
+        subprocess.run(
+            ["git", "--git-dir", str(self.authoritative), "worktree", "add", str(self.workspace), BRANCH],
+            check=True,
+            capture_output=True,
+        )
+        subprocess.run(["git", "config", "user.email", "writer@example.invalid"], cwd=self.workspace, check=True)
+        subprocess.run(["git", "config", "user.name", "Writer Test"], cwd=self.workspace, check=True)
+        subprocess.run(["git", "clone", "--bare", str(self.authoritative), str(self.gitea)], check=True, capture_output=True)
+        subprocess.run(["git", "remote", "add", "gitea-http", str(self.gitea)], cwd=self.workspace, check=True)
+        self.registry = TokenRegistry(self.root / "registry.sqlite3")
+        os.environ.update({
+            "AIOS_GITOPS_ALLOWED_BRANCH": BRANCH,
+            "AIOS_GITOPS_TOKEN_POLICY_VERSION_REQUIRED": "2",
+            "AIOS_GITOPS_SERVICE_AUDIENCE": AUDIENCE,
+            "AIOS_GITOPS_ISOLATED_WRITER_ENFORCEMENT": "required",
+        })
+
+    def tearDown(self):
+        os.environ.clear()
+        os.environ.update(self.saved)
+        self.temp.cleanup()
+
+    def grant(self, **updates):
+        now = datetime.now(timezone.utc)
+        grant = {
+            "token_id": "smoke-" + hashlib.sha256(os.urandom(16)).hexdigest()[:12],
+            "audience": AUDIENCE,
+            "task_id": TASK_ID,
+            "source_cidrs": ["127.0.0.1/32", "::1/128"],
+            "allowed_branch": BRANCH,
+            "approved_base_oid": self.base,
+            "allowed_operations": [
+                "runtime.status.read",
+                "baseline.read",
+                "patches.read",
+                "commit_file.atomic",
+            ],
+            "allowed_paths": [TARGET_PATH],
+            "not_before": (now - timedelta(seconds=2)).isoformat(),
+            "expires_at": (now + timedelta(minutes=20)).isoformat(),
+            "max_uses": {
+                "runtime.status.read": 2,
+                "baseline.read": 2,
+                "patches.read": 2,
+                "commit_file.atomic": 4,
+            },
+            "issued_by": "owner-authorized-local-maintenance-executor",
+            "created_at": (now - timedelta(seconds=3)).isoformat(),
+        }
+        grant.update(updates)
+        return grant
+
+    def issue(self, **updates):
+        return self.registry.issue(self.grant(**updates))
+
+    def auth(self, token, **updates):
+        args = {
+            "method": "GET",
+            "path": "/api/runtime-status",
+            "source_ip": "127.0.0.1",
+            "workspace": self.workspace,
+        }
+        args.update(updates)
+        return authorize(self.registry, token, **args)
+
+    def request(self, key="writer-smoke-0001", content='{"status":"READY"}\n'):
+        return {
+            "branch": BRANCH,
+            "path": TARGET_PATH,
+            "content": content,
+            "message": "chore(task-plane): verify Codex writer endpoint",
+            "expected_head_oid": run(self.workspace, "git", "rev-parse", "HEAD"),
+            "idempotency_key": key,
+        }
+
+    def writer(self, crash=None, remote="gitea-http", audit=None):
+        return AtomicCommitFile(
+            workspace=self.workspace,
+            bare_repo=self.authoritative,
+            ledger=OperationLedger(self.root / "operations.sqlite3"),
+            remote=remote,
+            audit=audit or (lambda record: None),
+            crash_after_state=crash,
+        )
+
+
+class TokenPolicyV2Tests(RepositoryCase):
+    def test_v1_coarse_scope_token_is_rejected(self):
+        token, record = self.issue()
+        record["schema_version"] = 1
+        with self.registry._connect() as db:
+            db.execute(
+                "UPDATE token_grants SET grant_json=? WHERE token_id=?",
+                (json.dumps(record), record["token_id"]),
+            )
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_policy_v2_required"):
+            self.auth(token)
+
+    def test_audience_mismatch_is_rejected(self):
+        token, _ = self.issue(audience="other.service")
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_audience_rejected"):
+            self.auth(token)
+
+    def test_source_ip_mismatch_is_rejected(self):
+        token, _ = self.issue(source_cidrs=["192.0.2.0/24"])
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_source_rejected"):
+            self.auth(token)
+
+    def test_branch_mismatch_is_rejected(self):
+        token, _ = self.issue(allowed_branch="patch/other")
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_branch_rejected"):
+            self.auth(token)
+
+    def test_operation_mismatch_is_rejected(self):
+        token, _ = self.issue(allowed_operations=["baseline.read"])
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_operation_rejected"):
+            self.auth(token)
+
+    def test_path_mismatch_is_rejected(self):
+        token, _ = self.issue()
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_path_rejected"):
+            self.auth(
+                token,
+                method="POST",
+                path="/api/commit-file",
+                branch=BRANCH,
+                target_path="docs/other.json",
+            )
+
+    def test_expired_token_is_rejected(self):
+        now = datetime.now(timezone.utc)
+        token, _ = self.issue(
+            created_at=(now - timedelta(minutes=10)).isoformat(),
+            not_before=(now - timedelta(minutes=9)).isoformat(),
+            expires_at=(now - timedelta(seconds=1)).isoformat(),
+        )
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_expired"):
+            self.auth(token)
+
+    def test_not_before_token_is_rejected(self):
+        now = datetime.now(timezone.utc)
+        token, _ = self.issue(
+            created_at=now.isoformat(),
+            not_before=(now + timedelta(minutes=1)).isoformat(),
+            expires_at=(now + timedelta(minutes=5)).isoformat(),
+        )
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_not_yet_valid"):
+            self.auth(token)
+
+    def test_revoked_token_is_rejected(self):
+        token, grant = self.issue()
+        self.assertTrue(self.registry.revoke(grant["token_id"]))
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_revoked"):
+            self.auth(token)
+
+    def test_operation_use_count_is_atomic_and_exhausted(self):
+        token, _ = self.issue(max_uses={
+            "runtime.status.read": 1,
+            "baseline.read": 1,
+            "patches.read": 1,
+            "commit_file.atomic": 1,
+        })
+        self.auth(token)
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_operation_exhausted"):
+            self.auth(token)
+
+    def test_concurrent_use_count_never_exceeds_limit(self):
+        token, _ = self.issue(max_uses={
+            "runtime.status.read": 2,
+            "baseline.read": 1,
+            "patches.read": 1,
+            "commit_file.atomic": 1,
+        })
+        successes, failures = [], []
+
+        def invoke():
+            try:
+                successes.append(self.auth(token))
+            except AuthorizationError as exc:
+                failures.append(exc.code)
+
+        threads = [threading.Thread(target=invoke) for _ in range(6)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertEqual(2, len(successes))
+        self.assertEqual(4, failures.count("writer_token_operation_exhausted"))
+
+    def test_registry_does_not_store_plaintext_token(self):
+        token, _ = self.issue()
+        registry_bytes = b"".join(
+            item.read_bytes()
+            for item in self.registry.path.parent.glob(self.registry.path.name + "*")
+            if item.is_file()
+        )
+        self.assertNotIn(token.encode(), registry_bytes)
+        with self.registry._connect() as db:
+            stored = db.execute("SELECT token_digest FROM token_grants").fetchone()[0]
+        self.assertEqual(token_digest(token), stored)
+
+    def test_coarse_routes_have_no_operation_mapping(self):
+        for path in ("/api/file", "/api/commit", "/api/tag", "/api/rollback", "/api/model"):
+            with self.subTest(path=path), self.assertRaises(AuthorizationError) as captured:
+                operation_for("POST", path)
+            self.assertEqual(403, captured.exception.status)
+
+    def test_route_mapping_is_fixed_not_request_controlled(self):
+        self.assertEqual("runtime.status.read", operation_for("GET", "/api/runtime-status"))
+        self.assertEqual("baseline.read", operation_for("GET", "/api/baseline"))
+        self.assertEqual("patches.read", operation_for("GET", "/api/patches"))
+        self.assertEqual("commit_file.atomic", operation_for("POST", "/api/commit-file"))
+
+    def test_x_forwarded_for_cannot_replace_socket_peer(self):
+        token, _ = self.issue(source_cidrs=["203.0.113.9/32"])
+        with self.assertRaisesRegex(AuthorizationError, "writer_token_source_rejected"):
+            self.auth(token, source_ip="127.0.0.1")
+
+
+class AtomicCommitFileTests(RepositoryCase):
+    def test_commit_file_succeeds_and_verifies_bare_and_gitea(self):
+        result = self.writer().execute(DummyGrant(), self.request())
+        self.assertEqual("COMPLETED", result["state"])
+        self.assertEqual(
+            result["commit_oid"],
+            run(self.root, "git", "--git-dir", str(self.gitea), "rev-parse", f"refs/heads/{BRANCH}"),
+        )
+
+    def test_same_idempotency_key_returns_same_commit(self):
+        request = self.request()
+        first = self.writer().execute(DummyGrant(), request)
+        count = run(self.workspace, "git", "rev-list", "--count", "HEAD")
+        second = self.writer().execute(DummyGrant(), request)
+        self.assertEqual(first["commit_oid"], second["commit_oid"])
+        self.assertEqual(count, run(self.workspace, "git", "rev-list", "--count", "HEAD"))
+        self.assertTrue(second["idempotent"])
+
+    def test_same_key_with_different_payload_returns_409(self):
+        request = self.request()
+        self.writer().execute(DummyGrant(), request)
+        request["content"] = '{"status":"DIFFERENT"}\n'
+        with self.assertRaises(CommitFileError) as captured:
+            self.writer().execute(DummyGrant(), request)
+        self.assertEqual(409, captured.exception.status)
+
+    def test_same_key_from_different_task_returns_409(self):
+        request = self.request()
+        self.writer().execute(DummyGrant(), request)
+        other = DummyGrant(token_id="other-token", task_id="OTHER-TASK")
+        with self.assertRaisesRegex(CommitFileError, "writer_idempotency_owner_conflict"):
+            self.writer().execute(other, request)
+
+    def test_file_written_crash_is_recoverable(self):
+        request = self.request()
+        with self.assertRaises(SimulatedCrash):
+            self.writer(crash="FILE_WRITTEN").execute(DummyGrant(), request)
+        result = self.writer().execute(DummyGrant(), request)
+        self.assertEqual("COMPLETED", result["state"])
+
+    def test_committed_crash_is_recoverable_without_second_commit(self):
+        request = self.request()
+        with self.assertRaises(SimulatedCrash):
+            self.writer(crash="COMMITTED").execute(DummyGrant(), request)
+        committed = run(self.workspace, "git", "rev-parse", "HEAD")
+        result = self.writer().execute(DummyGrant(), request)
+        self.assertEqual(committed, result["commit_oid"])
+
+    def test_pushed_crash_only_reverifies(self):
+        request = self.request()
+        with self.assertRaises(SimulatedCrash):
+            self.writer(crash="PUSHED").execute(DummyGrant(), request)
+        result = self.writer().execute(DummyGrant(), request)
+        self.assertEqual("COMPLETED", result["state"])
+
+    def test_push_failure_is_failed_recoverable(self):
+        request = self.request()
+        with self.assertRaisesRegex(CommitFileError, "writer_gitea_push_failed"):
+            self.writer(remote="missing-remote").execute(DummyGrant(), request)
+        row = OperationLedger(self.root / "operations.sqlite3").get(request["idempotency_key"])
+        self.assertEqual("FAILED_RECOVERABLE", row["state"])
+
+    def test_branch_head_conflict_requires_manual_review(self):
+        request = self.request()
+        ledger = OperationLedger(self.root / "operations.sqlite3")
+        ledger.get_or_create(DummyGrant(), request, __import__("isolated_writer_commit_file").canonical_digest(request))
+        (self.workspace / "README.md").write_text("drift\n", encoding="utf-8")
+        subprocess.run(["git", "commit", "-am", "drift"], cwd=self.workspace, check=True, capture_output=True)
+        with self.assertRaisesRegex(CommitFileError, "writer_branch_head_conflict"):
+            self.writer().execute(DummyGrant(), request)
+        self.assertEqual("MANUAL_REVIEW_REQUIRED", ledger.get(request["idempotency_key"])["state"])
+
+    def test_extra_staged_file_is_rejected(self):
+        (self.workspace / "README.md").write_text("staged\n", encoding="utf-8")
+        subprocess.run(["git", "add", "README.md"], cwd=self.workspace, check=True)
+        with self.assertRaisesRegex(CommitFileError, "writer_extra_staged_files_rejected"):
+            self.writer().execute(DummyGrant(), self.request())
+
+    def test_path_traversal_and_protected_roots_are_rejected(self):
+        for path in (
+            "../docs/x", "%2e%2e/docs/x", "/etc/x", r"C:\temp\x",
+            r"\\server\share\x", "baseline/current.json", "runtime/x",
+        ):
+            request = self.request(key="writer-path-" + hashlib.sha256(path.encode()).hexdigest()[:12])
+            request["path"] = path
+            with self.subTest(path=path), self.assertRaises((PermissionError, CommitFileError)):
+                self.writer().execute(DummyGrant(), request)
+
+    def test_audit_contains_ids_but_not_content_or_authorization(self):
+        records = []
+        request = self.request(content='{"value":"non-secret-body-marker"}\n')
+        self.writer(audit=records.append).execute(DummyGrant(), request)
+        encoded = json.dumps(records)
+        self.assertIn("test-token-id", encoded)
+        self.assertNotIn("non-secret-body-marker", encoded)
+        self.assertNotIn("Authorization", encoded)
+
+    def test_concurrent_same_idempotency_creates_one_commit(self):
+        request = self.request()
+        results, errors = [], []
+
+        def invoke():
+            try:
+                results.append(self.writer().execute(DummyGrant(), request))
+            except Exception as exc:
+                errors.append(exc)
+
+        threads = [threading.Thread(target=invoke) for _ in range(4)]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertFalse(errors)
+        self.assertEqual(1, len({item["commit_oid"] for item in results}))
+
+    def test_different_operations_are_serialized_by_branch_lock(self):
+        active = 0
+        maximum = 0
+        guard = threading.Lock()
+
+        class Probe(AtomicCommitFile):
+            def _execute_locked(inner, grant, data, digest):
+                nonlocal active, maximum
+                with guard:
+                    active += 1
+                    maximum = max(maximum, active)
+                time.sleep(0.05)
+                with guard:
+                    active -= 1
+                return {"state": "PROBED"}
+
+        probe = Probe(
+            workspace=self.workspace,
+            bare_repo=self.authoritative,
+            ledger=OperationLedger(self.root / "probe.sqlite3"),
+            remote="gitea-http",
+            audit=lambda record: None,
+        )
+        requests = [self.request(key=f"writer-lock-{index:04d}") for index in range(3)]
+        threads = [threading.Thread(target=probe.execute, args=(DummyGrant(), item)) for item in requests]
+        for thread in threads:
+            thread.start()
+        for thread in threads:
+            thread.join()
+        self.assertEqual(1, maximum)
+
+    def test_invalid_expected_head_and_idempotency_are_rejected(self):
+        request = self.request()
+        request["expected_head_oid"] = "short"
+        with self.assertRaisesRegex(CommitFileError, "writer_expected_head_rejected"):
+            self.writer().execute(DummyGrant(), request)
+        request = self.request()
+        request["idempotency_key"] = "../bad"
+        with self.assertRaisesRegex(CommitFileError, "writer_idempotency_key_rejected"):
+            self.writer().execute(DummyGrant(), request)
+
+    def test_invalid_message_and_nul_content_are_rejected(self):
+        request = self.request()
+        request["message"] = "bad\nmessage"
+        with self.assertRaisesRegex(CommitFileError, "writer_message_rejected"):
+            self.writer().execute(DummyGrant(), request)
+        request = self.request()
+        request["content"] = "bad\x00content"
+        with self.assertRaisesRegex(CommitFileError, "writer_content_rejected"):
+            self.writer().execute(DummyGrant(), request)
+
+    def test_force_flags_are_403_and_unknown_fields_are_rejected(self):
+        for field in ("force", "force_push"):
+            request = self.request(key=f"writer-force-{field}")
+            request[field] = True
+            with self.subTest(field=field), self.assertRaises(CommitFileError) as captured:
+                self.writer().execute(DummyGrant(), request)
+            self.assertEqual(403, captured.exception.status)
+        request = self.request(key="writer-extra-field")
+        request["operation"] = "commit_file.atomic"
+        with self.assertRaisesRegex(CommitFileError, "writer_request_fields_rejected"):
+            self.writer().execute(DummyGrant(), request)
+
+
+if __name__ == "__main__":
+    unittest.main()