test(security): allow exact historical idempotency fixture

Reviewed-on: https://g.cnnetmkt.top/git/aios/aios/pulls/71
Reviewed-by: owner <owner@cnnetmkt.top>

.gitleaks.toml

@@ -0,0 +1,10 @@
+title = "AIOS repository secret scan policy"
+
+[extend]
+useDefault = true
+
+[allowlist]
+description = "Exact non-secret Task Admission test idempotency fixture in the historical PR commit"
+condition = "AND"
+paths = ['''tests/test_task_admission_bootstrap_http\.py$''']
+regexes = ['''idempotency-1''']

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -6,6 +6,12 @@ import hashlib
 from pathlib import Path
 from typing import Any, Callable, Mapping
 
+from aios_task_admission_bridge_contract import (
+    TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS,
+    TASK_ADMISSION_OPERATION_REQUIRED_FIELDS,
+    TASK_ADMISSION_ROUTES,
+)
+
 
 BRIDGE_SCHEMA_VERSION = "aios.chatgpt-gitops-bridge.v2"
 BRIDGE_OPERATIONS = (
@@ -29,15 +35,17 @@ BRIDGE_OPERATIONS = (
     "getControlPlaneSnapshot",
 )
 
+TASK_ADMISSION_OPERATIONS = tuple(TASK_ADMISSION_OPERATION_REQUIRED_FIELDS)
+
 OPERATION_REQUIRED_FIELDS = {
     "createGovernedTask": ("task_id", "repository", "base_oid", "target_branch", "risk", "description", "idempotency_key"),
     "projectOnlineAgentExecution": ("task_id", "required_capabilities"),
-    "acquireWriterLease": ("schema", "task_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "lease_seconds"),
-    "commitFileGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
-    "commitFilesGoverned": ("task_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
-    "getGitOpsJob": ("job_id",),
-    "resumeGitOpsJob": ("job_id", "expected_revision"),
-    "reconcileGitOpsJob": ("job_id", "expected_revision"),
+    "acquireWriterLease": ("schema", "task_id", "job_id", "repository", "branch_mode", "base_oid", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "lease_seconds"),
+    "commitFileGoverned": ("task_id", "job_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "path", "content_base64", "message"),
+    "commitFilesGoverned": ("task_id", "job_id", "repository", "expected_branch_oid", "target_branch", "generation", "idempotency_key", "files", "message"),
+    "getGitOpsJob": ("task_id", "job_id"),
+    "resumeGitOpsJob": ("task_id", "job_id", "expected_revision"),
+    "reconcileGitOpsJob": ("task_id", "job_id", "expected_revision"),
     "createTagApprovalRequest": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id", "owner_approval", "execute"),
     "requestOrCreateTag": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id", "execute"),
     "getTagPublicationStatus": ("version",),
@@ -46,6 +54,7 @@ OPERATION_REQUIRED_FIELDS = {
     "getReleaseConsistency": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id"),
     "runReleaseAcceptance": ("version", "target_oid", "evidence_closure_oid", "approval_plan_id"),
     "getControlPlaneSnapshot": (),
+    **TASK_ADMISSION_OPERATION_REQUIRED_FIELDS,
 }
 
 _OPERATION_PATHS = {
@@ -60,19 +69,48 @@ _OPERATION_PATHS = {
     "getTagPublicationStatus": "/v2/tag/status",
     "getReleaseConsistency": "/v2/release/consistency",
     "runReleaseAcceptance": "/v2/release/acceptance",
+    **{operation: path for path, operation in TASK_ADMISSION_ROUTES.items()},
 }
 
 
-def operation_contracts() -> dict[str, dict[str, object]]:
+def _has_complete_task_admission_handlers(
+    handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None,
+) -> bool:
+    return handlers is not None and all(
+        callable(handlers.get(operation)) for operation in TASK_ADMISSION_OPERATIONS
+    )
+
+
+def operation_contracts(
+    handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None,
+) -> dict[str, dict[str, object]]:
     """Return the runtime contract consumed by adapters and schema tooling."""
-    return {
+    contracts = {
         operation: {
             "method": "POST",
             "path": _OPERATION_PATHS.get(operation),
             "required_fields": list(fields),
         }
         for operation, fields in OPERATION_REQUIRED_FIELDS.items()
+        if operation not in TASK_ADMISSION_OPERATIONS
     }
+    if _has_complete_task_admission_handlers(handlers):
+        contracts.update(
+            {
+                operation: {
+                    "method": "GET" if operation == "getTaskAdmissionStatus" else "POST",
+                    "path": _OPERATION_PATHS[operation],
+                    "required_fields": list(
+                        TASK_ADMISSION_OPERATION_REQUIRED_FIELDS[operation]
+                    ),
+                    "optional_fields": list(
+                        TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS[operation]
+                    ),
+                }
+                for operation in TASK_ADMISSION_OPERATIONS
+            }
+        )
+    return contracts
 
 
 def render_openapi_contract_fragment() -> str:
@@ -85,6 +123,8 @@ def render_openapi_contract_fragment() -> str:
         "    post:\n"
         "      operationId: commitFilesGoverned\n"
         "      summary: Submit one atomic governed commit containing multiple files\n"
+        "      security: [{taskPrincipalAuth: []}]\n"
+        "      x-aios-task-principal: Bridge verifies a fresh aios.task-principal-assertion.v1 and exact task/job/repository/branch/generation/revision/attempt/execution-digest/capability binding; caller JSON cannot provide a principal and missing verification is denied.\n"
         "      requestBody:\n"
         "        required: true\n"
         "        content:\n"
@@ -116,8 +156,12 @@ def render_openapi_contract_fragment() -> str:
         "                message: {type: string, minLength: 1, maxLength: 200}\n"
         "      responses:\n"
         "        '200': {description: Governed commit job and verified branch result}\n"
+        "        '202': {description: Existing external Job alias requires authoritative reconciliation; no replacement Job is created}\n"
+        "        '401': {description: No verified Task principal is available}\n"
+        "        '403': {description: Verified Task principal lacks exact task, external Job, or capability scope}\n"
         "        '400': {description: Contract or policy rejection}\n"
         "        '409': {description: Branch CAS or idempotency conflict}\n"
+        "        '423': {description: Narrow P0 Task-principal mutation gate is frozen; assertion is not consumed}\n"
     )
 
 
@@ -183,20 +227,38 @@ class ControlPlaneReleaseManifest:
 class BridgeV2Adapter:
     """Thin operation adapter; all mutating handlers are injected seams."""
 
-    def __init__(self, *, runtime_status: Mapping[str, Any], handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None) -> None:
-        self.runtime_status = dict(runtime_status)
+    def __init__(self, *, runtime_status: Mapping[str, Any] | Callable[[], Mapping[str, Any]], handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None) -> None:
+        self._runtime_status_source = runtime_status if callable(runtime_status) else dict(runtime_status)
         self.handlers = dict(handlers or {})
 
+    def _runtime_status(self) -> dict[str, Any]:
+        value = self._runtime_status_source() if callable(self._runtime_status_source) else self._runtime_status_source
+        if not isinstance(value, Mapping):
+            raise BridgeV2Error("RUNTIME_STATUS_INVALID")
+        return dict(value)
+
     def getRuntimeStatus(self) -> dict[str, Any]:
-        return {"bridge_schema_version": BRIDGE_SCHEMA_VERSION, **self.runtime_status}
+        return {"bridge_schema_version": BRIDGE_SCHEMA_VERSION, **self._runtime_status()}
 
     def getGovernedWriteCapabilities(self) -> dict[str, Any]:
+        runtime_status = self._runtime_status()
+        governed_writes_enabled = runtime_status.get("governed_writes_enabled") is True
         return {
             "bridge_schema_version": BRIDGE_SCHEMA_VERSION,
             "legacy_writes_enabled": False,
-            "governed_writes_enabled": True,
-            "operations": list(BRIDGE_OPERATIONS),
-            "operation_contracts": operation_contracts(),
+            "governed_writes_enabled": governed_writes_enabled,
+            "governed_write_blockers": list(
+                runtime_status.get("governed_write_blockers") or []
+            ),
+            "operations": [
+                *BRIDGE_OPERATIONS,
+                *(
+                    TASK_ADMISSION_OPERATIONS
+                    if _has_complete_task_admission_handlers(self.handlers)
+                    else ()
+                ),
+            ],
+            "operation_contracts": operation_contracts(self.handlers),
             "force_push": False,
             "main_write": False,
             "tag_write": False,
@@ -213,7 +275,7 @@ class BridgeV2Adapter:
 
     @staticmethod
     def legacy_write_disabled() -> dict[str, object]:
-        return {"error": "LEGACY_WRITE_DISABLED", "replacement": "createGovernedTask", "governed_writes_available": True}
+        return {"error": "LEGACY_WRITE_DISABLED", "replacement": "createGovernedTask", "governed_writes_available": False}
 
     def _call(self, operation: str, request: Mapping[str, Any]) -> Any:
         required = OPERATION_REQUIRED_FIELDS[operation]
@@ -239,5 +301,8 @@ class BridgeV2Adapter:
     def executeApprovedProductionRelease(self, request: Mapping[str, Any]) -> Any: return self._call("executeApprovedProductionRelease", request)
     def getReleaseConsistency(self, request: Mapping[str, Any]) -> Any: return self._call("getReleaseConsistency", request)
     def runReleaseAcceptance(self, request: Mapping[str, Any]) -> Any: return self._call("runReleaseAcceptance", request)
+    def getTaskAdmissionEnvelope(self, request: Mapping[str, Any]) -> Any: return self._call("getTaskAdmissionEnvelope", request)
+    def submitTaskAdmissionAuthorization(self, request: Mapping[str, Any]) -> Any: return self._call("submitTaskAdmissionAuthorization", request)
+    def getTaskAdmissionStatus(self, request: Mapping[str, Any]) -> Any: return self._call("getTaskAdmissionStatus", request)
 
     def getControlPlaneSnapshot(self, request: Mapping[str, Any] | None = None) -> Any: return self._call("getControlPlaneSnapshot", request or {})

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -11,19 +11,36 @@ from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
 from pathlib import Path
 import re
 import sqlite3
-from typing import Any, Mapping
+from typing import Any, Callable, Mapping
+from urllib.parse import parse_qsl, urlsplit
 
 from aios_authority_git_executor import AuthorityGitExecutor
-from aios_chatgpt_bridge_v2 import BridgeV2Adapter, BridgeV2Error, OPERATION_REQUIRED_FIELDS
+from aios_chatgpt_bridge_v2 import (
+    BridgeV2Adapter,
+    BridgeV2Error,
+    OPERATION_REQUIRED_FIELDS,
+    TASK_ADMISSION_OPERATIONS,
+    TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS,
+    TASK_ADMISSION_ROUTES,
+)
+from aios_task_admission_bootstrap import TaskAdmissionError
 from gitea_tag_publication import validate_gitea_api_url
 from aios_continuation_executor import AuthorityGitAdapter, ContinuationCommit, ContinuationExecutor, ContinuationPermissions, ContinuationRuntimeBinding, DetachedGitWorkspace, GiteaContinuationRemote, ServerSideCredentialProjection
 from aios_continuation_lease import WriterLeaseRequestV2
 from aios_control_plane_snapshot import build_control_plane_snapshot
 from aios_gitops_job_controller import GitOpsJobController, JobState, JobStore, _RECOVERABLE, sha256_json
-from aios_governed_task_bootstrap import BootstrapError, GovernedTaskBootstrap
+from aios_governed_task_bootstrap import (
+    BootstrapError,
+    GovernedTaskBinding,
+    GovernedTaskBootstrap,
+)
 from governed_production_release import build_executor_config
 
 MAX_BODY = 2 * 1024 * 1024
+TaskPrincipalVerifier = Callable[
+    [Mapping[str, Any], Mapping[str, Any]], Mapping[str, Any] | None
+]
+_PRIVATE_PRINCIPAL = "_verified_task_principal"
 ROUTES = {
     "/v2/runtime": "getRuntimeStatus",
     "/v2/capabilities": "getGovernedWriteCapabilities",
@@ -42,8 +59,21 @@ ROUTES = {
     "/v2/release/execute": "executeApprovedProductionRelease",
     "/v2/release/consistency": "getReleaseConsistency",
     "/v2/release/acceptance": "runReleaseAcceptance",
+    **TASK_ADMISSION_ROUTES,
 }
 
+# Admission is deliberately separate from a TaskPrincipal.  A successful
+# admission can create only its durable, Owner-authorized Task record; every
+# Git-mutating operation still requires a verified TaskPrincipal and this
+# independently evaluated runtime gate.
+_TASK_PRINCIPAL_MUTATING_OPERATIONS = frozenset({
+    "acquireWriterLease",
+    "commitFileGoverned",
+    "commitFilesGoverned",
+    "resumeGitOpsJob",
+    "reconcileGitOpsJob",
+})
+
 
 @dataclass(frozen=True, slots=True)
 class RuntimeConfig:
@@ -95,6 +125,34 @@ class GovernedRuntime:
         os.environ.setdefault("GIT_ASKPASS", str(askpass)); os.environ.setdefault("GIT_ASKPASS_REQUIRE", "force"); os.environ.setdefault("GIT_TERMINAL_PROMPT", "0"); os.environ.setdefault("GITEA_MACHINE_ACTOR", "aios-autonomy-agent")
 
     def runtime_status(self) -> dict[str, Any]:
+        verifier = BridgeHandler.principal_verifier
+        readiness_reader = getattr(verifier, "readiness", None)
+        if callable(readiness_reader):
+            try:
+                principal_readiness = dict(readiness_reader())
+            except Exception:
+                principal_readiness = {
+                    "ready": False,
+                    "error": "TASK_PRINCIPAL_SEALED_READINESS_UNAVAILABLE",
+                }
+        else:
+            principal_readiness = {
+                "ready": False,
+                "error": "TASK_PRINCIPAL_SEALED_READINESS_UNAVAILABLE",
+            }
+        legacy_global_writes_enabled = os.environ.get(
+            "AIOS_GITOPS_WRITES_ENABLED", "false"
+        ).strip().lower() in {"1", "true", "yes", "on"}
+        task_principal_writes_enabled = os.environ.get(
+            "AIOS_P0_TASK_PRINCIPAL_WRITES_ENABLED", "false"
+        ).strip().lower() in {"1", "true", "yes", "on"}
+        blockers: list[str] = []
+        if principal_readiness.get("ready") is not True:
+            blockers.append("TASK_PRINCIPAL_VERIFIER_NOT_READY")
+        if legacy_global_writes_enabled:
+            blockers.append("LEGACY_GLOBAL_WRITES_MUST_REMAIN_DISABLED")
+        if not task_principal_writes_enabled:
+            blockers.append("TASK_PRINCIPAL_WRITES_DISABLED")
         return {
             "status": "active",
             "health": 200,
@@ -109,9 +167,100 @@ class GovernedRuntime:
             "repository": self.config.repository,
             "repository_identity": self.config.repository,
             "authoritative_git_dir": str(self.config.authoritative_git_dir),
-            "legacy_global_writes_enabled": False,
+            "legacy_global_writes_enabled": legacy_global_writes_enabled,
+            "task_principal_writes_enabled": task_principal_writes_enabled,
+            "principal_verifier_configured": verifier is not None,
+            "task_principal_readiness": principal_readiness,
+            "governed_writes_enabled": not blockers,
+            "governed_write_blockers": blockers,
         }
 
+    def project_admitted_task(
+        self,
+        grant: Any,
+        task_result: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        """Durably bind one verified admission to its exact local Job input.
+
+        This hook is called only after the Agent Task API returns an exact
+        readback.  It writes a local binding, not a Job or a lease; later
+        TaskPrincipal-gated continuation remains responsible for every Git
+        side effect.
+        """
+        task_id = str(getattr(grant, "task_id", ""))
+        repository = str(getattr(grant, "repository", ""))
+        base_oid = str(getattr(grant, "authority_base_oid", ""))
+        target_branch = str(getattr(grant, "target_branch", ""))
+        risk = str(getattr(grant, "risk", ""))
+        generation = int(getattr(grant, "generation", 0))
+        external_job_id = str(task_result.get("external_job_id") or "")
+        if (
+            not task_id
+            or repository != self.config.repository
+            or not base_oid
+            or not target_branch
+            or generation < 1
+            or re.fullmatch(r"job-[0-9a-f]{32}", external_job_id) is None
+        ):
+            raise RuntimeError("TASK_ADMISSION_PROJECTION_COORDINATES_INVALID")
+        projection_key = "task-admission:" + task_id
+        description = (
+            "P0 governed execution projection for officially admitted Agent Task."
+        )
+        try:
+            existing = self.bootstrap.get_binding(task_id)
+        except BootstrapError as exc:
+            if str(exc) != "TASK_EXECUTION_PROJECTION_MISSING":
+                raise
+            existing = None
+        if existing is not None:
+            if (
+                existing.repository != repository
+                or existing.base_oid != base_oid
+                or existing.target_branch != target_branch
+                or existing.risk != risk
+                or existing.description != description
+                or existing.idempotency_key != projection_key
+                or existing.generation != generation
+                or existing.job_id != external_job_id
+            ):
+                raise RuntimeError("TASK_ADMISSION_PROJECTION_BINDING_CONFLICT")
+            result = existing.response(idempotent=True)
+        else:
+            if not self.authority.commit_exists(base_oid):
+                raise RuntimeError("TASK_ADMISSION_PROJECTION_BASE_NOT_FOUND")
+            current_branch_oid = self.authority.read_ref(
+                "refs/heads/" + target_branch
+            )
+            if current_branch_oid is None:
+                branch_mode = "CREATE_NEW"
+                expected_branch_oid = base_oid
+            elif current_branch_oid == base_oid:
+                branch_mode = "CONTINUE_EXISTING"
+                expected_branch_oid = current_branch_oid
+            else:
+                raise RuntimeError("TASK_ADMISSION_PROJECTION_BRANCH_CONFLICT")
+            binding = GovernedTaskBinding(
+                task_id=task_id,
+                repository=repository,
+                base_oid=base_oid,
+                target_branch=target_branch,
+                risk=risk,
+                description=description,
+                idempotency_key=projection_key,
+                expected_branch_oid=expected_branch_oid,
+                branch_mode=branch_mode,
+                generation=generation,
+                job_id=external_job_id,
+            )
+            self.bootstrap._write(binding)
+            readback = self.bootstrap.get_binding(task_id)
+            if readback != binding:
+                raise RuntimeError("TASK_ADMISSION_PROJECTION_READBACK_MISMATCH")
+            result = readback.response()
+        self.bindings[task_id] = dict(result)
+        return result
+
     def _job_root(self, task_id: str) -> Path:
         return self.config.state_root / "jobs" / _safe(task_id)
 
@@ -732,7 +881,8 @@ class GovernedRuntime:
 
     def snapshot(self) -> dict[str, Any]:
         task = self._snapshot_task()
-        return build_control_plane_snapshot(target_version=self.config.target_version, runtime=self.runtime_status(), repository={"identity": self.config.repository, "authority_git_dir": str(self.config.authoritative_git_dir), "authority_main_oid": self.authority.read_ref("refs/heads/main")}, production={"baseline_version": "V2R1C7", "target_oid": "c7adece1cc8beffe7c718d53c775898b45d6c7f8", "deployment_state": "ACTIVE"}, task=task, capabilities={"legacy_writes_enabled": False, "governed_writes_enabled": True, "force_push": False, "main_write": False, "tag_write": False, "baseline_write": False, "production_write": False}, credential_broker={"status": "SERVER_SIDE_ONLY", "provider": "FileCredentialProvider"})
+        runtime = self.runtime_status()
+        return build_control_plane_snapshot(target_version=self.config.target_version, runtime=runtime, repository={"identity": self.config.repository, "authority_git_dir": str(self.config.authoritative_git_dir), "authority_main_oid": self.authority.read_ref("refs/heads/main")}, production={"baseline_version": "V2R1C7", "target_oid": "c7adece1cc8beffe7c718d53c775898b45d6c7f8", "deployment_state": "ACTIVE"}, task=task, capabilities={"legacy_writes_enabled": False, "governed_writes_enabled": runtime["governed_writes_enabled"], "force_push": False, "main_write": False, "tag_write": False, "baseline_write": False, "production_write": False}, credential_broker={"status": "SERVER_SIDE_ONLY", "provider": "FileCredentialProvider"})
 
     def handlers(self) -> dict[str, Any]:
         return {
@@ -760,47 +910,197 @@ def build_governed_handlers(runtime_config: RuntimeConfig | Mapping[str, Any]) -
     return GovernedRuntime(runtime_config).handlers()
 
 
+def build_bridge_adapter(
+    runtime: GovernedRuntime,
+    *,
+    task_admission_handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None,
+) -> BridgeV2Adapter:
+    handlers = dict(runtime.handlers())
+    if task_admission_handlers is not None:
+        if (
+            set(task_admission_handlers) != set(TASK_ADMISSION_OPERATIONS)
+            or any(
+                not callable(task_admission_handlers.get(operation))
+                for operation in TASK_ADMISSION_OPERATIONS
+            )
+        ):
+            raise ValueError("TASK_ADMISSION_HANDLERS_INCOMPLETE")
+        handlers.update(task_admission_handlers)
+    return BridgeV2Adapter(runtime_status=runtime.runtime_status, handlers=handlers)
+
+
 class _HTTPError(RuntimeError): pass
 
 
 class BridgeHandler(BaseHTTPRequestHandler):
     adapter: BridgeV2Adapter | None = None
+    principal_verifier: TaskPrincipalVerifier | None = None
+
     def log_message(self, *_args: Any) -> None: return
+
     def _json(self, value: Any, status: int = 200) -> None:
         raw = json.dumps(value, sort_keys=True, separators=(",", ":")).encode(); self.send_response(status); self.send_header("Content-Type", "application/json"); self.send_header("Content-Length", str(len(raw))); self.end_headers(); self.wfile.write(raw)
+
+    @staticmethod
+    def _result_status(value: Any) -> int:
+        status = value.get("http_status") if isinstance(value, Mapping) else None
+        return status if isinstance(status, int) and 100 <= status <= 599 else 200
+
     def _body(self) -> dict[str, Any]:
         try: length = int(self.headers.get("Content-Length", "-1"))
         except ValueError as exc: raise _HTTPError("CONTENT_LENGTH_INVALID") from exc
         if length < 0 or length > MAX_BODY: raise _HTTPError("REQUEST_TOO_LARGE")
-        value = json.loads(self.rfile.read(length).decode("utf-8"))
+        def strict_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+            value: dict[str, Any] = {}
+            for key, item in pairs:
+                if key in value:
+                    raise _HTTPError("JSON_DUPLICATE_FIELD")
+                value[key] = item
+            return value
+        value = json.loads(
+            self.rfile.read(length).decode("utf-8"),
+            object_pairs_hook=strict_object,
+        )
         if not isinstance(value, dict): raise _HTTPError("JSON_OBJECT_REQUIRED")
         return value
+
     def _allowed(self, operation: str, body: Mapping[str, Any]) -> None:
-        allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | {"schema", "lease_seconds", "create_target_branch", "worktree_mode", "cas_ref_update_required", "expected_branch_oid", "generation", "branch_mode", "source_lease_epoch"}
+        if operation in TASK_ADMISSION_OPERATIONS:
+            allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | set(
+                TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS[operation]
+            )
+        else:
+            allowed = set(OPERATION_REQUIRED_FIELDS[operation]) | {"schema", "lease_seconds", "create_target_branch", "worktree_mode", "cas_ref_update_required", "expected_branch_oid", "expected_revision", "generation", "branch_mode", "source_lease_epoch"}
+        if operation == "projectOnlineAgentExecution": allowed.add("execution")
         if operation == "commitFilesGoverned": allowed.add("files")
-        allowed.update({"owner_approval", "execute", "job_id", "historical_lease_head_oid"})
+        if operation not in TASK_ADMISSION_OPERATIONS:
+            allowed.update({"owner_approval", "execute", "job_id", "historical_lease_head_oid"})
+        if "principal" in body or _PRIVATE_PRINCIPAL in body:
+            raise _HTTPError("CALLER_PRINCIPAL_FORBIDDEN")
         if set(body) - allowed: raise _HTTPError("UNKNOWN_FIELDS")
+
+    def _verified_request(self, operation: str, body: Mapping[str, Any]) -> dict[str, Any]:
+        protected = {
+            "createGovernedTask", "projectOnlineAgentExecution",
+            "acquireWriterLease", "commitFileGoverned", "commitFilesGoverned",
+            "getGitOpsJob", "resumeGitOpsJob", "reconcileGitOpsJob",
+        }
+        if operation not in protected:
+            return dict(body)
+        if operation in _TASK_PRINCIPAL_MUTATING_OPERATIONS:
+            adapter = self.adapter
+            if adapter is None:
+                raise _HTTPError("TASK_PRINCIPAL_WRITES_DISABLED")
+            try:
+                runtime = adapter.getRuntimeStatus()
+            except Exception as exc:
+                raise _HTTPError("TASK_PRINCIPAL_WRITES_DISABLED") from exc
+            if (
+                not isinstance(runtime, Mapping)
+                or runtime.get("governed_writes_enabled") is not True
+                or runtime.get("legacy_global_writes_enabled") is not False
+                or runtime.get("task_principal_writes_enabled") is not True
+            ):
+                raise _HTTPError("TASK_PRINCIPAL_WRITES_DISABLED")
+        verifier = type(self).principal_verifier
+        if verifier is None:
+            raise _HTTPError("TASK_PRINCIPAL_VERIFIER_UNAVAILABLE")
+        get_all = getattr(self.headers, "get_all", None)
+        if callable(get_all):
+            authorization_values = get_all("Authorization") or []
+        else:
+            authorization = self.headers.get("Authorization")
+            authorization_values = [] if authorization is None else [authorization]
+        transport = {
+            "operation": operation,
+            "client_address": self.client_address[0] if self.client_address else None,
+            "headers": {
+                "authorization": authorization_values[0] if len(authorization_values) == 1 else None,
+                "x-request-id": self.headers.get("X-Request-Id"),
+            },
+        }
+        principal = verifier(transport, dict(body))
+        if not isinstance(principal, Mapping):
+            raise _HTTPError("TASK_PRINCIPAL_REJECTED")
+        return {**body, _PRIVATE_PRINCIPAL: dict(principal)}
+
     def do_GET(self) -> None:
         if self.adapter is None: self._json({"error": "RUNTIME_UNAVAILABLE"}, 503); return
         if self.path == "/v2/runtime": self._json(self.adapter.getRuntimeStatus()); return
         if self.path == "/v2/capabilities": self._json(self.adapter.getGovernedWriteCapabilities()); return
         if self.path == "/v2/control-plane-snapshot": self._json(self.adapter.getControlPlaneSnapshot({})); return
+        parsed = urlsplit(self.path)
+        if parsed.path == "/v2/task-admission/status":
+            if not callable(self.adapter.handlers.get("getTaskAdmissionStatus")):
+                self._json({"error": "NOT_FOUND"}, 404)
+                return
+            try:
+                parameters = parse_qsl(parsed.query, keep_blank_values=True)
+                if (
+                    len(parameters) != 1
+                    or parameters[0][0] != "envelope_id"
+                    or not parameters[0][1]
+                ):
+                    raise _HTTPError("TASK_ADMISSION_QUERY_INVALID")
+                result = self.adapter.getTaskAdmissionStatus(
+                    {"envelope_id": parameters[0][1]}
+                )
+                self._json(result, self._result_status(result))
+            except TaskAdmissionError as exc:
+                self._json({"error": str(exc)}, exc.status)
+            except _HTTPError as exc:
+                self._json({"error": str(exc)}, 400)
+            except (ValueError, BridgeV2Error) as exc:
+                self._json({"error": str(exc)}, 400)
+            except Exception as exc:
+                self._json({"error": str(exc) or type(exc).__name__}, 503)
+            return
         self._json({"error": "NOT_FOUND"}, 404)
+
     def do_POST(self) -> None:
         operation = ROUTES.get(self.path)
         if operation is None or operation in ("getRuntimeStatus", "getGovernedWriteCapabilities") or self.adapter is None: self._json({"error": "NOT_FOUND"}, 404); return
+        if operation in TASK_ADMISSION_OPERATIONS and not callable(self.adapter.handlers.get(operation)):
+            self._json({"error": "NOT_FOUND"}, 404)
+            return
         try:
-            body = self._body(); self._allowed(operation, body); self._json(getattr(self.adapter, operation)(body))
-        except _HTTPError as exc: self._json({"error": str(exc)}, 413 if str(exc) == "REQUEST_TOO_LARGE" else 400)
+            body = self._body(); self._allowed(operation, body)
+            result = getattr(self.adapter, operation)(self._verified_request(operation, body))
+            self._json(result, self._result_status(result))
+        except _HTTPError as exc:
+            error = str(exc)
+            if error == "REQUEST_TOO_LARGE":
+                status = 413
+            elif error == "TASK_PRINCIPAL_WRITES_DISABLED":
+                status = 423
+            else:
+                status = 401 if error.startswith("TASK_PRINCIPAL_") else 400
+            self._json({"error": error}, status)
+        except TaskAdmissionError as exc: self._json({"error": str(exc)}, exc.status)
         except (BootstrapError, ValueError, BridgeV2Error) as exc: self._json({"error": str(exc)}, 409 if "CONFLICT" in str(exc) else 400)
-        except Exception as exc: self._json({"error": str(exc) or type(exc).__name__}, 409 if any(token in str(exc) for token in ("CONFLICT", "EXPECTATION", "MONOTONIC")) else 503)
+        except Exception as exc:
+            error = str(exc) or type(exc).__name__
+            if error.startswith("TASK_PRINCIPAL_"):
+                status = 403
+            elif any(token in error for token in ("CONFLICT", "EXPECTATION", "MONOTONIC", "REVISION_CAS")):
+                status = 409
+            else:
+                status = 503
+            self._json({"error": error}, status)
 
 
 def serve(adapter: BridgeV2Adapter, host: str = "127.0.0.1", port: int = 18892) -> None:
     BridgeHandler.adapter = adapter; ThreadingHTTPServer((host, port), BridgeHandler).serve_forever()
 
 
-def main() -> None:
+def main(
+    *,
+    task_admission_handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None,
+) -> None:
+    activation_raw = os.environ.get(
+        "AIOS_TASK_ADMISSION_ACTIVATION_FILE",
+        "/var/lib/aios-gitops/bridge-v2-staging/task-admission-activation.json",
+    ).strip()
     state_root = Path("/var/lib/aios-gitops/bridge-v2-staging")
     token_path = Path(os.environ.get("AIOS_GITEA_TOKEN_FILE", "/run/credentials/aios-chatgpt-bridge-v2-staging.service/gitea_token"))
     ssh_key_path = Path(os.environ.get("AIOS_GITEA_SSH_KEY_PATH", "/home/aiman/.ssh/id_ed25519_codex_tmp"))
@@ -822,8 +1122,32 @@ def main() -> None:
         )),
     )
     runtime = GovernedRuntime(config)
+    # A TaskPrincipal verifier is deliberately supplied only by the hosting
+    # service.  Without that independently authenticated verifier, all Git
+    # mutation requests fail closed while the admission-only routes remain
+    # available for a configured Owner-approved epoch.
+    BridgeHandler.principal_verifier = None
+    if task_admission_handlers is None:
+        if os.environ.get("AIOS_TASK_ADMISSION_EPOCH_ROOT", "").strip():
+            from aios_task_admission_live_adapters import (
+                build_task_admission_handlers_from_environment,
+            )
+            task_admission_handlers = build_task_admission_handlers_from_environment(
+                post_admission=runtime.project_admitted_task,
+            )
+        elif activation_raw and Path(activation_raw).is_file():
+            from aios_task_admission_live_adapters import (
+                build_task_admission_handlers_from_activation_file,
+            )
+            task_admission_handlers = build_task_admission_handlers_from_activation_file(
+                Path(activation_raw),
+                post_admission=runtime.project_admitted_task,
+            )
     serve(
-        BridgeV2Adapter(runtime_status=runtime.runtime_status(), handlers=runtime.handlers()),
+        build_bridge_adapter(
+            runtime,
+            task_admission_handlers=task_admission_handlers,
+        ),
         port=int(os.environ.get("AIOS_BRIDGE_V2_PORT", "18892")),
     )
 

active-runtime/app/aios_task_admission_bootstrap.py

@@ -0,0 +1,945 @@
+"""Fail-closed bootstrap for narrow Owner-delegated Agent Task admission.
+
+This module solves the authorization bootstrap problem without turning the
+runtime anchor into a writer and without weakening TaskPrincipal semantics.
+
+Security model:
+
+* A task-admission envelope is created from a server-normalized request.
+* The envelope is authorized either by an already-valid OwnerDelegationEpoch
+  or by an Owner signature verified by an injected trust backend.
+* The derived TaskAdmissionGrant can only narrow the parent authorization.
+* The grant is handed to the existing Agent Task control plane through an
+  injected TaskAdmissionPort. This module never writes the Task database.
+* TaskPrincipal issuance remains downstream of a real Task claim/TaskLease.
+* Exact idempotent replay returns the same admitted Task; conflicting reuse is
+  rejected.
+"""
+
+from __future__ import annotations
+
+import base64
+from contextlib import closing, contextmanager
+from dataclasses import asdict, dataclass
+from datetime import datetime, timezone
+import hashlib
+import json
+from pathlib import Path, PurePosixPath
+import re
+import secrets
+import sqlite3
+import threading
+from typing import Any, Callable, Mapping, Protocol, Sequence
+
+
+SCHEMA_ENVELOPE = "aios.task-admission-envelope.v1"
+SCHEMA_GRANT = "aios.task-admission-grant.v1"
+SCHEMA_EPOCH = "aios.owner-delegation-epoch.v1"
+SCHEMA_RECEIPT = "aios.task-admission-receipt.v1"
+
+OID_RE = re.compile(r"^[0-9a-f]{40}$")
+TASK_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{2,179}$")
+BRANCH_RE = re.compile(r"^agent/[A-Za-z0-9][A-Za-z0-9._/-]{2,179}$")
+IDEMPOTENCY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@/-]{7,255}$")
+CAPABILITY_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{1,127}$")
+KEY_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:@+-]{2,255}$")
+
+PROHIBITED_CAPABILITIES = frozenset(
+    {
+        "git.global_write",
+        "git.main_write",
+        "git.force_push",
+        "tag.publish",
+        "baseline.finalize",
+        "production.write",
+        "production.switch",
+        "runtime_anchor.write",
+        "shell.arbitrary",
+        "credential.export",
+    }
+)
+
+
+class TaskAdmissionError(RuntimeError):
+    """Stable fail-closed task-admission error."""
+
+    def __init__(
+        self,
+        code: str,
+        status: int = 400,
+        *,
+        details: Mapping[str, Any] | None = None,
+    ) -> None:
+        super().__init__(code)
+        self.code = code
+        self.status = status
+        self.details = dict(details or {})
+
+
+@dataclass(frozen=True, slots=True)
+class TaskAdmissionPolicy:
+    repository: str = "aios/aios"
+    trust_domain: str = "aios.owner"
+    branch_prefix: str = "agent/"
+    allowed_capabilities: frozenset[str] = frozenset(
+        {
+            "governed.task.create",
+            "agent-task:control",
+            "task.principal.issue",
+            "writer.lease",
+            "governed.write.files",
+            "git.read",
+            "git.test",
+            "governed.write.branch",
+            "governed.write.file",
+            "testing",
+            "verification",
+        }
+    )
+    max_file_count: int = 64
+    max_task_ttl_seconds: int = 3600
+    max_envelope_ttl_seconds: int = 900
+    max_risk: str = "medium"
+    policy_digest: str = ""
+
+    def normalized_policy_digest(self) -> str:
+        if self.policy_digest:
+            return self.policy_digest
+        return sha256_json(
+            {
+                "repository": self.repository,
+                "trust_domain": self.trust_domain,
+                "branch_prefix": self.branch_prefix,
+                "allowed_capabilities": sorted(self.allowed_capabilities),
+                "max_file_count": self.max_file_count,
+                "max_task_ttl_seconds": self.max_task_ttl_seconds,
+                "max_envelope_ttl_seconds": self.max_envelope_ttl_seconds,
+                "max_risk": self.max_risk,
+                "prohibited_capabilities": sorted(PROHIBITED_CAPABILITIES),
+            }
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class OwnerDelegationEpoch:
+    """Already-verified durable Owner authorization.
+
+    The verifier/backend is responsible for authenticating the epoch before
+    returning it to this service.
+    """
+
+    epoch_id: str
+    owner_identity: str
+    trust_domain: str
+    repository: str
+    allowed_branch_prefixes: tuple[str, ...]
+    allowed_capabilities: frozenset[str]
+    allowed_file_prefixes: tuple[str, ...]
+    risk_ceiling: str
+    issued_at: int
+    not_before: int
+    expires_at: int
+    revocation_epoch: int
+    policy_digest: str
+    schema: str = SCHEMA_EPOCH
+
+    def as_public_dict(self) -> dict[str, Any]:
+        value = asdict(self)
+        value["allowed_capabilities"] = sorted(self.allowed_capabilities)
+        value["allowed_branch_prefixes"] = list(self.allowed_branch_prefixes)
+        value["allowed_file_prefixes"] = list(self.allowed_file_prefixes)
+        return value
+
+
+@dataclass(frozen=True, slots=True)
+class TaskAdmissionEnvelope:
+    envelope_id: str
+    idempotency_key: str
+    task_id: str
+    repository: str
+    authority_base_oid: str
+    target_branch: str
+    generation: int
+    execution_digest: str
+    requested_capabilities: tuple[str, ...]
+    file_allowlist: tuple[str, ...]
+    risk: str
+    trust_domain: str
+    policy_digest: str
+    issued_at: int
+    not_before: int
+    expires_at: int
+    nonce: str
+    request_digest: str
+    schema: str = SCHEMA_ENVELOPE
+
+    def as_dict(self) -> dict[str, Any]:
+        value = asdict(self)
+        value["requested_capabilities"] = list(self.requested_capabilities)
+        value["file_allowlist"] = list(self.file_allowlist)
+        return value
+
+
+@dataclass(frozen=True, slots=True)
+class TaskAdmissionGrant:
+    grant_id: str
+    envelope_id: str
+    task_id: str
+    repository: str
+    authority_base_oid: str
+    target_branch: str
+    generation: int
+    execution_digest: str
+    capabilities: tuple[str, ...]
+    file_allowlist: tuple[str, ...]
+    risk: str
+    trust_domain: str
+    policy_digest: str
+    authorization_kind: str
+    owner_identity: str
+    parent_epoch_id: str | None
+    issued_at: int
+    expires_at: int
+    one_time_use: bool = True
+    schema: str = SCHEMA_GRANT
+
+    def as_dict(self) -> dict[str, Any]:
+        value = asdict(self)
+        value["capabilities"] = list(self.capabilities)
+        value["file_allowlist"] = list(self.file_allowlist)
+        return value
+
+
+class OwnerAuthorizationBackend(Protocol):
+    """Bridge to existing trusted Owner authorization.
+
+    No permissive default implementation is provided.
+    """
+
+    def verify_delegation_epoch(self, epoch_id: str) -> OwnerDelegationEpoch:
+        ...
+
+    def verify_owner_signature(
+        self,
+        *,
+        key_id: str,
+        canonical_payload: bytes,
+        signature_b64: str,
+    ) -> str:
+        """Return the verified Owner identity or raise."""
+        ...
+
+
+class TaskAdmissionPort(Protocol):
+    """Existing Agent Task control-plane port."""
+
+    def admit(self, grant: TaskAdmissionGrant) -> Mapping[str, Any]:
+        ...
+
+
+def utc_seconds() -> int:
+    return int(datetime.now(timezone.utc).timestamp())
+
+
+def canonical_json_bytes(value: Any) -> bytes:
+    return json.dumps(
+        value,
+        ensure_ascii=False,
+        sort_keys=True,
+        separators=(",", ":"),
+    ).encode("utf-8")
+
+
+def sha256_json(value: Any) -> str:
+    return hashlib.sha256(canonical_json_bytes(value)).hexdigest()
+
+
+def _normalize_files(raw: Sequence[object], *, maximum: int) -> tuple[str, ...]:
+    if not isinstance(raw, (list, tuple)):
+        raise TaskAdmissionError("TASK_ADMISSION_FILE_ALLOWLIST_REJECTED", 403)
+    if not raw or len(raw) > maximum:
+        raise TaskAdmissionError("TASK_ADMISSION_FILE_ALLOWLIST_REJECTED", 403)
+    normalized: list[str] = []
+    seen: set[str] = set()
+    for item in raw:
+        if (
+            not isinstance(item, str)
+            or not item
+            or "\\" in item
+            or any(ord(char) < 32 or ord(char) == 127 for char in item)
+        ):
+            raise TaskAdmissionError("TASK_ADMISSION_FILE_PATH_INVALID", 403)
+        path = PurePosixPath(item)
+        if path.is_absolute() or ".." in path.parts or "." in path.parts:
+            raise TaskAdmissionError("TASK_ADMISSION_FILE_PATH_INVALID", 403)
+        text = path.as_posix()
+        # Do not normalize a caller supplied path.  A signature must bind the
+        # exact spelling, not a convenient equivalent such as ``a//b``.
+        if text != item:
+            raise TaskAdmissionError("TASK_ADMISSION_FILE_PATH_INVALID", 403)
+        if text.startswith(".git/") or text == ".git":
+            raise TaskAdmissionError("TASK_ADMISSION_FILE_PATH_INVALID", 403)
+        if text not in seen:
+            normalized.append(text)
+            seen.add(text)
+    if not normalized:
+        raise TaskAdmissionError("TASK_ADMISSION_FILE_ALLOWLIST_REJECTED", 403)
+    return tuple(sorted(normalized))
+
+
+def _normalize_capabilities(
+    raw: Sequence[object],
+    *,
+    policy: TaskAdmissionPolicy,
+) -> tuple[str, ...]:
+    if not raw:
+        raise TaskAdmissionError("TASK_ADMISSION_CAPABILITIES_REQUIRED", 403)
+    values: set[str] = set()
+    for item in raw:
+        if not isinstance(item, str) or not CAPABILITY_RE.fullmatch(item):
+            raise TaskAdmissionError("TASK_ADMISSION_CAPABILITY_INVALID", 403)
+        if "*" in item:
+            raise TaskAdmissionError("TASK_ADMISSION_WILDCARD_CAPABILITY_REJECTED", 403)
+        values.add(item)
+    if values & PROHIBITED_CAPABILITIES:
+        raise TaskAdmissionError("TASK_ADMISSION_PROHIBITED_CAPABILITY", 403)
+    if not values.issubset(policy.allowed_capabilities):
+        raise TaskAdmissionError("TASK_ADMISSION_CAPABILITY_OUT_OF_POLICY", 403)
+    return tuple(sorted(values))
+
+
+def _risk_rank(value: str) -> int:
+    ranks = {"low": 0, "medium": 1, "high": 2}
+    try:
+        return ranks[value]
+    except KeyError:
+        raise TaskAdmissionError("TASK_ADMISSION_RISK_INVALID", 403) from None
+
+
+def _request_for_digest(
+    *,
+    task_id: str,
+    repository: str,
+    authority_base_oid: str,
+    target_branch: str,
+    generation: int,
+    execution_digest: str,
+    requested_capabilities: tuple[str, ...],
+    file_allowlist: tuple[str, ...],
+    risk: str,
+    trust_domain: str,
+    policy_digest: str,
+) -> dict[str, Any]:
+    return {
+        "task_id": task_id,
+        "repository": repository,
+        "authority_base_oid": authority_base_oid,
+        "target_branch": target_branch,
+        "generation": generation,
+        "execution_digest": execution_digest,
+        "requested_capabilities": list(requested_capabilities),
+        "file_allowlist": list(file_allowlist),
+        "risk": risk,
+        "trust_domain": trust_domain,
+        "policy_digest": policy_digest,
+    }
+
+
+class TaskAdmissionStore:
+    """Portable durable state; all replay decisions are transactionally stored."""
+
+    def __init__(self, path: Path) -> None:
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.lock_root = Path(str(self.path) + ".locks")
+        self.lock_root.mkdir(parents=True, exist_ok=True)
+        self._lock = threading.RLock()
+        with self._connection() as con:
+            con.executescript(
+                """
+                PRAGMA journal_mode=WAL;
+                CREATE TABLE IF NOT EXISTS task_admission_envelopes(
+                    idempotency_key TEXT PRIMARY KEY,
+                    envelope_id TEXT NOT NULL UNIQUE,
+                    request_digest TEXT NOT NULL,
+                    envelope_json TEXT NOT NULL,
+                    status TEXT NOT NULL,
+                    created_at INTEGER NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS task_admission_grants(
+                    envelope_id TEXT PRIMARY KEY,
+                    grant_id TEXT NOT NULL UNIQUE,
+                    grant_digest TEXT NOT NULL,
+                    grant_json TEXT NOT NULL,
+                    authorization_digest TEXT NOT NULL DEFAULT '',
+                    created_at INTEGER NOT NULL
+                );
+                CREATE TABLE IF NOT EXISTS task_admission_receipts(
+                    envelope_id TEXT PRIMARY KEY,
+                    task_id TEXT NOT NULL UNIQUE,
+                    receipt_json TEXT NOT NULL,
+                    created_at INTEGER NOT NULL
+                );
+                """
+            )
+            columns = {
+                row["name"] for row in con.execute("PRAGMA table_info(task_admission_grants)")
+            }
+            if "authorization_digest" not in columns:
+                con.execute(
+                    "ALTER TABLE task_admission_grants ADD COLUMN authorization_digest TEXT NOT NULL DEFAULT ''"
+                )
+
+    def _connect(self) -> sqlite3.Connection:
+        con = sqlite3.connect(self.path, timeout=30, isolation_level=None)
+        con.row_factory = sqlite3.Row
+        return con
+
+    @contextmanager
+    def _connection(self):
+        """Yield an owned main-store connection and close it deterministically."""
+        with closing(self._connect()) as con:
+            try:
+                yield con
+            finally:
+                if con.in_transaction:
+                    con.rollback()
+
+    @contextmanager
+    def admission_lock(self, envelope_id: str, *, timeout_seconds: float = 5.0):
+        """Hold a crash-releasing lock for one envelope, outside the main DB."""
+        lock_name = hashlib.sha256(envelope_id.encode("utf-8")).hexdigest() + ".sqlite3"
+        lock_path = self.lock_root / lock_name
+        with closing(sqlite3.connect(lock_path, timeout=timeout_seconds, isolation_level=None)) as con:
+            try:
+                con.execute("BEGIN EXCLUSIVE")
+            except sqlite3.OperationalError as exc:
+                raise TaskAdmissionError("TASK_ADMISSION_IN_PROGRESS", 503) from exc
+            try:
+                yield
+                con.execute("COMMIT")
+            finally:
+                if con.in_transaction:
+                    con.rollback()
+
+    def get_by_idempotency(self, key: str) -> dict[str, Any] | None:
+        with self._connection() as con:
+            row = con.execute(
+                "SELECT * FROM task_admission_envelopes WHERE idempotency_key=?",
+                (key,),
+            ).fetchone()
+        return dict(row) if row else None
+
+    def get_envelope(self, envelope_id: str) -> dict[str, Any] | None:
+        with self._connection() as con:
+            row = con.execute(
+                "SELECT * FROM task_admission_envelopes WHERE envelope_id=?",
+                (envelope_id,),
+            ).fetchone()
+        return dict(row) if row else None
+
+    def save_envelope(
+        self,
+        *,
+        idempotency_key: str,
+        envelope_id: str,
+        request_digest: str,
+        envelope_json: str,
+        created_at: int,
+    ) -> dict[str, Any]:
+        with self._lock, self._connection() as con:
+            con.execute("BEGIN IMMEDIATE")
+            row = con.execute(
+                "SELECT request_digest,envelope_id FROM task_admission_envelopes WHERE idempotency_key=?",
+                (idempotency_key,),
+            ).fetchone()
+            if row is not None:
+                con.execute("ROLLBACK")
+                if row["request_digest"] != request_digest:
+                    raise TaskAdmissionError("TASK_ADMISSION_IDEMPOTENCY_CONFLICT", 409)
+                return json.loads(
+                    con.execute(
+                        "SELECT envelope_json FROM task_admission_envelopes WHERE idempotency_key=?",
+                        (idempotency_key,),
+                    ).fetchone()["envelope_json"]
+                )
+            con.execute(
+                """
+                INSERT INTO task_admission_envelopes(
+                    idempotency_key,envelope_id,request_digest,envelope_json,status,created_at
+                ) VALUES(?,?,?,?,?,?)
+                """,
+                (
+                    idempotency_key,
+                    envelope_id,
+                    request_digest,
+                    envelope_json,
+                    "PENDING_AUTHORIZATION",
+                    created_at,
+                ),
+            )
+            con.execute("COMMIT")
+        return json.loads(envelope_json)
+
+    def claim_grant(
+        self,
+        *,
+        envelope_id: str,
+        grant: TaskAdmissionGrant,
+        authorization_digest: str,
+        created_at: int,
+    ) -> tuple[TaskAdmissionGrant, bool]:
+        """Persist the immutable grant before the external Task side effect.
+
+        ``BEGIN IMMEDIATE`` serializes only this SQLite writer transaction;
+        it does not serialize unrelated envelopes.  A second proof cannot
+        replace the winning proof/grant for the same envelope.
+        """
+        grant_json = canonical_json_bytes(grant.as_dict()).decode("utf-8")
+        grant_digest = hashlib.sha256(grant_json.encode("utf-8")).hexdigest()
+        with self._lock, self._connection() as con:
+            con.execute("BEGIN IMMEDIATE")
+            row = con.execute(
+                "SELECT grant_json,authorization_digest FROM task_admission_grants WHERE envelope_id=?",
+                (envelope_id,),
+            ).fetchone()
+            if row is not None:
+                con.execute("ROLLBACK")
+                if row["authorization_digest"] != authorization_digest:
+                    raise TaskAdmissionError("TASK_ADMISSION_AUTHORIZATION_CONFLICT", 409)
+                raw = json.loads(row["grant_json"])
+                return (
+                    TaskAdmissionGrant(
+                        **{
+                            **raw,
+                            "capabilities": tuple(raw["capabilities"]),
+                            "file_allowlist": tuple(raw["file_allowlist"]),
+                        }
+                    ),
+                    False,
+                )
+            con.execute(
+                "INSERT INTO task_admission_grants(envelope_id,grant_id,grant_digest,grant_json,authorization_digest,created_at) VALUES(?,?,?,?,?,?)",
+                (envelope_id, grant.grant_id, grant_digest, grant_json, authorization_digest, created_at),
+            )
+            con.execute(
+                "UPDATE task_admission_envelopes SET status='ADMITTING' WHERE envelope_id=?",
+                (envelope_id,),
+            )
+            con.execute("COMMIT")
+        return grant, True
+
+    def receipt(self, envelope_id: str) -> dict[str, Any] | None:
+        with self._connection() as con:
+            row = con.execute(
+                "SELECT receipt_json FROM task_admission_receipts WHERE envelope_id=?",
+                (envelope_id,),
+            ).fetchone()
+        return json.loads(row["receipt_json"]) if row else None
+
+    def save_grant_and_receipt(
+        self,
+        *,
+        envelope_id: str,
+        grant: TaskAdmissionGrant,
+        receipt: Mapping[str, Any],
+        created_at: int,
+    ) -> dict[str, Any]:
+        grant_json = canonical_json_bytes(grant.as_dict()).decode("utf-8")
+        grant_digest = hashlib.sha256(grant_json.encode("utf-8")).hexdigest()
+        receipt_json = canonical_json_bytes(dict(receipt)).decode("utf-8")
+        with self._lock, self._connection() as con:
+            con.execute("BEGIN IMMEDIATE")
+            existing = con.execute(
+                "SELECT receipt_json FROM task_admission_receipts WHERE envelope_id=?",
+                (envelope_id,),
+            ).fetchone()
+            if existing is not None:
+                con.execute("ROLLBACK")
+                return json.loads(existing["receipt_json"])
+            stored = con.execute(
+                "SELECT grant_id,grant_digest FROM task_admission_grants WHERE envelope_id=?",
+                (envelope_id,),
+            ).fetchone()
+            if stored is None:
+                con.execute(
+                    "INSERT INTO task_admission_grants(envelope_id,grant_id,grant_digest,grant_json,authorization_digest,created_at) VALUES(?,?,?,?,?,?)",
+                    (envelope_id, grant.grant_id, grant_digest, grant_json, "", created_at),
+                )
+            elif stored["grant_id"] != grant.grant_id or stored["grant_digest"] != grant_digest:
+                con.execute("ROLLBACK")
+                raise TaskAdmissionError("TASK_ADMISSION_GRANT_CONFLICT", 409)
+            con.execute(
+                """
+                INSERT INTO task_admission_receipts(
+                    envelope_id,task_id,receipt_json,created_at
+                ) VALUES(?,?,?,?)
+                """,
+                (
+                    envelope_id,
+                    grant.task_id,
+                    receipt_json,
+                    created_at,
+                ),
+            )
+            con.execute(
+                "UPDATE task_admission_envelopes SET status='ADMITTED' WHERE envelope_id=?",
+                (envelope_id,),
+            )
+            con.execute("COMMIT")
+        return dict(receipt)
+
+    def status(self, envelope_id: str) -> dict[str, Any]:
+        row = self.get_envelope(envelope_id)
+        if row is None:
+            raise TaskAdmissionError("TASK_ADMISSION_ENVELOPE_NOT_FOUND", 404)
+        receipt = self.receipt(envelope_id)
+        return {
+            "envelope_id": envelope_id,
+            "status": row["status"],
+            "task": receipt,
+        }
+
+
+class TaskAdmissionBootstrapService:
+    def __init__(
+        self,
+        *,
+        store: TaskAdmissionStore,
+        authorization: OwnerAuthorizationBackend,
+        task_port: TaskAdmissionPort,
+        policy: TaskAdmissionPolicy | None = None,
+        clock=utc_seconds,
+        post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
+    ) -> None:
+        self.store = store
+        self.authorization = authorization
+        self.task_port = task_port
+        self.policy = policy or TaskAdmissionPolicy()
+        self.clock = clock
+        # This is a narrow server-side projection boundary. The official
+        # Agent Task port remains the admission authority; a projection may
+        # only establish local execution state after its exact Task readback.
+        self.post_admission = post_admission
+        self._admit_lock = threading.RLock()
+
+    def create_envelope(self, raw: Mapping[str, Any]) -> dict[str, Any]:
+        required = {
+            "idempotency_key",
+            "task_id",
+            "repository",
+            "authority_base_oid",
+            "target_branch",
+            "generation",
+            "execution_digest",
+            "requested_capabilities",
+            "file_allowlist",
+            "risk",
+        }
+        optional = {"expires_in_seconds"}
+        if set(raw) - (required | optional) or not required.issubset(raw):
+            raise TaskAdmissionError("TASK_ADMISSION_REQUEST_FIELDS_INVALID")
+        idempotency_key = str(raw["idempotency_key"])
+        task_id = str(raw["task_id"])
+        repository = str(raw["repository"])
+        authority_base_oid = str(raw["authority_base_oid"])
+        target_branch = str(raw["target_branch"])
+        execution_digest = str(raw["execution_digest"])
+        if not IDEMPOTENCY_RE.fullmatch(idempotency_key):
+            raise TaskAdmissionError("TASK_ADMISSION_IDEMPOTENCY_KEY_INVALID")
+        if not TASK_RE.fullmatch(task_id):
+            raise TaskAdmissionError("TASK_ADMISSION_TASK_ID_INVALID")
+        if repository != self.policy.repository:
+            raise TaskAdmissionError("TASK_ADMISSION_REPOSITORY_REJECTED", 403)
+        if not OID_RE.fullmatch(authority_base_oid):
+            raise TaskAdmissionError("TASK_ADMISSION_BASE_OID_INVALID")
+        if (
+            not BRANCH_RE.fullmatch(target_branch)
+            or not target_branch.startswith(self.policy.branch_prefix)
+            or "//" in target_branch
+            or ".." in target_branch
+            or "@{" in target_branch
+            or target_branch.endswith(("/", "."))
+            or any(
+                not component
+                or component.startswith(".")
+                or component.endswith(".lock")
+                for component in target_branch.split("/")
+            )
+        ):
+            raise TaskAdmissionError("TASK_ADMISSION_BRANCH_REJECTED", 403)
+        generation = raw["generation"]
+        if not isinstance(generation, int) or isinstance(generation, bool):
+            raise TaskAdmissionError("TASK_ADMISSION_GENERATION_INVALID")
+        if generation < 1:
+            raise TaskAdmissionError("TASK_ADMISSION_GENERATION_INVALID")
+        if not re.fullmatch(r"^[0-9a-f]{64}$", execution_digest):
+            raise TaskAdmissionError("TASK_ADMISSION_EXECUTION_DIGEST_INVALID")
+        requested_capabilities = _normalize_capabilities(
+            raw["requested_capabilities"],
+            policy=self.policy,
+        )
+        file_allowlist = _normalize_files(
+            raw["file_allowlist"],
+            maximum=self.policy.max_file_count,
+        )
+        risk = str(raw["risk"])
+        if _risk_rank(risk) > _risk_rank(self.policy.max_risk):
+            raise TaskAdmissionError("TASK_ADMISSION_RISK_OUT_OF_POLICY", 403)
+        ttl = raw.get("expires_in_seconds", self.policy.max_envelope_ttl_seconds)
+        if (
+            not isinstance(ttl, int)
+            or isinstance(ttl, bool)
+            or ttl < 60
+            or ttl > self.policy.max_envelope_ttl_seconds
+        ):
+            raise TaskAdmissionError("TASK_ADMISSION_ENVELOPE_TTL_INVALID")
+        now = int(self.clock())
+        policy_digest = self.policy.normalized_policy_digest()
+        request_value = _request_for_digest(
+            task_id=task_id,
+            repository=repository,
+            authority_base_oid=authority_base_oid,
+            target_branch=target_branch,
+            generation=generation,
+            execution_digest=execution_digest,
+            requested_capabilities=requested_capabilities,
+            file_allowlist=file_allowlist,
+            risk=risk,
+            trust_domain=self.policy.trust_domain,
+            policy_digest=policy_digest,
+        )
+        request_digest = sha256_json(request_value)
+        nonce = secrets.token_urlsafe(24)
+        envelope_id = "task-admission-" + hashlib.sha256(
+            (idempotency_key + "\0" + request_digest + "\0" + nonce).encode("utf-8")
+        ).hexdigest()[:32]
+        envelope = TaskAdmissionEnvelope(
+            envelope_id=envelope_id,
+            idempotency_key=idempotency_key,
+            task_id=task_id,
+            repository=repository,
+            authority_base_oid=authority_base_oid,
+            target_branch=target_branch,
+            generation=generation,
+            execution_digest=execution_digest,
+            requested_capabilities=requested_capabilities,
+            file_allowlist=file_allowlist,
+            risk=risk,
+            trust_domain=self.policy.trust_domain,
+            policy_digest=policy_digest,
+            issued_at=now,
+            not_before=now,
+            expires_at=now + ttl,
+            nonce=nonce,
+            request_digest=request_digest,
+        )
+        envelope_dict = envelope.as_dict()
+        persisted = self.store.save_envelope(
+            idempotency_key=idempotency_key,
+            envelope_id=envelope_id,
+            request_digest=request_digest,
+            envelope_json=canonical_json_bytes(envelope_dict).decode("utf-8"),
+            created_at=now,
+        )
+        won = persisted["envelope_id"] == envelope_id
+        return {
+            "schema": SCHEMA_ENVELOPE,
+            "envelope": persisted,
+            "canonical_payload_base64": base64.b64encode(
+                canonical_json_bytes(persisted)
+            ).decode("ascii"),
+            "idempotent": not won,
+        }
+
+    def _load_envelope(self, envelope_id: str) -> TaskAdmissionEnvelope:
+        row = self.store.get_envelope(envelope_id)
+        if row is None:
+            raise TaskAdmissionError("TASK_ADMISSION_ENVELOPE_NOT_FOUND", 404)
+        raw = json.loads(row["envelope_json"])
+        return TaskAdmissionEnvelope(
+            **{
+                **raw,
+                "requested_capabilities": tuple(raw["requested_capabilities"]),
+                "file_allowlist": tuple(raw["file_allowlist"]),
+            }
+        )
+
+    def _verify_epoch(
+        self,
+        *,
+        epoch_id: str,
+        envelope: TaskAdmissionEnvelope,
+        now: int,
+    ) -> OwnerDelegationEpoch:
+        epoch = self.authorization.verify_delegation_epoch(epoch_id)
+        if epoch.schema != SCHEMA_EPOCH:
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_SCHEMA_INVALID", 403)
+        if epoch.epoch_id != epoch_id:
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_ID_MISMATCH", 403)
+        if (
+            epoch.trust_domain != envelope.trust_domain
+            or epoch.repository != envelope.repository
+            or epoch.policy_digest != envelope.policy_digest
+        ):
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_BINDING_MISMATCH", 403)
+        if epoch.revocation_epoch != 0:
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_REVOKED", 403)
+        if not epoch.not_before <= now < epoch.expires_at:
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_EXPIRED", 403)
+        if not any(
+            envelope.target_branch.startswith(prefix)
+            for prefix in epoch.allowed_branch_prefixes
+        ):
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_BRANCH_REJECTED", 403)
+        caps = set(envelope.requested_capabilities)
+        if not caps.issubset(epoch.allowed_capabilities):
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_CAPABILITY_WIDENING", 403)
+        if _risk_rank(envelope.risk) > _risk_rank(epoch.risk_ceiling):
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_RISK_WIDENING", 403)
+        for path in envelope.file_allowlist:
+            if not any(
+                path == prefix.rstrip("/")
+                or path.startswith(prefix.rstrip("/") + "/")
+                for prefix in epoch.allowed_file_prefixes
+            ):
+                raise TaskAdmissionError("TASK_ADMISSION_EPOCH_FILE_WIDENING", 403)
+        return epoch
+
+    def authorize_and_admit(self, raw: Mapping[str, Any]) -> dict[str, Any]:
+        required = {"envelope_id", "authorization_kind"}
+        allowed = required | {
+            "delegation_epoch_id",
+            "owner_key_id",
+            "signature_base64",
+        }
+        if set(raw) - allowed or not required.issubset(raw):
+            raise TaskAdmissionError("TASK_ADMISSION_AUTH_FIELDS_INVALID")
+        envelope_id = str(raw["envelope_id"])
+        with self.store.admission_lock(envelope_id):
+            # Every replay revalidates policy, temporal bounds, revocation and
+            # proof after the crash-releasing lock is owned.
+            envelope = self._load_envelope(envelope_id)
+            now = int(self.clock())
+            if envelope.policy_digest != self.policy.normalized_policy_digest():
+                raise TaskAdmissionError("TASK_ADMISSION_POLICY_DIGEST_DRIFT", 403)
+            if now < envelope.not_before or now >= envelope.expires_at:
+                raise TaskAdmissionError("TASK_ADMISSION_ENVELOPE_EXPIRED", 403)
+
+            kind = str(raw["authorization_kind"])
+            parent_epoch: OwnerDelegationEpoch | None = None
+            if kind == "delegation_epoch":
+                epoch_id = str(raw.get("delegation_epoch_id") or "")
+                if not epoch_id:
+                    raise TaskAdmissionError("TASK_ADMISSION_EPOCH_REQUIRED")
+                parent_epoch = self._verify_epoch(epoch_id=epoch_id, envelope=envelope, now=now)
+                owner_identity = parent_epoch.owner_identity
+                parent_epoch_id: str | None = parent_epoch.epoch_id
+            elif kind == "owner_signature":
+                key_id = str(raw.get("owner_key_id") or "")
+                signature = str(raw.get("signature_base64") or "")
+                if not KEY_ID_RE.fullmatch(key_id) or not signature:
+                    raise TaskAdmissionError("TASK_ADMISSION_OWNER_SIGNATURE_REQUIRED")
+                owner_identity = self.authorization.verify_owner_signature(
+                    key_id=key_id,
+                    canonical_payload=canonical_json_bytes(envelope.as_dict()),
+                    signature_b64=signature,
+                )
+                if not isinstance(owner_identity, str) or not owner_identity:
+                    raise TaskAdmissionError("TASK_ADMISSION_OWNER_SIGNATURE_INVALID", 403)
+                parent_epoch_id = None
+            else:
+                raise TaskAdmissionError("TASK_ADMISSION_AUTHORIZATION_KIND_REJECTED", 403)
+
+            authorization_digest = sha256_json(dict(raw))
+            grant_id = "task-grant-" + hashlib.sha256(
+                (envelope.envelope_id + "\0" + kind + "\0" + owner_identity + "\0" + (parent_epoch_id or "")).encode("utf-8")
+            ).hexdigest()[:32]
+            grant = TaskAdmissionGrant(
+                grant_id=grant_id, envelope_id=envelope.envelope_id, task_id=envelope.task_id,
+                repository=envelope.repository, authority_base_oid=envelope.authority_base_oid,
+                target_branch=envelope.target_branch, generation=envelope.generation,
+                execution_digest=envelope.execution_digest, capabilities=envelope.requested_capabilities,
+                file_allowlist=envelope.file_allowlist, risk=envelope.risk,
+                trust_domain=envelope.trust_domain, policy_digest=envelope.policy_digest,
+                authorization_kind=kind, owner_identity=owner_identity,
+                parent_epoch_id=parent_epoch_id, issued_at=now,
+                expires_at=min(
+                    envelope.expires_at,
+                    now + self.policy.max_task_ttl_seconds,
+                    parent_epoch.expires_at if parent_epoch is not None else envelope.expires_at,
+                ),
+            )
+            persisted_grant, _ = self.store.claim_grant(
+                envelope_id=envelope_id, grant=grant,
+                authorization_digest=authorization_digest, created_at=now,
+            )
+            if now >= persisted_grant.expires_at:
+                raise TaskAdmissionError("TASK_ADMISSION_GRANT_EXPIRED", 403)
+            if parent_epoch is not None and persisted_grant.expires_at > parent_epoch.expires_at:
+                raise TaskAdmissionError("TASK_ADMISSION_GRANT_PARENT_EXPIRY_CONFLICT", 409)
+            existing_receipt = self.store.receipt(envelope_id)
+            if existing_receipt is not None:
+                return {**existing_receipt, "idempotent": True}
+
+            try:
+                result = self.task_port.admit(persisted_grant)
+            except TaskAdmissionError:
+                raise
+            except Exception as exc:
+                raise TaskAdmissionError("TASK_ADMISSION_TASK_STATE_UNKNOWN", 503) from exc
+            if not isinstance(result, Mapping):
+                raise TaskAdmissionError("TASK_ADMISSION_PORT_RESPONSE_INVALID", 502)
+            state = str(result.get("state") or result.get("status") or "")
+            if str(result.get("task_id") or "") != persisted_grant.task_id:
+                raise TaskAdmissionError("TASK_ADMISSION_PORT_TASK_MISMATCH", 502)
+            if state not in {"QUEUED", "CLAIMED", "RUNNING"}:
+                raise TaskAdmissionError("TASK_ADMISSION_PORT_STATE_INVALID", 502, details={"state": state})
+            if self.post_admission is not None:
+                projection_result: Mapping[str, Any] = result
+                projection_reader = getattr(self.task_port, "projection_result", None)
+                if callable(projection_reader):
+                    try:
+                        candidate = projection_reader(persisted_grant, result)
+                    except TaskAdmissionError:
+                        raise
+                    except Exception as exc:
+                        raise TaskAdmissionError(
+                            "TASK_ADMISSION_EXECUTION_PROJECTION_FAILED", 503
+                        ) from exc
+                    if not isinstance(candidate, Mapping):
+                        raise TaskAdmissionError(
+                            "TASK_ADMISSION_EXECUTION_PROJECTION_INVALID", 503
+                        )
+                    projection_result = candidate
+                try:
+                    projection = self.post_admission(
+                        persisted_grant, projection_result
+                    )
+                except TaskAdmissionError:
+                    raise
+                except Exception as exc:
+                    raise TaskAdmissionError(
+                        "TASK_ADMISSION_EXECUTION_PROJECTION_FAILED", 503
+                    ) from exc
+                if projection is not None and not isinstance(projection, Mapping):
+                    raise TaskAdmissionError(
+                        "TASK_ADMISSION_EXECUTION_PROJECTION_INVALID", 503
+                    )
+            receipt = {
+                "schema": SCHEMA_RECEIPT, "envelope_id": persisted_grant.envelope_id,
+                "grant_id": persisted_grant.grant_id, "task_id": persisted_grant.task_id,
+                "task_state": state, "authorization_kind": persisted_grant.authorization_kind,
+                "owner_identity": persisted_grant.owner_identity, "parent_epoch_id": persisted_grant.parent_epoch_id,
+                "authority_base_oid": persisted_grant.authority_base_oid,
+                "target_branch": persisted_grant.target_branch, "generation": persisted_grant.generation,
+                "execution_digest": persisted_grant.execution_digest,
+                "grant_digest_sha256": sha256_json(persisted_grant.as_dict()),
+                "admitted_at": now, "idempotent": False,
+            }
+            return self.store.save_grant_and_receipt(
+                envelope_id=envelope_id, grant=persisted_grant, receipt=receipt, created_at=now,
+            )
+
+    def status(self, envelope_id: str) -> dict[str, Any]:
+        return self.store.status(envelope_id)

active-runtime/app/aios_task_admission_bridge_contract.py

@@ -0,0 +1,87 @@
+"""Bridge contract surface for Task Admission Bootstrap.
+
+This module is intentionally transport-agnostic. The live Bridge may merge
+these routes/operations into its existing authenticated/HTTPS surface, but the
+authorization proof for the authorize operation is the verified delegation
+epoch or Owner signature carried by the request itself.
+
+No handler here can issue TaskPrincipal assertions or WriterLeases. Those remain
+strictly downstream of a real admitted/claimed Agent Task.
+"""
+
+from __future__ import annotations
+
+from typing import Any, Mapping
+
+from aios_task_admission_bootstrap import TaskAdmissionBootstrapService
+
+
+TASK_ADMISSION_ROUTES = {
+    "/v2/task-admission/envelope": "getTaskAdmissionEnvelope",
+    "/v2/task-admission/authorize": "submitTaskAdmissionAuthorization",
+    "/v2/task-admission/status": "getTaskAdmissionStatus",
+}
+
+TASK_ADMISSION_OPERATION_REQUIRED_FIELDS = {
+    "getTaskAdmissionEnvelope": (
+        "idempotency_key",
+        "task_id",
+        "repository",
+        "authority_base_oid",
+        "target_branch",
+        "generation",
+        "execution_digest",
+        "requested_capabilities",
+        "file_allowlist",
+        "risk",
+    ),
+    "submitTaskAdmissionAuthorization": (
+        "envelope_id",
+        "authorization_kind",
+    ),
+    "getTaskAdmissionStatus": ("envelope_id",),
+}
+
+TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS = {
+    "getTaskAdmissionEnvelope": ("expires_in_seconds",),
+    "submitTaskAdmissionAuthorization": (
+        "delegation_epoch_id",
+        "owner_key_id",
+        "signature_base64",
+    ),
+    "getTaskAdmissionStatus": (),
+}
+
+
+class TaskAdmissionBridgeContract:
+    def __init__(self, service: TaskAdmissionBootstrapService) -> None:
+        self.service = service
+
+    @staticmethod
+    def _validate_fields(operation: str, request: Mapping[str, Any]) -> None:
+        required = set(TASK_ADMISSION_OPERATION_REQUIRED_FIELDS[operation])
+        allowed = required | set(TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS[operation])
+        if not required.issubset(request):
+            raise ValueError("TASK_ADMISSION_BRIDGE_REQUIRED_FIELDS_MISSING")
+        if set(request) - allowed:
+            raise ValueError("TASK_ADMISSION_BRIDGE_UNKNOWN_FIELDS")
+
+    def dispatch(self, operation: str, request: Mapping[str, Any]) -> dict[str, Any]:
+        if operation not in TASK_ADMISSION_OPERATION_REQUIRED_FIELDS:
+            raise ValueError("TASK_ADMISSION_BRIDGE_OPERATION_NOT_FOUND")
+        self._validate_fields(operation, request)
+        if operation == "getTaskAdmissionEnvelope":
+            return self.service.create_envelope(request)
+        if operation == "submitTaskAdmissionAuthorization":
+            return self.service.authorize_and_admit(request)
+        if operation == "getTaskAdmissionStatus":
+            return self.service.status(str(request["envelope_id"]))
+        raise ValueError("TASK_ADMISSION_BRIDGE_OPERATION_NOT_FOUND")
+
+    def handlers(self) -> dict[str, Any]:
+        return {
+            operation: (
+                lambda request, operation=operation: self.dispatch(operation, request)
+            )
+            for operation in TASK_ADMISSION_OPERATION_REQUIRED_FIELDS
+        }

active-runtime/app/aios_task_admission_live_adapters.py

@@ -0,0 +1,841 @@
+"""Narrow adapters for the already-authorized Owner and Agent Task seams.
+
+This file intentionally contains no credential lookup, HTTP client construction,
+or fallback identity.  The service host injects its authenticated durable-epoch
+provider and its service-managed Agent Task transport; a request body can never
+provide either authority.
+"""
+from __future__ import annotations
+
+import base64
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import stat
+import time
+from typing import Any, Callable, Mapping, Protocol
+import urllib.error
+import urllib.request
+
+from aios_task_admission_bootstrap import (
+    OwnerDelegationEpoch,
+    TaskAdmissionError,
+    TaskAdmissionGrant,
+    canonical_json_bytes,
+    sha256_json,
+)
+
+
+OWNER_NAMESPACE = "aios-owner-approval-v1"
+OWNER_SIGNER_IDENTITY = "aios-task-grant-signer"
+
+
+class OfficialTaskTransport(Protocol):
+    """Service-managed transport; it owns the Task API control credential."""
+
+    def request(self, method: str, path: str, body: Mapping[str, Any] | None = None) -> Mapping[str, Any]:
+        ...
+
+
+class BoundOwnerAuthorizationBackend:
+    """Bind epoch lookup to trusted transport identity, never caller JSON."""
+
+    def __init__(
+        self,
+        *,
+        authenticated_caller: Mapping[str, Any],
+        verified_epoch_provider: Callable[..., OwnerDelegationEpoch],
+        signature_verifier: Any,
+    ) -> None:
+        if not isinstance(authenticated_caller, Mapping) or not authenticated_caller:
+            raise ValueError("AUTHENTICATED_CALLER_REQUIRED")
+        self._caller = dict(authenticated_caller)
+        self._epoch_provider = verified_epoch_provider
+        self._signature_verifier = signature_verifier
+
+    def verify_delegation_epoch(self, epoch_id: str) -> OwnerDelegationEpoch:
+        try:
+            epoch = self._epoch_provider(epoch_id=epoch_id, caller=dict(self._caller))
+        except TypeError as exc:
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_BACKEND_INVALID", 503) from exc
+        if not isinstance(epoch, OwnerDelegationEpoch):
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_BACKEND_INVALID", 503)
+        return epoch
+
+    def verify_owner_signature(
+        self, *, key_id: str, canonical_payload: bytes, signature_b64: str
+    ) -> str:
+        try:
+            armored = base64.b64decode(signature_b64, validate=True).decode("utf-8")
+        except (ValueError, UnicodeDecodeError) as exc:
+            raise TaskAdmissionError("TASK_ADMISSION_OWNER_SIGNATURE_INVALID", 403) from exc
+        verify = getattr(self._signature_verifier, "verify", None)
+        if not callable(verify) or verify(
+            namespace=OWNER_NAMESPACE,
+            payload=canonical_payload,
+            signature=armored,
+            key_id=key_id,
+        ) is not True:
+            raise TaskAdmissionError("TASK_ADMISSION_OWNER_SIGNATURE_INVALID", 403)
+        return OWNER_SIGNER_IDENTITY
+
+
+class AgentTaskAdmissionPort:
+    """Official Task API adapter with exact identity reconciliation.
+
+    It never exposes or returns the API control bearer.  A timeout or 5xx is
+    reconciled by exact GET; if that cannot prove the state, the durable caller
+    receives UNKNOWN instead of a second blind POST.
+    """
+
+    def __init__(
+        self,
+        transport: OfficialTaskTransport,
+        *,
+        execution_plan_provider: Callable[[TaskAdmissionGrant], Mapping[str, Any]] | None = None,
+    ) -> None:
+        if not callable(getattr(transport, "request", None)):
+            raise ValueError("OFFICIAL_TASK_TRANSPORT_REQUIRED")
+        self._transport = transport
+        self._execution_plan_provider = execution_plan_provider
+
+    def _execution_plan(self, grant: TaskAdmissionGrant) -> dict[str, Any]:
+        if self._execution_plan_provider is None:
+            raise TaskAdmissionError("TASK_ADMISSION_EXECUTION_PLAN_REQUIRED", 503)
+        try:
+            raw = self._execution_plan_provider(grant)
+        except TaskAdmissionError:
+            raise
+        except Exception as exc:
+            raise TaskAdmissionError("TASK_ADMISSION_EXECUTION_PLAN_UNAVAILABLE", 503) from exc
+        if not isinstance(raw, Mapping):
+            raise TaskAdmissionError("TASK_ADMISSION_EXECUTION_PLAN_INVALID", 403)
+        expected = {
+            "repository": grant.repository,
+            "target_branch": grant.target_branch,
+            "generation": grant.generation,
+            "execution_digest": grant.execution_digest,
+        }
+        if any(canonical_json_bytes(raw.get(key)) != canonical_json_bytes(value) for key, value in expected.items()):
+            raise TaskAdmissionError("TASK_ADMISSION_EXECUTION_PLAN_CONFLICT", 409)
+        external_job_id = raw.get("external_job_id")
+        if not isinstance(external_job_id, str) or not external_job_id.strip():
+            raise TaskAdmissionError("TASK_ADMISSION_EXECUTION_PLAN_INVALID", 403)
+        # The trusted provider resolves the immutable execution payload by its
+        # signed digest. Preserve that payload: returning just coordinates
+        # would discard the executable plan needed by the real worker.
+        return dict(raw)
+
+    @staticmethod
+    def _metadata(grant: TaskAdmissionGrant, execution: Mapping[str, Any]) -> dict[str, Any]:
+        return {
+            "task_admission_schema": "aios.task-admission-grant.v1",
+            "grant_id": grant.grant_id,
+            "grant_digest_sha256": sha256_json(grant.as_dict()),
+            "repository": grant.repository,
+            "authority_base_oid": grant.authority_base_oid,
+            "target_branch": grant.target_branch,
+            "generation": grant.generation,
+            "execution_digest": grant.execution_digest,
+            "file_allowlist": list(grant.file_allowlist),
+            "capabilities": list(grant.capabilities),
+            "risk": grant.risk,
+            "execution": dict(execution),
+        }
+
+    @staticmethod
+    def _deadline(epoch: int) -> str:
+        # TaskStore.iso preserves the explicit UTC offset; exact reconciliation
+        # compares against that stored representation rather than a Z variant.
+        return datetime.fromtimestamp(epoch, timezone.utc).isoformat()
+
+    def _body(self, grant: TaskAdmissionGrant, execution: Mapping[str, Any]) -> dict[str, Any]:
+        metadata = self._metadata(grant, execution)
+        immutable = canonical_json_bytes(metadata).decode("utf-8")
+        return {
+            "task_id": grant.task_id,
+            "goal": "Execute the immutable governed task admission grant.",
+            "instructions": "Use only the admitted immutable metadata; execution plan digest: " + grant.execution_digest,
+            "success_criteria": ["Task processing preserves the admitted immutable identity."],
+            "required_capabilities": sorted(grant.capabilities),
+            "constraints": {
+                "repository": grant.repository,
+                "authority_base_oid": grant.authority_base_oid,
+                "target_branch": grant.target_branch,
+                "generation": grant.generation,
+                "file_allowlist": list(grant.file_allowlist),
+                "immutable_metadata": immutable,
+            },
+            "metadata": metadata,
+            "deadline": self._deadline(grant.expires_at),
+            "priority": 0,
+            "max_attempts": 1,
+            "idempotency_key": "task-admission/" + grant.grant_id,
+        }
+
+    def _exact(
+        self, task: Mapping[str, Any], grant: TaskAdmissionGrant, execution: Mapping[str, Any]
+    ) -> dict[str, Any]:
+        expected = self._body(grant, execution)
+        immutable_fields = (
+            "task_id", "goal", "instructions", "success_criteria",
+            "required_capabilities", "constraints", "metadata", "deadline",
+            "priority", "max_attempts",
+        )
+        def stable(field: str, value: Any) -> Any:
+            """Exclude only re-admission provenance, never execution scope."""
+            if field == "metadata" and isinstance(value, Mapping):
+                return {
+                    key: item
+                    for key, item in value.items()
+                    if key not in {"grant_id", "grant_digest_sha256"}
+                }
+            if field == "constraints" and isinstance(value, Mapping):
+                return {
+                    key: item
+                    for key, item in value.items()
+                    if key != "immutable_metadata"
+                }
+            return value
+
+        if any(
+            canonical_json_bytes(stable(field, task.get(field)))
+            != canonical_json_bytes(stable(field, expected[field]))
+            for field in immutable_fields
+        ):
+            raise TaskAdmissionError("TASK_ADMISSION_TASK_IDENTITY_CONFLICT", 409)
+        state = str(task.get("state") or task.get("status") or "")
+        if state not in {"QUEUED", "CLAIMED", "RUNNING"}:
+            raise TaskAdmissionError("TASK_ADMISSION_PORT_STATE_INVALID", 502, details={"state": state})
+        # The public admission port reports only authoritative Task identity
+        # and state.  The execution-plan identity is an internal capability
+        # used later by a trusted post-admission projection, never a caller
+        # visible part of this HTTP read model.
+        return {"task_id": grant.task_id, "state": state}
+
+    def _read(self, grant: TaskAdmissionGrant, execution: Mapping[str, Any]) -> dict[str, Any] | None:
+        try:
+            task = self._transport.request("GET", "/v1/tasks/" + grant.task_id)
+        except Exception as exc:
+            # A transport is expected to expose a status for normal 404s.  A
+            # generic failure is ambiguous and must not be treated as absent.
+            statuses = [
+                getattr(exc, attribute, None)
+                for attribute in ("status", "http_status", "code")
+                if getattr(exc, attribute, None) is not None
+            ]
+            if statuses and all(status == 404 for status in statuses):
+                return None
+            raise TaskAdmissionError("TASK_ADMISSION_TASK_STATE_UNKNOWN", 503) from exc
+        if not isinstance(task, Mapping):
+            raise TaskAdmissionError("TASK_ADMISSION_PORT_RESPONSE_INVALID", 502)
+        return self._exact(task, grant, execution)
+
+    def admit(self, grant: TaskAdmissionGrant) -> Mapping[str, Any]:
+        execution = self._execution_plan(grant)
+        existing = self._read(grant, execution)
+        if existing is not None:
+            return existing
+        try:
+            created = self._transport.request("POST", "/v1/tasks", self._body(grant, execution))
+        except Exception as exc:
+            recovered = self._read(grant, execution)
+            if recovered is not None:
+                return recovered
+            raise TaskAdmissionError("TASK_ADMISSION_TASK_STATE_UNKNOWN", 503) from exc
+        if not isinstance(created, Mapping):
+            raise TaskAdmissionError("TASK_ADMISSION_PORT_RESPONSE_INVALID", 502)
+        # Authoritative GET prevents a POST response from becoming the source
+        # of task identity/state and handles a task claimed before readback.
+        recovered = self._read(grant, execution)
+        if recovered is None:
+            raise TaskAdmissionError("TASK_ADMISSION_TASK_STATE_UNKNOWN", 503)
+        return recovered
+
+    def projection_result(
+        self,
+        grant: TaskAdmissionGrant,
+        admitted: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        """Return the server-side execution identity for a validated readback.
+
+        `admit()` intentionally keeps this identity out of its public result.
+        The bootstrap service calls this only for its local post-admission
+        projection; the value comes from the content-addressed execution-plan
+        provider, not from the HTTP caller or the Task API response.
+        """
+        if str(admitted.get("task_id") or "") != grant.task_id:
+            raise TaskAdmissionError("TASK_ADMISSION_PORT_TASK_MISMATCH", 502)
+        state = str(admitted.get("state") or admitted.get("status") or "")
+        if state not in {"QUEUED", "CLAIMED", "RUNNING"}:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_PORT_STATE_INVALID", 502, details={"state": state}
+            )
+        execution = self._execution_plan(grant)
+        external_job_id = execution.get("external_job_id")
+        if not isinstance(external_job_id, str) or not external_job_id.strip():
+            raise TaskAdmissionError("TASK_ADMISSION_EXECUTION_PLAN_INVALID", 403)
+        return {
+            "task_id": grant.task_id,
+            "state": state,
+            "external_job_id": external_job_id,
+        }
+
+
+class TaskAdmissionHostError(RuntimeError):
+    """Stable fail-closed host-composition error."""
+
+
+class OfficialTaskHTTPError(RuntimeError):
+    def __init__(self, status: int, body: object | None = None) -> None:
+        super().__init__(f"AGENT_TASK_HTTP_{status}")
+        self.status = status
+        self.http_status = status
+        self.code = status
+        self.body = body
+
+
+def _host_strict_object(pairs: list[tuple[str, Any]]) -> dict[str, Any]:
+    value: dict[str, Any] = {}
+    for key, item in pairs:
+        if key in value:
+            raise TaskAdmissionHostError("DUPLICATE_JSON_FIELD")
+        value[key] = item
+    return value
+
+
+def _host_read_json_object(path: Path) -> dict[str, Any]:
+    try:
+        value = json.loads(
+            path.read_text(encoding="utf-8"),
+            object_pairs_hook=_host_strict_object,
+        )
+    except (OSError, UnicodeError, json.JSONDecodeError) as exc:
+        raise TaskAdmissionHostError("JSON_INPUT_INVALID") from exc
+    if not isinstance(value, dict):
+        raise TaskAdmissionHostError("JSON_OBJECT_REQUIRED")
+    return value
+
+
+def _host_posix_metadata_enforced() -> bool:
+    """Return whether POSIX ownership/mode bits are security-significant here.
+
+    The live AIOS authority/runtime is Linux, where mode/uid checks are a real
+    trust boundary. Windows development/test files expose synthesized POSIX
+    mode bits that do not represent ACL authority, so interpreting ``022``
+    there would reject every secure temporary directory without improving
+    live security.
+    """
+    return os.name == "posix"
+
+
+def _host_require_regular(path: Path, *, root_owned: bool) -> None:
+    try:
+        info = path.lstat()
+    except OSError as exc:
+        raise TaskAdmissionHostError("SECURE_FILE_MISSING") from exc
+    if path.is_symlink() or not stat.S_ISREG(info.st_mode):
+        raise TaskAdmissionHostError("SECURE_FILE_INVALID")
+    if _host_posix_metadata_enforced():
+        if stat.S_IMODE(info.st_mode) & 0o022:
+            raise TaskAdmissionHostError("SECURE_FILE_WRITABLE_BY_GROUP_OR_OTHER")
+        if root_owned and info.st_uid != 0:
+            raise TaskAdmissionHostError("SECURE_FILE_NOT_ROOT_OWNED")
+
+
+def _host_require_directory(path: Path, *, root_owned: bool) -> None:
+    try:
+        info = path.lstat()
+    except OSError as exc:
+        raise TaskAdmissionHostError("SECURE_DIRECTORY_MISSING") from exc
+    if path.is_symlink() or not stat.S_ISDIR(info.st_mode):
+        raise TaskAdmissionHostError("SECURE_DIRECTORY_INVALID")
+    if _host_posix_metadata_enforced():
+        if stat.S_IMODE(info.st_mode) & 0o022:
+            raise TaskAdmissionHostError("SECURE_DIRECTORY_WRITABLE_BY_GROUP_OR_OTHER")
+        if root_owned and info.st_uid != 0:
+            raise TaskAdmissionHostError("SECURE_DIRECTORY_NOT_ROOT_OWNED")
+
+
+def load_owner_approval_verifier(module_path: Path) -> Any:
+    """Load the already-installed trusted public Owner verifier."""
+    _host_require_regular(module_path, root_owned=True)
+    spec = importlib.util.spec_from_file_location(
+        "aios_live_owner_approval_verifier",
+        module_path,
+    )
+    if spec is None or spec.loader is None:
+        raise TaskAdmissionHostError("OWNER_VERIFIER_IMPORT_INVALID")
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    cls = getattr(module, "GrantRootOwnerApprovalVerifier", None)
+    if cls is None:
+        raise TaskAdmissionHostError("OWNER_VERIFIER_CLASS_MISSING")
+    verifier = cls()
+    if not callable(getattr(verifier, "verify", None)):
+        raise TaskAdmissionHostError("OWNER_VERIFIER_INVALID")
+    return verifier
+
+
+@dataclass(frozen=True, slots=True)
+class SignedEpochDirectoryProvider:
+    """Re-verify the exact signed Owner epoch on every admission use."""
+
+    root: Path
+    verifier: Any
+    revocation_file: Path
+    expected_caller: Mapping[str, Any]
+    require_root_owner: bool = True
+    clock: Any = time.time
+
+    def _revoked_ids(self) -> set[str]:
+        _host_require_regular(
+            self.revocation_file,
+            root_owned=self.require_root_owner,
+        )
+        value = _host_read_json_object(self.revocation_file)
+        if (
+            set(value) != {"schema", "revoked_epoch_ids"}
+            or value.get("schema") != "aios.owner-delegation-revocations.v1"
+            or not isinstance(value.get("revoked_epoch_ids"), list)
+            or not all(
+                isinstance(item, str) and item
+                for item in value["revoked_epoch_ids"]
+            )
+        ):
+            raise TaskAdmissionHostError(
+                "OWNER_EPOCH_REVOCATION_REGISTRY_INVALID"
+            )
+        return set(value["revoked_epoch_ids"])
+
+    def __call__(
+        self,
+        *,
+        epoch_id: str,
+        caller: Mapping[str, Any],
+    ) -> OwnerDelegationEpoch:
+        if dict(caller) != dict(self.expected_caller):
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_CALLER_BINDING_MISMATCH",
+                403,
+            )
+        _host_require_directory(
+            self.root,
+            root_owned=self.require_root_owner,
+        )
+        canonical = self.root / "owner-delegation-epoch.canonical.json"
+        signature = self.root / "owner-delegation-epoch.sig"
+        key_id_file = self.root / "owner-delegation-epoch.key-id"
+        for path in (canonical, signature, key_id_file):
+            _host_require_regular(
+                path,
+                root_owned=self.require_root_owner,
+            )
+
+        payload = canonical.read_bytes()
+        value = _host_read_json_object(canonical)
+        if canonical_json_bytes(value) != payload:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EPOCH_CANONICAL_MISMATCH",
+                403,
+            )
+        if value.get("schema") != "aios.owner-delegation-epoch.v1":
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EPOCH_SCHEMA_INVALID",
+                403,
+            )
+        if value.get("epoch_id") != epoch_id:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EPOCH_ID_MISMATCH",
+                403,
+            )
+        if value.get("owner_identity") != OWNER_SIGNER_IDENTITY:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EPOCH_OWNER_IDENTITY_INVALID",
+                403,
+            )
+        if value.get("revocation_epoch") != 0:
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_REVOKED", 403)
+        if epoch_id in self._revoked_ids():
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_REVOKED", 403)
+
+        key_id = key_id_file.read_text(encoding="utf-8").strip()
+        armored = signature.read_text(encoding="utf-8")
+        if self.verifier.verify(
+            namespace=OWNER_NAMESPACE,
+            payload=payload,
+            signature=armored,
+            key_id=key_id,
+        ) is not True:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EPOCH_SIGNATURE_INVALID",
+                403,
+            )
+
+        now = int(self.clock())
+        not_before = value.get("not_before")
+        expires_at = value.get("expires_at")
+        if (
+            not isinstance(not_before, int)
+            or isinstance(not_before, bool)
+            or not isinstance(expires_at, int)
+            or isinstance(expires_at, bool)
+            or not not_before <= now < expires_at
+        ):
+            raise TaskAdmissionError("TASK_ADMISSION_EPOCH_EXPIRED", 403)
+
+        branches = value.get("allowed_branch_prefixes")
+        capabilities = value.get("allowed_capabilities")
+        files = value.get("allowed_file_prefixes")
+        if (
+            not isinstance(branches, list)
+            or not all(isinstance(item, str) and item for item in branches)
+            or not isinstance(capabilities, list)
+            or not all(
+                isinstance(item, str) and item
+                for item in capabilities
+            )
+            or not isinstance(files, list)
+            or not all(isinstance(item, str) and item for item in files)
+        ):
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EPOCH_SCOPE_INVALID",
+                403,
+            )
+
+        return OwnerDelegationEpoch(
+            epoch_id=str(value["epoch_id"]),
+            owner_identity=str(value["owner_identity"]),
+            trust_domain=str(value["trust_domain"]),
+            repository=str(value["repository"]),
+            allowed_branch_prefixes=tuple(branches),
+            allowed_capabilities=frozenset(capabilities),
+            allowed_file_prefixes=tuple(files),
+            risk_ceiling=str(value["risk_ceiling"]),
+            issued_at=int(value["issued_at"]),
+            not_before=not_before,
+            expires_at=expires_at,
+            revocation_epoch=int(value["revocation_epoch"]),
+            policy_digest=str(value["policy_digest"]),
+            schema=str(value["schema"]),
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ExecutionPlanDirectoryProvider:
+    """Resolve a trusted execution plan by its canonical SHA256 digest."""
+
+    root: Path
+    require_root_owner: bool = True
+
+    def __call__(self, grant: TaskAdmissionGrant) -> Mapping[str, Any]:
+        digest = str(grant.execution_digest)
+        if (
+            len(digest) != 64
+            or any(ch not in "0123456789abcdef" for ch in digest)
+        ):
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EXECUTION_PLAN_DIGEST_INVALID",
+                403,
+            )
+        _host_require_directory(
+            self.root,
+            root_owned=self.require_root_owner,
+        )
+        path = self.root / (digest + ".json")
+        _host_require_regular(
+            path,
+            root_owned=self.require_root_owner,
+        )
+        value = _host_read_json_object(path)
+        if hashlib.sha256(canonical_json_bytes(value)).hexdigest() != digest:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EXECUTION_PLAN_DIGEST_MISMATCH",
+                403,
+            )
+        expected = {
+            "task_id": grant.task_id,
+            "repository": grant.repository,
+            "target_branch": grant.target_branch,
+            "generation": grant.generation,
+        }
+        if any(value.get(key) != item for key, item in expected.items()):
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EXECUTION_PLAN_CONFLICT",
+                409,
+            )
+        external_job_id = value.get("external_job_id")
+        if not isinstance(external_job_id, str) or not external_job_id.strip():
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_EXECUTION_PLAN_INVALID",
+                403,
+            )
+        return {**value, "execution_digest": digest}
+
+
+@dataclass(frozen=True, slots=True)
+class LocalAgentTaskHTTPTransport:
+    """Server-side localhost Task API transport; token never enters caller JSON."""
+
+    base_url: str
+    token_file: Path
+    timeout_seconds: float = 5.0
+    require_root_owner: bool = True
+
+    def _token(self) -> str:
+        _host_require_regular(
+            self.token_file,
+            root_owned=self.require_root_owner,
+        )
+        token = self.token_file.read_text(encoding="utf-8").strip()
+        if len(token) < 16 or any(ch.isspace() for ch in token):
+            raise TaskAdmissionHostError("AGENT_TASK_CONTROL_TOKEN_INVALID")
+        return token
+
+    def request(
+        self,
+        method: str,
+        path: str,
+        body: Mapping[str, Any] | None = None,
+    ) -> Mapping[str, Any]:
+        if method not in {"GET", "POST"}:
+            raise TaskAdmissionHostError("AGENT_TASK_HTTP_METHOD_INVALID")
+        if not path.startswith("/v1/tasks"):
+            raise TaskAdmissionHostError("AGENT_TASK_HTTP_PATH_INVALID")
+        raw = None if body is None else canonical_json_bytes(dict(body))
+        request = urllib.request.Request(
+            self.base_url.rstrip("/") + path,
+            data=raw,
+            method=method,
+            headers={
+                "Accept": "application/json",
+                "Authorization": "Bearer " + self._token(),
+                **(
+                    {"Content-Type": "application/json"}
+                    if raw is not None
+                    else {}
+                ),
+            },
+        )
+        try:
+            with urllib.request.urlopen(
+                request,
+                timeout=self.timeout_seconds,
+            ) as response:
+                value = json.loads(
+                    response.read().decode("utf-8"),
+                    object_pairs_hook=_host_strict_object,
+                )
+        except urllib.error.HTTPError as exc:
+            body_value: object | None = None
+            try:
+                body_value = json.loads(exc.read().decode("utf-8"))
+            except Exception:
+                body_value = None
+            raise OfficialTaskHTTPError(
+                exc.code,
+                body_value,
+            ) from exc
+        if not isinstance(value, Mapping):
+            raise TaskAdmissionHostError("AGENT_TASK_HTTP_RESPONSE_INVALID")
+        return dict(value)
+
+
+def task_admission_policy_from_epoch(
+    epoch: OwnerDelegationEpoch,
+) -> Any:
+    """Build a local ceiling exactly narrowed to the signed epoch."""
+    from aios_task_admission_bootstrap import TaskAdmissionPolicy
+
+    return TaskAdmissionPolicy(
+        repository=epoch.repository,
+        trust_domain=epoch.trust_domain,
+        branch_prefix="agent/",
+        allowed_capabilities=frozenset(epoch.allowed_capabilities),
+        max_file_count=min(
+            64,
+            max(1, len(epoch.allowed_file_prefixes)),
+        ),
+        max_task_ttl_seconds=min(
+            3600,
+            max(60, epoch.expires_at - epoch.not_before),
+        ),
+        max_envelope_ttl_seconds=900,
+        max_risk=epoch.risk_ceiling,
+        policy_digest=epoch.policy_digest,
+    )
+
+
+def _build_task_admission_handlers(
+    *,
+    caller_service: str,
+    epoch_root: Path,
+    epoch_id: str,
+    revocation_file: Path,
+    execution_plan_dir: Path,
+    state_db: Path,
+    task_api_url: str,
+    task_control_token_file: Path,
+    owner_verifier_module: Path,
+    post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
+) -> Mapping[str, Any]:
+    """Compose handlers from already-trusted server-side coordinates."""
+    from aios_task_admission_bootstrap import (
+        TaskAdmissionBootstrapService,
+        TaskAdmissionStore,
+    )
+    from aios_task_admission_bridge_contract import TaskAdmissionBridgeContract
+
+    verifier = load_owner_approval_verifier(owner_verifier_module)
+    caller = {
+        "service": caller_service,
+        "trust_domain": "aios.owner",
+    }
+    provider = SignedEpochDirectoryProvider(
+        root=epoch_root,
+        verifier=verifier,
+        revocation_file=revocation_file,
+        expected_caller=caller,
+        require_root_owner=True,
+    )
+    epoch = provider(epoch_id=epoch_id, caller=caller)
+    authorization = BoundOwnerAuthorizationBackend(
+        authenticated_caller=caller,
+        verified_epoch_provider=provider,
+        signature_verifier=verifier,
+    )
+    task_port = AgentTaskAdmissionPort(
+        LocalAgentTaskHTTPTransport(
+            base_url=task_api_url,
+            token_file=task_control_token_file,
+            require_root_owner=True,
+        ),
+        execution_plan_provider=ExecutionPlanDirectoryProvider(
+            root=execution_plan_dir,
+            require_root_owner=True,
+        ),
+    )
+    service = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(state_db),
+        authorization=authorization,
+        task_port=task_port,
+        policy=task_admission_policy_from_epoch(epoch),
+        post_admission=post_admission,
+    )
+    return TaskAdmissionBridgeContract(service).handlers()
+
+
+def build_task_admission_handlers_from_activation_file(
+    activation_file: Path,
+    *,
+    caller_service: str = "aios-chatgpt-bridge-v2-staging.service",
+    post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
+) -> Mapping[str, Any]:
+    """Load one root-owned, fail-closed staging activation descriptor."""
+    _host_require_regular(activation_file, root_owned=True)
+    value = _host_read_json_object(activation_file)
+    required = {
+        "schema",
+        "epoch_root",
+        "epoch_id",
+        "revocation_file",
+        "execution_plan_dir",
+        "state_db",
+        "task_api_url",
+        "task_control_token_file",
+        "owner_verifier_module",
+    }
+    if set(value) != required or value.get("schema") != "aios.task-admission-host-activation.v1":
+        raise TaskAdmissionHostError("TASK_ADMISSION_ACTIVATION_FILE_INVALID")
+
+    path_fields = (
+        "epoch_root",
+        "revocation_file",
+        "execution_plan_dir",
+        "state_db",
+        "task_control_token_file",
+        "owner_verifier_module",
+    )
+    paths: dict[str, Path] = {}
+    for key in path_fields:
+        raw = value.get(key)
+        if not isinstance(raw, str) or not raw.startswith("/"):
+            raise TaskAdmissionHostError("TASK_ADMISSION_ACTIVATION_PATH_INVALID")
+        path = Path(raw)
+        if ".." in path.parts:
+            raise TaskAdmissionHostError("TASK_ADMISSION_ACTIVATION_PATH_INVALID")
+        paths[key] = path
+
+    epoch_id = value.get("epoch_id")
+    task_api_url = value.get("task_api_url")
+    if not isinstance(epoch_id, str) or not epoch_id:
+        raise TaskAdmissionHostError("TASK_ADMISSION_ACTIVATION_EPOCH_INVALID")
+    if task_api_url != "http://127.0.0.1:18940":
+        raise TaskAdmissionHostError("TASK_ADMISSION_ACTIVATION_TASK_API_INVALID")
+
+    return _build_task_admission_handlers(
+        caller_service=caller_service,
+        epoch_root=paths["epoch_root"],
+        epoch_id=epoch_id,
+        revocation_file=paths["revocation_file"],
+        execution_plan_dir=paths["execution_plan_dir"],
+        state_db=paths["state_db"],
+        task_api_url=task_api_url,
+        task_control_token_file=paths["task_control_token_file"],
+        owner_verifier_module=paths["owner_verifier_module"],
+        post_admission=post_admission,
+    )
+
+
+def build_task_admission_handlers_from_environment(
+    *,
+    caller_service: str = "aios-chatgpt-bridge-v2-staging.service",
+    post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
+) -> Mapping[str, Any]:
+    """Compose all three handlers from server-side trusted environment only."""
+    required = (
+        "AIOS_TASK_ADMISSION_EPOCH_ROOT",
+        "AIOS_TASK_ADMISSION_EPOCH_ID",
+        "AIOS_TASK_ADMISSION_REVOCATION_FILE",
+        "AIOS_TASK_ADMISSION_EXECUTION_PLAN_DIR",
+    )
+    missing = [
+        key
+        for key in required
+        if not os.environ.get(key, "").strip()
+    ]
+    if missing:
+        raise TaskAdmissionHostError(
+            "TASK_ADMISSION_HOST_ENV_MISSING:"
+            + ",".join(sorted(missing))
+        )
+    return _build_task_admission_handlers(
+        caller_service=caller_service,
+        epoch_root=Path(os.environ["AIOS_TASK_ADMISSION_EPOCH_ROOT"]),
+        epoch_id=os.environ["AIOS_TASK_ADMISSION_EPOCH_ID"],
+        revocation_file=Path(os.environ["AIOS_TASK_ADMISSION_REVOCATION_FILE"]),
+        execution_plan_dir=Path(os.environ["AIOS_TASK_ADMISSION_EXECUTION_PLAN_DIR"]),
+        state_db=Path(os.environ.get(
+            "AIOS_TASK_ADMISSION_STATE_DB",
+            "/var/lib/aios-gitops/bridge-v2-staging/task-admission.sqlite3",
+        )),
+        task_api_url=os.environ.get(
+            "AIOS_AGENT_TASK_API_URL",
+            "http://127.0.0.1:18940",
+        ),
+        task_control_token_file=Path(os.environ.get(
+            "AIOS_AGENT_TASK_CONTROL_TOKEN_FILE",
+            "/etc/aios-gitops/agent-task-control.token",
+        )),
+        owner_verifier_module=Path(os.environ.get(
+            "AIOS_OWNER_APPROVAL_VERIFIER_MODULE",
+            "/opt/aios/agent-task-api/"
+            "914eb324f657a695e2635f80cf53202eba636eb350630b12883e2741ce67e535"
+            "-p0-verification/runtime/app/aios_owner_approval_verifier.py",
+        )),
+        post_admission=post_admission,
+    )

active-runtime/app/trusted_builder_signer_provisioning.py

@@ -428,7 +428,7 @@ def _append(path: Path, row: Mapping[str, Any]) -> None:
     fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600)
     try:
         fcntl.flock(fd, fcntl.LOCK_EX)
-        os.write(fd, pretty_json(dict(row)))
+        os.write(fd, canonical_json(dict(row)) + b"\n")
         os.fsync(fd)
         fcntl.flock(fd, fcntl.LOCK_UN)
     finally:
@@ -437,17 +437,45 @@ def _append(path: Path, row: Mapping[str, Any]) -> None:
 
 
 def _receipts(path: Path) -> list[dict[str, Any]]:
-    if not path.is_file() or path.is_symlink():
+    if not path.exists() and not path.is_symlink():
         return []
+    if path.is_symlink() or not path.is_file():
+        raise ProvisioningError("RECEIPT_LOG_INVALID")
+    try:
+        text = path.read_text(encoding="utf-8")
+    except (OSError, UnicodeDecodeError) as exc:
+        raise ProvisioningError("RECEIPT_LOG_INVALID") from exc
+    if not text:
+        return []
+
+    # New receipts are canonical JSONL: exactly one object per physical line.
+    # If that framing does not parse, accept exactly one complete legacy object
+    # so the original pretty-printed receipt remains immutable and replayable.
+    lines = text.splitlines()
     result: list[dict[str, Any]] = []
-    for line in path.read_text(encoding="utf-8").splitlines():
+    jsonl_valid = bool(lines)
+    for line in lines:
+        if not line:
+            jsonl_valid = False
+            break
         try:
             value = json.loads(line)
         except json.JSONDecodeError:
-            continue
-        if isinstance(value, dict):
-            result.append(value)
-    return result
+            jsonl_valid = False
+            break
+        if not isinstance(value, dict):
+            jsonl_valid = False
+            break
+        result.append(value)
+    if jsonl_valid:
+        return result
+    try:
+        legacy = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise ProvisioningError("RECEIPT_LOG_INVALID") from exc
+    if not isinstance(legacy, dict):
+        raise ProvisioningError("RECEIPT_LOG_INVALID")
+    return [legacy]
 
 
 def _run_signature_verify(config: ProvisionerConfig, payload: bytes, signature: Path) -> None:
@@ -465,6 +493,26 @@ def _run_signature_verify(config: ProvisionerConfig, payload: bytes, signature:
         raise ProvisioningError("OWNER_SIGNATURE_VERIFY_FAILED")
 
 
+def _verify_service_policy(config: ProvisionerConfig) -> None:
+    if config.service_unit.is_symlink() or not config.service_unit.is_file():
+        raise ProvisioningError("BUILDER_SERVICE_UNIT_MISSING")
+    text = config.service_unit.read_text(encoding="utf-8")
+    required = (
+        "User=root", "LoadCredentialEncrypted=signing_private_key:",
+        "LoadCredentialEncrypted=builder_hmac:", "NoNewPrivileges=true",
+        "PrivateNetwork=true", "RestrictAddressFamilies=AF_UNIX",
+    )
+    if any(item not in text for item in required):
+        raise ProvisioningError("BUILDER_SERVICE_POLICY_INVALID")
+
+
+def _git(config: ProvisionerConfig, *args: str) -> subprocess.CompletedProcess:
+    return subprocess.run(
+        ["git", "--git-dir", str(config.authority_git_dir), *args],
+        capture_output=True, check=False, timeout=30,
+    )
+
+
 def _verify_source(config: ProvisionerConfig, source: Mapping[str, Any]) -> None:
     live_main = subprocess.run(
         ["git", "--git-dir", str(config.authority_git_dir), "rev-parse", "refs/heads/main"],
@@ -495,16 +543,38 @@ def _verify_source(config: ProvisionerConfig, source: Mapping[str, Any]) -> None
         actual = _digest_file(config.source_root / relative)
         if actual != expected:
             raise ProvisioningError("SOURCE_EVIDENCE_DIGEST_MISMATCH")
-    if config.service_unit.is_symlink() or not config.service_unit.is_file():
-        raise ProvisioningError("BUILDER_SERVICE_UNIT_MISSING")
-    text = config.service_unit.read_text(encoding="utf-8")
-    required = (
-        "User=root", "LoadCredentialEncrypted=signing_private_key:",
-        "LoadCredentialEncrypted=builder_hmac:", "NoNewPrivileges=true",
-        "PrivateNetwork=true", "RestrictAddressFamilies=AF_UNIX",
-    )
-    if any(item not in text for item in required):
-        raise ProvisioningError("BUILDER_SERVICE_POLICY_INVALID")
+    _verify_service_policy(config)
+
+
+def _verify_replay_source(config: ProvisionerConfig, source: Mapping[str, Any]) -> None:
+    """Verify the historical f8-bound source after authority main advances."""
+    commit = source["protected_merge_commit"]
+    live_main = _git(config, "rev-parse", "refs/heads/main")
+    tree = _git(config, "rev-parse", f"{commit}^{{tree}}")
+    parents = _git(config, "rev-list", "--parents", "-n", "1", commit)
+    source_ancestry = _git(config, "merge-base", "--is-ancestor", commit, live_main.stdout.strip().decode("ascii", "ignore"))
+    pr_ancestry = _git(config, "merge-base", "--is-ancestor", source["pr_head_oid"], commit)
+    expected_parents = [commit, *source["merge_parent_oids"]]
+    if (
+        live_main.returncode != 0
+        or tree.returncode != 0
+        or tree.stdout.strip().decode("ascii", "ignore").lower() != source["gitea_main_tree_oid"]
+        or parents.returncode != 0
+        or parents.stdout.decode("utf-8", "replace").strip().split() != expected_parents
+        or source_ancestry.returncode != 0
+        or pr_ancestry.returncode != 0
+    ):
+        raise ProvisioningError("AUTHORITY_HISTORICAL_SOURCE_DRIFT")
+    if source["merge_parent_oids"][1] != source["pr_head_oid"]:
+        raise ProvisioningError("SOURCE_PR_HEAD_SECOND_PARENT_INVALID")
+    for relative, expected in source["source_evidence_sha256"].items():
+        historical = _git(config, "show", f"{commit}:{relative}")
+        if historical.returncode != 0 or sha256_bytes(historical.stdout) != expected:
+            raise ProvisioningError("SOURCE_EVIDENCE_DIGEST_MISMATCH")
+    # The installed repair package must still satisfy the governed package and
+    # service policy, but its bytes are expected to differ from historical f8.
+    verify_package_contract(config.source_root)
+    _verify_service_policy(config)
 
 
 def _keyring(identity: Mapping[str, Any], public: bytes, now: dt.datetime) -> dict[str, Any]:
@@ -537,14 +607,158 @@ def _encrypt(config: ProvisionerConfig, plain: Path, target: Path, name: str) ->
     os.chmod(target, 0o600)
 
 
+def _require_private_regular_file(path: Path, code: str) -> None:
+    if path.is_symlink() or not path.is_file():
+        raise ProvisioningError(code)
+    metadata = path.stat()
+    if stat.S_IMODE(metadata.st_mode) != 0o600 or metadata.st_size <= 0:
+        raise ProvisioningError(code)
+
+
+def _read_state_json(path: Path, code: str) -> dict[str, Any]:
+    _require_private_regular_file(path, code)
+    try:
+        value = json.loads(path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ProvisioningError(code) from exc
+    if not isinstance(value, dict):
+        raise ProvisioningError(code)
+    return value
+
+
+def _parse_active_window(record: Mapping[str, Any]) -> None:
+    try:
+        not_before = dt.datetime.fromisoformat(str(record["not_before"]).replace("Z", "+00:00"))
+        expires_at = dt.datetime.fromisoformat(str(record["expires_at"]).replace("Z", "+00:00"))
+    except (KeyError, ValueError) as exc:
+        raise ProvisioningError("KEYRING_BINDING_INVALID") from exc
+    now = utc_now()
+    if not_before.tzinfo is None or expires_at.tzinfo is None or not (not_before <= now < expires_at):
+        raise ProvisioningError("KEYRING_BINDING_INVALID")
+
+
+def _verify_existing_state(
+    config: ProvisionerConfig,
+    payload: Mapping[str, Any],
+    raw: bytes,
+    signature_bytes: bytes,
+    row: Mapping[str, Any],
+) -> None:
+    receipt_keys = {
+        "schema", "status", "created_at", "operation_key", "payload_sha256",
+        "signature_sha256", "source_commit", "source_tree", "key_id",
+        "generation", "algorithm", "purpose", "service_principal",
+        "service_principal_bound", "public_key_sha256",
+        "private_key_exportable", "credential_projection",
+        "negative_capability_tests", "production_mutation",
+        "protected_ref_mutation", "tag_mutation", "baseline_mutation",
+    }
+    if set(row) != receipt_keys:
+        raise ProvisioningError("RECEIPT_CONFLICT")
+    try:
+        created_at = dt.datetime.fromisoformat(str(row["created_at"]).replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ProvisioningError("RECEIPT_CONFLICT") from exc
+    identity = payload["signing_identity"]
+    source = payload["source"]
+    expected = {
+        "schema": "aios.trusted-builder-signer-provisioning-receipt.v1",
+        "status": "COMPLETED",
+        "operation_key": "payload:" + sha256_bytes(raw) + ":signature:" + sha256_bytes(signature_bytes),
+        "payload_sha256": sha256_bytes(raw),
+        "signature_sha256": sha256_bytes(signature_bytes),
+        "source_commit": source["gitea_main_oid"],
+        "source_tree": source["gitea_main_tree_oid"],
+        "key_id": identity["key_id"],
+        "generation": identity["key_generation"],
+        "algorithm": "Ed25519",
+        "purpose": PURPOSE,
+        "service_principal": identity["service_principal"],
+        "service_principal_bound": True,
+        "private_key_exportable": False,
+        "credential_projection": "systemd-encrypted-host-credential",
+        "negative_capability_tests": {name: True for name in sorted(DENIED_CAPABILITIES)},
+        "production_mutation": False,
+        "protected_ref_mutation": False,
+        "tag_mutation": False,
+        "baseline_mutation": False,
+    }
+    if created_at.tzinfo is None or any(row.get(key) != value for key, value in expected.items()):
+        raise ProvisioningError("RECEIPT_CONFLICT")
+
+    public_keyring = _read_state_json(config.public_keyring, "KEYRING_BINDING_INVALID")
+    if set(public_keyring) != {"algorithm", "purpose", "keys", "keyring_sha256"}:
+        raise ProvisioningError("KEYRING_BINDING_INVALID")
+    keys = public_keyring.get("keys")
+    if not isinstance(keys, list) or len(keys) != 1 or not isinstance(keys[0], Mapping):
+        raise ProvisioningError("KEYRING_BINDING_INVALID")
+    record = dict(keys[0])
+    if set(record) != {
+        "algorithm", "purpose", "key_id", "generation", "not_before",
+        "expires_at", "status", "public_key", "public_key_sha256",
+    }:
+        raise ProvisioningError("KEYRING_BINDING_INVALID")
+    try:
+        public_key = base64.b64decode(str(record["public_key"]), validate=True)
+    except (ValueError, TypeError) as exc:
+        raise ProvisioningError("KEYRING_BINDING_INVALID") from exc
+    public_sha = sha256_bytes(public_key)
+    keyring_body = {key: public_keyring[key] for key in ("algorithm", "purpose", "keys")}
+    if (
+        len(public_key) != 32
+        or public_keyring["algorithm"] != "Ed25519"
+        or public_keyring["purpose"] != PURPOSE
+        or record.get("algorithm") != "Ed25519"
+        or record.get("purpose") != PURPOSE
+        or record.get("key_id") != identity["key_id"]
+        or record.get("generation") != identity["key_generation"]
+        or record.get("status") != "ACTIVE"
+        or record.get("public_key_sha256") != public_sha
+        or row.get("public_key_sha256") != public_sha
+        or public_keyring.get("keyring_sha256") != sha256_bytes(canonical_json(keyring_body))
+    ):
+        raise ProvisioningError("KEYRING_BINDING_INVALID")
+    _parse_active_window(record)
+
+    binding = _read_state_json(config.binding_file, "SERVICE_PRINCIPAL_BINDING_INVALID")
+    expected_binding = {
+        "schema": "aios.trusted-builder.service-principal-binding.v1",
+        "purpose": PURPOSE,
+        "service_principal": identity["service_principal"],
+        "key_id": identity["key_id"],
+        "generation": identity["key_generation"],
+        "algorithm": "Ed25519",
+        "private_key_export": False,
+        "public_key_sha256": public_sha,
+        "source_commit": source["gitea_main_oid"],
+        "source_tree": source["gitea_main_tree_oid"],
+        "payload_sha256": sha256_bytes(raw),
+    }
+    if binding != expected_binding:
+        raise ProvisioningError("SERVICE_PRINCIPAL_BINDING_INVALID")
+
+    if (
+        config.credential_store.is_symlink()
+        or not config.credential_store.is_dir()
+        or stat.S_IMODE(config.credential_store.stat().st_mode) != 0o700
+    ):
+        raise ProvisioningError("CREDENTIAL_STORE_INVALID")
+    _require_private_regular_file(
+        config.credential_store / "signing_private_key", "RECEIPT_OUTPUT_MISSING"
+    )
+    _require_private_regular_file(
+        config.credential_store / "builder_hmac", "RECEIPT_OUTPUT_MISSING"
+    )
+
+
 def provision(config: ProvisionerConfig, payload_path: Path, signature_path: Path) -> dict[str, Any]:
     if os.name != "nt" and os.geteuid() != 0:
         raise ProvisioningError("ROOT_REQUIRED")
     raw, payload = _read_json(payload_path)
     _validate_payload(payload)
     _run_signature_verify(config, raw, signature_path)
-    _verify_source(config, payload["source"])
-    operation_key = "payload:" + sha256_bytes(raw) + ":signature:" + sha256_bytes(signature_path.read_bytes())
+    signature_bytes = signature_path.read_bytes()
+    operation_key = "payload:" + sha256_bytes(raw) + ":signature:" + sha256_bytes(signature_bytes)
     lock_path = config.receipt_log.with_suffix(".lock")
     _safe_parent(lock_path)
     with lock_path.open("a+b") as lock:
@@ -553,12 +767,24 @@ def provision(config: ProvisionerConfig, payload_path: Path, signature_path: Pat
         previous = [row for row in rows if row.get("operation_key") == operation_key]
         if previous:
             row = previous[-1]
-            if row.get("status") != "COMPLETED" or row.get("purpose") != PURPOSE or row.get("key_id") != payload["signing_identity"]["key_id"]:
+            if any(canonical_json(item) != canonical_json(row) for item in previous):
                 raise ProvisioningError("RECEIPT_CONFLICT")
-            for required in (config.public_keyring, config.binding_file, config.credential_store / "signing_private_key", config.credential_store / "builder_hmac"):
-                if required.is_symlink() or not required.is_file():
-                    raise ProvisioningError("RECEIPT_OUTPUT_MISSING")
-            return {"status": "IDEMPOTENT_REPLAY_PASS", "idempotent": True, **{key: row[key] for key in ("operation_key", "payload_sha256", "signature_sha256", "key_id", "generation", "public_key_sha256", "purpose", "service_principal_bound", "private_key_exportable")}}
+            _verify_replay_source(config, payload["source"])
+            _verify_existing_state(config, payload, raw, signature_bytes, row)
+            return {
+                "status": "IDEMPOTENT_REPLAY_PASS",
+                "idempotent": True,
+                "mutation_executed": False,
+                "existing_signer_state_adopted": True,
+                **{key: row[key] for key in (
+                    "operation_key", "payload_sha256", "signature_sha256",
+                    "source_commit", "source_tree", "key_id", "generation",
+                    "public_key_sha256", "purpose", "service_principal_bound",
+                    "private_key_exportable",
+                )},
+                "negative_capability_tests": row["negative_capability_tests"],
+            }
+        _verify_source(config, payload["source"])
         outputs = [config.public_keyring, config.binding_file, config.credential_store / "signing_private_key", config.credential_store / "builder_hmac"]
         if any(path.exists() or path.is_symlink() for path in outputs):
             raise ProvisioningError("PREEXISTING_STATE_CONFLICT")
@@ -612,7 +838,7 @@ def provision(config: ProvisionerConfig, payload_path: Path, signature_path: Pat
                 "created_at": iso(now),
                 "operation_key": operation_key,
                 "payload_sha256": sha256_bytes(raw),
-                "signature_sha256": sha256_bytes(signature_path.read_bytes()),
+                "signature_sha256": sha256_bytes(signature_bytes),
                 "source_commit": payload["source"]["gitea_main_oid"],
                 "source_tree": payload["source"]["gitea_main_tree_oid"],
                 "key_id": payload["signing_identity"]["key_id"],
@@ -631,7 +857,7 @@ def provision(config: ProvisionerConfig, payload_path: Path, signature_path: Pat
                 "baseline_mutation": False,
             }
             _append(config.receipt_log, receipt)
-            return {"status": "PROVISIONED", "idempotent": False, **{key: receipt[key] for key in ("operation_key", "payload_sha256", "signature_sha256", "source_commit", "source_tree", "key_id", "generation", "purpose", "service_principal_bound", "public_key_sha256", "private_key_exportable")}, "negative_capability_tests": receipt["negative_capability_tests"]}
+            return {"status": "PROVISIONED", "idempotent": False, "mutation_executed": True, "existing_signer_state_adopted": False, **{key: receipt[key] for key in ("operation_key", "payload_sha256", "signature_sha256", "source_commit", "source_tree", "key_id", "generation", "purpose", "service_principal_bound", "public_key_sha256", "private_key_exportable")}, "negative_capability_tests": receipt["negative_capability_tests"]}
         finally:
             shutil.rmtree(temp_root, ignore_errors=True)
 

docs/architecture/adr-task-admission-bootstrap-p0.md

@@ -0,0 +1,102 @@
+# ADR: P0-0 Task Admission Bootstrap without weakening TaskPrincipal
+
+Date: 2026-09-03
+Status: Candidate for semantic port
+
+## Context
+
+The live control plane can expose governed writer operations while still being
+unable to create the Agent Task/TaskLease required by TaskPrincipal. This is a
+bootstrap deadlock, not a reason to enable global repository writes.
+
+The safety invariant is:
+
+```text
+Owner / durable delegation
+  -> Task admission grant
+  -> real Agent Task
+  -> claim + heartbeat + TaskLease
+  -> TaskPrincipal
+  -> WriterLease
+  -> governed commit
+```
+
+`GovernedTaskBootstrap.create()` may create a GitOps binding/Job. It is not, by
+itself, evidence of an Agent Task claim or TaskLease.
+
+## Decision
+
+Introduce a narrow Task Admission Bootstrap substrate with three operations:
+
+- `getTaskAdmissionEnvelope`
+- `submitTaskAdmissionAuthorization`
+- `getTaskAdmissionStatus`
+
+The envelope normalizes and binds repository, source OID, target branch,
+generation, execution digest, capabilities, file allowlist, risk, policy digest,
+expiry and nonce.
+
+Authorization is accepted only from:
+
+1. an already-verified `OwnerDelegationEpoch`; or
+2. an Owner signature over the exact canonical envelope.
+
+There is deliberately no permissive default authorization backend.
+
+The derived `TaskAdmissionGrant` can only narrow the parent authorization.
+Production, tag, baseline, main, global-write, force-push, runtime-anchor and
+arbitrary-shell capabilities are explicitly rejected.
+
+The grant is passed to the existing Agent Task control plane via an injected
+`TaskAdmissionPort`. This component never writes the Task database.
+
+The production adapter accepts only a service-managed official Task transport;
+it never returns its control credential. It first reconciles the exact task
+identity, then POSTs the official `/v1/tasks` contract only if absent, and
+reconciles an ambiguous response before reporting `UNKNOWN`. Task receipts use
+actual `QUEUED`, `CLAIMED`, or `RUNNING` state, never a synthetic ready state.
+
+Durable epochs are resolved through an injected provider that receives the
+authenticated caller identity from trusted middleware. A body-supplied epoch
+identifier is only a lookup key. The existing durable preparation authorization
+does not become file-write or TaskPrincipal authority.
+
+TaskPrincipal is not used to bootstrap the Task. It remains downstream of a
+real claim/TaskLease.
+
+## Crash / replay semantics
+
+Envelope creation is idempotent by exact request digest.
+
+Authorization/admission is durable and exact replay returns the same Task
+receipt. A SQLite `BEGIN IMMEDIATE` claim persists the immutable grant and its
+authorization-proof digest before the Task side effect. Cross-instance callers
+observe the durable envelope winner; a different proof cannot replace the
+stored grant. The live Task API retains the second exact-identity boundary.
+
+A crash after authorization but before Task API admission must resume the same
+envelope/grant identity. It must not create a replacement task or widen scope.
+
+## Integration rule for current live authority
+
+The source snapshot used for this candidate is not proven byte-identical to the
+2026-09-03 live authority. Therefore the code is a semantic-port candidate.
+
+The live integration must:
+
+- preserve existing TaskPrincipal verification;
+- adapt `TaskAdmissionPort` to the actual Agent Task API;
+- adapt `OwnerAuthorizationBackend` to the actual trusted Owner/delegation
+  verifier;
+- expose admission routes outside the TaskPrincipal gate, because the
+  cryptographic/delegation proof is the bootstrap authority;
+- never turn the runtime anchor into a writer;
+- never use Gitea/SSH/downstream credentials as Owner authority;
+- publish OpenAPI/capability metadata only when handlers are actually wired.
+
+## Consequences
+
+The control plane can become unattended after one durable Owner delegation
+epoch, while retaining fail-closed boundaries. If no valid delegation exists,
+the only legitimate stop is an already-generated canonical signing envelope
+awaiting the Owner's cryptographic signature.

docs/gitops/openapi-task-admission-bootstrap-fragment.yaml

@@ -0,0 +1,109 @@
+# AIOS Task Admission Bootstrap OpenAPI fragment.
+# Integrate into the canonical Bridge OpenAPI only when the corresponding live
+# handlers and authorization backend are configured. This fragment does not
+# enable legacy repository mutation or change TaskPrincipal/WriterLease gates.
+
+components:
+  schemas:
+    TaskAdmissionEnvelopeRequest:
+      type: object
+      additionalProperties: false
+      required:
+        - idempotency_key
+        - task_id
+        - repository
+        - authority_base_oid
+        - target_branch
+        - generation
+        - execution_digest
+        - requested_capabilities
+        - file_allowlist
+        - risk
+      properties:
+        idempotency_key: {type: string, minLength: 8, maxLength: 256}
+        task_id: {type: string, minLength: 3, maxLength: 180}
+        repository: {type: string, enum: [aios/aios]}
+        authority_base_oid: {type: string, pattern: '^[0-9a-f]{40}$'}
+        target_branch: {type: string, pattern: '^agent/.+'}
+        generation: {type: integer, minimum: 1}
+        execution_digest: {type: string, pattern: '^[0-9a-f]{64}$'}
+        requested_capabilities:
+          type: array
+          minItems: 1
+          uniqueItems: true
+          items: {type: string}
+        file_allowlist:
+          type: array
+          minItems: 1
+          uniqueItems: true
+          items: {type: string}
+        risk: {type: string, enum: [low, medium]}
+        expires_in_seconds:
+          type: integer
+          minimum: 60
+          maximum: 900
+
+    TaskAdmissionAuthorizationRequest:
+      type: object
+      additionalProperties: false
+      required: [envelope_id, authorization_kind]
+      properties:
+        envelope_id: {type: string}
+        authorization_kind:
+          type: string
+          enum: [delegation_epoch, owner_signature]
+        delegation_epoch_id: {type: string}
+        owner_key_id: {type: string}
+        signature_base64: {type: string}
+
+paths:
+  /v2/task-admission/envelope:
+    post:
+      operationId: getTaskAdmissionEnvelope
+      summary: Build and persist a canonical narrow Task admission envelope
+      description: >
+        No Git/Task mutation occurs. The server normalizes repository, branch,
+        capabilities, file allowlist, risk, policy digest, expiry and nonce.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema: {$ref: '#/components/schemas/TaskAdmissionEnvelopeRequest'}
+      responses:
+        '200': {description: Canonical envelope and signing payload}
+        '403': {description: Requested scope exceeds bootstrap policy}
+        '409': {description: Idempotency key conflicts with a different request}
+
+  /v2/task-admission/authorize:
+    post:
+      operationId: submitTaskAdmissionAuthorization
+      summary: Verify durable delegation or Owner signature and admit one exact Task
+      description: >
+        This operation does not accept a general-purpose bearer as authority.
+        Authorization is proven by a verified OwnerDelegationEpoch or by the
+        Owner signature over the exact canonical envelope. The resulting grant
+        can only narrow scope and is handed to the existing Agent Task API.
+        A delegation identifier in caller JSON is insufficient: the backend
+        binds the verified parent epoch to the authenticated service caller.
+      requestBody:
+        required: true
+        content:
+          application/json:
+            schema: {$ref: '#/components/schemas/TaskAdmissionAuthorizationRequest'}
+      responses:
+        '200': {description: Exact QUEUED, CLAIMED, or RUNNING Task receipt or idempotent replay}
+        '403': {description: Invalid, expired, revoked or widened authorization}
+        '409': {description: Conflicting idempotent admission}
+
+  /v2/task-admission/status:
+    get:
+      operationId: getTaskAdmissionStatus
+      summary: Read durable Task-admission state
+      parameters:
+        - in: query
+          name: envelope_id
+          required: true
+          schema: {type: string}
+      responses:
+        '200': {description: Pending or admitted receipt}
+        '404': {description: Envelope not found}

docs/gitops/task-admission-bootstrap-p0-runbook.md

@@ -0,0 +1,102 @@
+# Task Admission Bootstrap P0 Runbook
+
+## Goal
+
+Repair the Task-admission bootstrap chain without enabling legacy/global writes.
+
+## Preconditions
+
+Re-read live authority, runtime mode, OpenAPI, capabilities and current Task
+API. Do not assume the package source paths are current.
+
+Required invariants:
+
+```text
+AIOS_GITOPS_WRITES_ENABLED=false
+legacy_global_writes_enabled=false
+18893=IMMUTABLE_STAGING
+main/tag/baseline/production writes=false
+```
+
+## Semantic-port sequence
+
+1. Locate the live Agent Task create/admission implementation.
+2. Locate the current Owner/delegation signature verifier.
+3. Port `aios_task_admission_bootstrap.py` as the common substrate.
+4. Implement `TaskAdmissionPort.admit()` with the official Agent Task API.
+5. Implement `OwnerAuthorizationBackend` with the current trusted verifier.
+6. Wire the three task-admission operations into the existing Bridge.
+7. Keep TaskPrincipal issuance downstream of real Task claim/TaskLease.
+8. Merge the OpenAPI fragment only after the live handlers are wired.
+9. Run impacted Linux/POSIX, TaskPrincipal, Bridge and writer regression.
+10. Create the bootstrap-repair governed commit through the existing sanctioned
+    recovery path.
+11. Use the repaired admission path to create the Repository Discovery P0 Task.
+12. Claim/heartbeat -> TaskLease -> TaskPrincipal -> WriterLease.
+13. Commit only the existing Repository Discovery allowlist through
+    `commitFilesGoverned`.
+14. Reconcile ambiguous outcomes before any retry.
+    Admission uses a per-envelope, crash-releasing SQLite lock under the
+    admission state path's `.locks` directory.  It persists the immutable
+    grant before an effect, then the official Task API GET reconciles any
+    crashed or ambiguous effect before a POST is permitted.
+15. Replays revalidate the supplied proof, policy, expiry and revocation before
+    reading a receipt.  A different otherwise-valid proof conflicts with the
+    persisted authorization digest; it never becomes a harmless receipt replay.
+16. The Task API adapter must receive a server-resolved trusted execution plan
+    for the signed execution digest.  It supplies `metadata.execution` with
+    the exact repository, branch, generation, digest and nonempty external job
+    identifier.  Do not derive a job identity from request JSON or generate one.
+17. Require Authority / remote Git / API OID equality and idempotent replay.
+18. Publish immutable staging artifact and run Repository Discovery live E2E.
+
+## Never do
+
+- direct Task DB inserts/updates;
+- direct bare-repo/ref changes;
+- raw Git push as a substitute for governed publication;
+- borrow SSH/Gitea/service credentials as Owner authorization;
+- disable TaskPrincipal verification;
+- issue TaskPrincipal without a live TaskLease;
+- set global writes true;
+- turn 18893 into a writer;
+- include production/tag/baseline/main permissions in a task grant.
+
+## Acceptance
+
+Bootstrap repair:
+
+```text
+ENVELOPE_EXACT_REPLAY=PASS
+SCOPE_NARROWING=PASS
+DELEGATION_REVOKE_EXPIRY=PASS
+OWNER_SIGNATURE_VERIFY=PASS
+CONCURRENT_ADMISSION_SINGLE_TASK=PASS
+TASK_DB_DIRECT_WRITE=false
+TASKPRINCIPAL_BYPASS=false
+GLOBAL_WRITES=false
+CRASH_AFTER_EFFECT_REPLAY=PASS
+CRASH_BEFORE_EFFECT_REPLAY=PASS
+CHANGED_VALID_PROOF_CONFLICT=PASS
+TASK_API_EXACT_IDENTITY=PASS
+```
+
+These are implementation and test acceptance conditions only.  They are not
+evidence of a live claim, governed publication, remote OID equality, or staging
+E2E execution; obtain those independently through the sanctioned live runbook.
+
+Repository Discovery publication:
+
+```text
+TASK_CREATED=PASS
+TASK_CLAIMED=PASS
+TASK_HEARTBEAT=PASS
+TASK_LEASE=PASS
+TASK_PRINCIPAL=PASS
+WRITER_LEASE=PASS
+COMMIT_FILES_GOVERNED=PASS
+AUTHORITY_REMOTE_API_OID_EQUAL=PASS
+IDEMPOTENCY_REPLAY=PASS
+STAGING_LIVE_E2E=PASS
+TASK_VERIFIED=SUCCEEDED
+```

scripts/phase2_introspect_and_freeze.py

@@ -0,0 +1,543 @@
+"""Read-only Phase 2 runtime introspection with a bounded SSH transport.
+
+The wrapper is deliberately sent as ``subprocess.run(input=payload)``.  No
+JSON, ABI data, or base64 representation is placed in the Windows SSH argv;
+the default argv is the fixed, short command documented by
+``DEFAULT_REMOTE_COMMAND``.
+
+The remote wrapper reads the current issuer unit's MainPID, WorkingDirectory,
+and PYTHONPATH, then resolves only the named importable modules.  It does not
+walk release trees, ``/opt``, historical worktrees, SOUL, or MEMORY, and it
+does not call a database, signer, Bridge, Task API, or mutation endpoint.
+
+This module is locally testable, but the module import and systemd/POSIX
+behaviour still require external Linux live validation.  Phase 2 status is
+always explicit about all mutation classes being false.
+"""
+
+from __future__ import annotations
+
+import argparse
+import ast
+import hashlib
+import json
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any, Callable, Mapping, Sequence
+
+
+DEFAULT_SSH = "ssh.exe"
+DEFAULT_HOST = "hw"
+DEFAULT_REMOTE_COMMAND = 'test "$(hostname)" = cn-network-mkt && sudo -n python3 -'
+DEFAULT_TIMEOUT_SECONDS = 60
+DEFAULT_ISSUER_UNIT = "aios-task-principal-issuer.service"
+DEFAULT_ADMISSION_MODULE = "aios_task_principal_admission"
+DEFAULT_MIGRATION_MODULE = "aios_task_principal_storage_migration"
+DEFAULT_REQUIRED_ADMISSION_FUNCTIONS = (
+    "offline_migrate_task_principal_schema",
+    "supersede_p0_scope",
+)
+DEFAULT_MIGRATION_VERIFIER = "SSHOwnerBootstrapVerifier"
+MAX_ARGV_BYTES = 512
+MAX_REQUEST_BYTES = 32 * 1024
+
+MUTATION_FLAGS = (
+    "task_db_mutation",
+    "scope_mutation",
+    "git_mutation",
+    "service_mutation",
+)
+
+
+class RemoteTransportError(RuntimeError):
+    """A safe, non-secret description of an SSH transport failure."""
+
+
+class Phase2StatusError(RuntimeError):
+    """The remote result was not a valid fail-closed Phase2 status."""
+
+
+def build_ssh_argv(
+    ssh: str = DEFAULT_SSH,
+    host: str = DEFAULT_HOST,
+) -> list[str]:
+    """Return the only SSH argv shape permitted by the Phase2 transport."""
+
+    argv = [
+        ssh,
+        "-T",
+        "-o",
+        "BatchMode=yes",
+        "-o",
+        "ConnectTimeout=12",
+        host,
+        DEFAULT_REMOTE_COMMAND,
+    ]
+    argv_bytes = sum(len(part.encode("utf-8")) + 1 for part in argv)
+    if argv_bytes > MAX_ARGV_BYTES:
+        raise ValueError("SSH argv exceeds the Phase2 hard limit")
+    return argv
+
+
+def run_remote_python(
+    payload: bytes,
+    *,
+    ssh: str = DEFAULT_SSH,
+    host: str = DEFAULT_HOST,
+    timeout: float = DEFAULT_TIMEOUT_SECONDS,
+    runner: Callable[..., Any] = subprocess.run,
+) -> bytes:
+    """Send *payload* as stdin to the fixed remote Python command.
+
+    Error messages intentionally contain no stderr text.  SSH stderr can
+    include command output, environment values, or credentials emitted by a
+    misconfigured remote service.
+    """
+
+    if not isinstance(payload, bytes):
+        raise TypeError("payload must be bytes")
+    if timeout <= 0:
+        raise ValueError("timeout must be positive")
+
+    argv = build_ssh_argv(ssh=ssh, host=host)
+    try:
+        completed = runner(
+            argv,
+            input=payload,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            timeout=timeout,
+            check=False,
+        )
+    except subprocess.TimeoutExpired as exc:
+        raise RemoteTransportError(
+            f"remote SSH timed out after {timeout:g} seconds"
+        ) from exc
+    except OSError as exc:
+        raise RemoteTransportError("remote SSH could not be started") from exc
+
+    return_code = getattr(completed, "returncode", None)
+    if return_code != 0:
+        raise RemoteTransportError(f"remote SSH exited with status {return_code}")
+    output = getattr(completed, "stdout", b"")
+    if isinstance(output, str):
+        output = output.encode("utf-8")
+    if not isinstance(output, bytes):
+        raise RemoteTransportError("remote SSH returned a non-byte stdout value")
+    return output
+
+
+def module_sha256(path: str | Path) -> str:
+    """Hash one resolved module file without searching neighbouring files."""
+
+    digest = hashlib.sha256()
+    with Path(path).open("rb") as source:
+        for block in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+
+def validate_public_function_ast(
+    source: str,
+    required_functions: Sequence[str] = DEFAULT_REQUIRED_ADMISSION_FUNCTIONS,
+) -> dict[str, Any]:
+    """Validate public symbols and the source-level export mechanism.
+
+    The live admission overlay intentionally re-exports the legacy public API
+    through a small ``globals()[name] = value`` loop.  Requiring every public
+    function to be a direct ``def`` would therefore reject the exact deployed
+    module even though the runtime symbol and signature are present.  The AST
+    check still proves that the module has either a direct/imported binding or
+    an explicit dynamic re-export mechanism; the caller separately verifies
+    the runtime callable and its signature.
+    """
+
+    try:
+        tree = ast.parse(source)
+    except SyntaxError as exc:
+        return {
+            "ok": False,
+            "missing": list(required_functions),
+            "error": f"AST_PARSE_ERROR:{exc.msg}",
+        }
+
+    top_level = {
+        node.name: node
+        for node in tree.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and not node.name.startswith("_")
+    }
+    imported = {
+        alias.asname or alias.name.split(".", 1)[0]
+        for node in tree.body
+        if isinstance(node, (ast.Import, ast.ImportFrom))
+        for alias in node.names
+        if alias.name != "*"
+    }
+    dynamic_reexport = any(
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "globals"
+        for node in ast.walk(tree)
+    )
+    bound = set(top_level) | imported
+    missing = [
+        name
+        for name in required_functions
+        if name not in bound and not dynamic_reexport
+    ]
+    return {
+        "ok": not missing,
+        "missing": missing,
+        "functions": sorted(top_level),
+        "imported": sorted(imported),
+        "dynamic_reexport": dynamic_reexport,
+    }
+
+
+def make_phase2_status(
+    status: str,
+    errors: Sequence[str] = (),
+    **details: Any,
+) -> dict[str, Any]:
+    """Create a status object whose mutation flags are always false."""
+
+    result: dict[str, Any] = {
+        "kind": "phase2-status",
+        "version": 1,
+        "status": status,
+        "read_only": True,
+        "fail_closed": status != "PASS",
+    }
+    result.update({flag: False for flag in MUTATION_FLAGS})
+    if errors:
+        result["errors"] = list(errors)
+    result.update(details)
+    return result
+
+
+def validate_phase2_status(status: Mapping[str, Any]) -> dict[str, Any]:
+    """Validate the result contract and reject any mutation claim."""
+
+    if status.get("kind") != "phase2-status":
+        raise Phase2StatusError("remote result is not phase2-status")
+    for flag in MUTATION_FLAGS:
+        if status.get(flag) is not False:
+            raise Phase2StatusError(f"remote result has invalid {flag}")
+    if status.get("status") not in {"PASS", "FAIL"}:
+        raise Phase2StatusError("remote result has invalid status")
+    return dict(status)
+
+
+def _json_request(request: Mapping[str, Any]) -> str:
+    """Serialize only a small request into the stdin-delivered wrapper."""
+
+    encoded = json.dumps(dict(request), ensure_ascii=True, separators=(",", ":"))
+    if len(encoded.encode("utf-8")) > MAX_REQUEST_BYTES:
+        raise ValueError("Phase2 request is unexpectedly large")
+    return encoded
+
+
+REMOTE_SCRIPT_PREFIX = r'''# Phase2 read-only runtime wrapper.  Do not add mutations here.
+import ast
+import hashlib
+import importlib
+import importlib.util
+import inspect
+import json
+import os
+import shlex
+import subprocess
+import sys
+from pathlib import Path
+
+sys.dont_write_bytecode = True
+REQUEST = json.loads('''
+REMOTE_SCRIPT_SUFFIX = r''')
+
+MUTATION_FLAGS = ("task_db_mutation", "scope_mutation", "git_mutation", "service_mutation")
+
+def status(status_name, errors=(), **details):
+    result = {
+        "kind": "phase2-status",
+        "version": 1,
+        "status": status_name,
+        "read_only": True,
+        "fail_closed": status_name != "PASS",
+    }
+    result.update({flag: False for flag in MUTATION_FLAGS})
+    if errors:
+        result["errors"] = list(errors)
+    result.update(details)
+    return result
+
+def systemd_value(unit, property_name):
+    try:
+        completed = subprocess.run(
+            ["systemctl", "show", unit, "--property=" + property_name, "--value", "--no-pager"],
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+            check=False,
+            timeout=12,
+        )
+    except (OSError, subprocess.TimeoutExpired):
+        raise RuntimeError("SYSTEMD_QUERY_FAILED")
+    if completed.returncode != 0:
+        raise RuntimeError("SYSTEMD_QUERY_FAILED")
+    return completed.stdout.decode("utf-8", "strict").strip()
+
+def parse_pythonpath(environment_value):
+    for token in shlex.split(environment_value):
+        if token.startswith("PYTHONPATH="):
+            return token[len("PYTHONPATH="):].split(os.pathsep)
+    return []
+
+def runtime_identity():
+    unit = REQUEST.get("issuer_unit", "aios-task-principal-issuer.service")
+    pid = REQUEST.get("issuer_pid")
+    working_directory = REQUEST.get("working_directory")
+    pythonpath = REQUEST.get("pythonpath")
+    # Regression callers may pin one identity component while obtaining the
+    # remaining components from the live unit.  In particular, never replace
+    # an explicitly supplied PID with a newly queried MainPID.
+    if pid is None:
+        pid = int(systemd_value(unit, "MainPID"))
+    if working_directory is None:
+        working_directory = systemd_value(unit, "WorkingDirectory")
+    if pythonpath is None:
+        pythonpath = parse_pythonpath(systemd_value(unit, "Environment"))
+    if not pid or not working_directory or working_directory == "-":
+        raise RuntimeError("ISSUER_RUNTIME_IDENTITY_INCOMPLETE")
+    if isinstance(pythonpath, str):
+        pythonpath = pythonpath.split(os.pathsep)
+    return {
+        "unit": unit,
+        "main_pid": int(pid),
+        "working_directory": working_directory,
+        "pythonpath": [item for item in pythonpath if item],
+    }
+
+def setup_import_path(identity):
+    path = [identity["working_directory"]]
+    path.extend(identity["pythonpath"])
+    path.extend(item for item in sys.path if item and item not in path)
+    sys.path[:] = path
+
+def importable_module(module_name):
+    try:
+        spec = importlib.util.find_spec(module_name)
+        if spec is None or not spec.origin or spec.origin in {"built-in", "frozen"}:
+            raise RuntimeError("MODULE_NOT_IMPORTABLE:" + module_name)
+        module = importlib.import_module(module_name)
+    except Exception as exc:
+        if isinstance(exc, RuntimeError):
+            raise
+        raise RuntimeError("MODULE_IMPORT_FAILED:" + module_name) from exc
+    module_path = getattr(module, "__file__", None)
+    if not module_path:
+        raise RuntimeError("MODULE_HAS_NO_FILE:" + module_name)
+    return module, Path(module_path).resolve()
+
+def hash_file(path):
+    digest = hashlib.sha256()
+    with path.open("rb") as source:
+        for block in iter(lambda: source.read(1024 * 1024), b""):
+            digest.update(block)
+    return digest.hexdigest()
+
+def inspect_admission(module, path, required, expected_hash):
+    source = path.read_text(encoding="utf-8")
+    tree = ast.parse(source)
+    nodes = {
+        node.name: node for node in tree.body
+        if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
+        and not node.name.startswith("_")
+    }
+    imported = {
+        alias.asname or alias.name.split(".", 1)[0]
+        for node in tree.body
+        if isinstance(node, (ast.Import, ast.ImportFrom))
+        for alias in node.names
+        if alias.name != "*"
+    }
+    dynamic_reexport = any(
+        isinstance(node, ast.Call)
+        and isinstance(node.func, ast.Name)
+        and node.func.id == "globals"
+        for node in ast.walk(tree)
+    )
+    bound = set(nodes) | imported
+    missing = [
+        name
+        for name in required
+        if name not in bound and not dynamic_reexport
+    ]
+    if missing:
+        raise RuntimeError("PUBLIC_SYMBOL_MISSING:" + ",".join(missing))
+    signatures = {}
+    for name in required:
+        value = getattr(module, name, None)
+        if value is None:
+            raise RuntimeError("PUBLIC_SYMBOL_NOT_IMPORTABLE:" + name)
+        try:
+            signatures[name] = str(inspect.signature(value))
+        except (TypeError, ValueError):
+            raise RuntimeError("SIGNATURE_UNAVAILABLE:" + name)
+    actual_hash = hash_file(path)
+    if expected_hash and actual_hash != expected_hash:
+        raise RuntimeError("MODULE_HASH_MISMATCH:admission")
+    return {
+        "name": module.__name__,
+        "path": str(path),
+        "sha256": actual_hash,
+        "signatures": signatures,
+        "ast_functions": sorted(nodes),
+        "ast_imported": sorted(imported),
+        "ast_dynamic_reexport": dynamic_reexport,
+    }
+
+def inspect_migration(module, path, verifier_name, expected_hash):
+    verifier = getattr(module, verifier_name, None)
+    if verifier is None:
+        raise RuntimeError("MIGRATION_VERIFIER_MISSING:" + verifier_name)
+    try:
+        signature = str(inspect.signature(verifier))
+    except (TypeError, ValueError):
+        raise RuntimeError("MIGRATION_VERIFIER_SIGNATURE_UNAVAILABLE")
+    actual_hash = hash_file(path)
+    if expected_hash and actual_hash != expected_hash:
+        raise RuntimeError("MODULE_HASH_MISMATCH:migration")
+    return {
+        "name": module.__name__,
+        "path": str(path),
+        "sha256": actual_hash,
+        "verifier": verifier_name,
+        "signature": signature,
+    }
+
+def main():
+    try:
+        identity = runtime_identity()
+        setup_import_path(identity)
+        admission_name = REQUEST.get("admission_module", "aios_task_principal_admission")
+        migration_name = REQUEST.get("migration_module", "aios_task_principal_storage_migration")
+        if migration_name == admission_name:
+            raise RuntimeError("MIGRATION_MODULE_NOT_DISTINCT")
+        required = tuple(REQUEST.get("required_admission_functions", (
+            "offline_migrate_task_principal_schema", "supersede_p0_scope")))
+        admission, admission_path = importable_module(admission_name)
+        migration, migration_path = importable_module(migration_name)
+        admission_info = inspect_admission(
+            admission,
+            admission_path,
+            required,
+            REQUEST.get("expected_admission_sha256"),
+        )
+        migration_info = inspect_migration(
+            migration,
+            migration_path,
+            REQUEST.get("migration_verifier", "SSHOwnerBootstrapVerifier"),
+            REQUEST.get("expected_migration_sha256"),
+        )
+        print(json.dumps(status(
+            "PASS",
+            issuer=identity,
+            modules={"admission": admission_info, "migration": migration_info},
+            ast_public_functions=list(required),
+        ), sort_keys=True, separators=(",", ":")))
+    except Exception as exc:
+        # Emit only stable error classes and never exception text from imports.
+        error = str(exc).split(":", 1)[0] or "PHASE2_INTROSPECTION_FAILED"
+        print(json.dumps(status("FAIL", errors=[error]), sort_keys=True, separators=(",", ":")))
+
+if __name__ == "__main__":
+    main()
+'''
+
+
+def build_remote_payload(request: Mapping[str, Any] | None = None) -> bytes:
+    """Build the stdin wrapper; request data never enters SSH argv."""
+
+    request = {} if request is None else request
+    # repr() makes the embedded JSON a valid Python string for every request
+    # value, including apostrophes and backslashes; it still remains stdin.
+    request_literal = repr(_json_request(request))
+    return (REMOTE_SCRIPT_PREFIX + request_literal + REMOTE_SCRIPT_SUFFIX).encode("utf-8")
+
+
+def _load_request(args: argparse.Namespace) -> dict[str, Any]:
+    request: dict[str, Any] = {}
+    if args.request_json:
+        loaded = json.loads(args.request_json)
+        if not isinstance(loaded, dict):
+            raise ValueError("request JSON must be an object")
+        request.update(loaded)
+    if args.request_file:
+        loaded = json.loads(Path(args.request_file).read_text(encoding="utf-8"))
+        if not isinstance(loaded, dict):
+            raise ValueError("request JSON must be an object")
+        request.update(loaded)
+
+    overrides = {
+        "issuer_unit": args.issuer_unit,
+        "issuer_pid": args.issuer_pid,
+        "working_directory": args.working_directory,
+        "pythonpath": args.pythonpath,
+        "admission_module": args.admission_module,
+        "migration_module": args.migration_module,
+        "expected_admission_sha256": args.expected_admission_sha256,
+        "expected_migration_sha256": args.expected_migration_sha256,
+    }
+    request.update({key: value for key, value in overrides.items() if value is not None})
+    return request
+
+
+def _write_status(path: str | Path, status: Mapping[str, Any]) -> None:
+    destination = Path(path)
+    destination.write_text(
+        json.dumps(dict(status), ensure_ascii=True, indent=2, sort_keys=True) + "\n",
+        encoding="utf-8",
+    )
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--request-json")
+    parser.add_argument("--request-file")
+    parser.add_argument("--output", default="phase2-status.json")
+    parser.add_argument("--ssh", default=DEFAULT_SSH)
+    parser.add_argument("--host", default=DEFAULT_HOST)
+    parser.add_argument("--timeout", type=float, default=DEFAULT_TIMEOUT_SECONDS)
+    parser.add_argument("--issuer-unit")
+    parser.add_argument("--issuer-pid", type=int)
+    parser.add_argument("--working-directory")
+    parser.add_argument("--pythonpath", nargs="*")
+    parser.add_argument("--admission-module")
+    parser.add_argument("--migration-module")
+    parser.add_argument("--expected-admission-sha256")
+    parser.add_argument("--expected-migration-sha256")
+    args = parser.parse_args(argv)
+
+    try:
+        request = _load_request(args)
+        payload = build_remote_payload(request)
+        stdout = run_remote_python(
+            payload,
+            ssh=args.ssh,
+            host=args.host,
+            timeout=args.timeout,
+        )
+        remote_status = json.loads(stdout.decode("utf-8"))
+        status = validate_phase2_status(remote_status)
+    except (OSError, ValueError, json.JSONDecodeError, RemoteTransportError, Phase2StatusError) as exc:
+        status = make_phase2_status("FAIL", errors=[str(exc).split(":", 1)[0]])
+        _write_status(args.output, status)
+        return 1
+
+    _write_status(args.output, status)
+    return 0 if status["status"] == "PASS" else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_bridge_v2_governed_http_bootstrap.py

@@ -102,6 +102,9 @@ def test_http_route_contract_is_complete():
         "/v2/release/execute",
         "/v2/release/consistency",
         "/v2/release/acceptance",
+        "/v2/task-admission/envelope",
+        "/v2/task-admission/authorize",
+        "/v2/task-admission/status",
     }
     from aios_chatgpt_bridge_v2_server import ROUTES
 

tests/test_phase2_introspect_and_freeze.py

@@ -0,0 +1,131 @@
+"""Local, non-SSH tests for the bounded Phase2 transport and status contract."""
+
+from __future__ import annotations
+
+import importlib.util
+import subprocess
+from pathlib import Path
+from types import SimpleNamespace
+
+import pytest
+
+
+SCRIPT = Path(__file__).parents[1] / "scripts" / "phase2_introspect_and_freeze.py"
+SPEC = importlib.util.spec_from_file_location("phase2_introspect_and_freeze", SCRIPT)
+assert SPEC and SPEC.loader
+phase2 = importlib.util.module_from_spec(SPEC)
+SPEC.loader.exec_module(phase2)
+
+
+def test_transport_sends_large_payload_without_argv_growth():
+    payload = b"x" * (256 * 1024)
+    observed = {}
+
+    def runner(argv, **kwargs):
+        observed["argv"] = argv
+        observed["kwargs"] = kwargs
+        return SimpleNamespace(returncode=0, stdout=b'{"ok":true}')
+
+    assert phase2.run_remote_python(payload, runner=runner) == b'{"ok":true}'
+    assert observed["kwargs"]["input"] == payload
+    argv = observed["argv"]
+    assert argv == [
+        "ssh.exe",
+        "-T",
+        "-o",
+        "BatchMode=yes",
+        "-o",
+        "ConnectTimeout=12",
+        "hw",
+        'test "$(hostname)" = cn-network-mkt && sudo -n python3 -',
+    ]
+    assert sum(len(part.encode("utf-8")) + 1 for part in argv) <= 512
+    assert all(payload not in part.encode("utf-8") for part in argv)
+    assert all("base64" not in part.lower() for part in argv)
+
+
+def test_transport_reports_timeout_without_stderr():
+    def runner(argv, **kwargs):
+        raise subprocess.TimeoutExpired(argv, kwargs["timeout"], stderr=b"secret=do-not-leak")
+
+    with pytest.raises(phase2.RemoteTransportError, match="timed out") as error:
+        phase2.run_remote_python(b"payload", runner=runner, timeout=3)
+    assert "secret" not in str(error.value)
+
+
+def test_transport_reports_nonzero_without_stderr():
+    def runner(argv, **kwargs):
+        return SimpleNamespace(returncode=23, stdout=b"", stderr=b"token=do-not-leak")
+
+    with pytest.raises(phase2.RemoteTransportError, match="status 23") as error:
+        phase2.run_remote_python(b"payload", runner=runner)
+    assert "token" not in str(error.value)
+
+
+def test_status_flags_are_explicit_and_missing_public_symbol_fails_closed():
+    valid = phase2.validate_public_function_ast(
+        "def offline_migrate_task_principal_schema():\n    pass\n"
+        "def supersede_p0_scope():\n    pass\n"
+    )
+    assert valid["ok"] is True
+
+    missing = phase2.validate_public_function_ast(
+        "def offline_migrate_task_principal_schema():\n    pass\n"
+    )
+    assert missing["ok"] is False
+    assert missing["missing"] == ["supersede_p0_scope"]
+
+    status = phase2.make_phase2_status(
+        "FAIL", errors=["PUBLIC_SYMBOL_MISSING:supersede_p0_scope"]
+    )
+    assert status["status"] == "FAIL"
+    assert status["fail_closed"] is True
+    assert all(status[flag] is False for flag in phase2.MUTATION_FLAGS)
+
+
+def test_dynamic_reexport_is_accepted_only_with_explicit_globals_export_mechanism():
+    source = (
+        "for _name, _value in _legacy.__dict__.items():\n"
+        "    globals()[_name] = _value\n"
+    )
+    result = phase2.validate_public_function_ast(source)
+    assert result["ok"] is True
+    assert result["dynamic_reexport"] is True
+
+    no_export = "x = 1\n"
+    result = phase2.validate_public_function_ast(no_export)
+    assert result["ok"] is False
+    assert result["missing"] == list(phase2.DEFAULT_REQUIRED_ADMISSION_FUNCTIONS)
+
+
+def test_status_validator_rejects_mutation_claims():
+    status = phase2.make_phase2_status("PASS")
+    assert phase2.validate_phase2_status(status)["read_only"] is True
+    status["git_mutation"] = True
+    with pytest.raises(phase2.Phase2StatusError, match="git_mutation"):
+        phase2.validate_phase2_status(status)
+
+
+def test_request_payload_contains_request_but_argv_does_not():
+    payload = phase2.build_remote_payload(
+        {
+            "issuer_pid": 123,
+            "expected_admission_sha256": "a" * 64,
+        }
+    )
+    assert b'"issuer_pid":123' in payload
+    assert b'"expected_admission_sha256":"' in payload
+    argv = phase2.build_ssh_argv()
+    assert all("123" not in part and "aaaa" not in part for part in argv)
+
+
+def test_live_defaults_use_current_issuer_and_storage_migration_module():
+    assert phase2.DEFAULT_ISSUER_UNIT == "aios-task-principal-issuer.service"
+    assert phase2.DEFAULT_MIGRATION_MODULE == "aios_task_principal_storage_migration"
+    payload = phase2.build_remote_payload()
+    assert b"aios-task-principal-issuer.service" in payload
+    assert b"aios_task_principal_storage_migration" in payload
+    stale_issuer = "aios-task-principal-" + "admission.service"
+    stale_migration = "aios_task_principal_" + "migration"
+    assert stale_issuer.encode() not in payload
+    assert stale_migration.encode() not in payload

tests/test_task_admission_bootstrap.py

@@ -0,0 +1,640 @@
+from __future__ import annotations
+
+import base64
+from concurrent.futures import ThreadPoolExecutor
+import hashlib
+from pathlib import Path
+import sys
+import threading
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime" / "app"
+if str(APP) not in sys.path:
+    sys.path.insert(0, str(APP))
+
+from aios_task_admission_bootstrap import (  # noqa: E402
+    OwnerDelegationEpoch,
+    TaskAdmissionBootstrapService,
+    TaskAdmissionError,
+    TaskAdmissionGrant,
+    TaskAdmissionPolicy,
+    TaskAdmissionStore,
+    canonical_json_bytes,
+)
+from aios_task_admission_live_adapters import (  # noqa: E402
+    AgentTaskAdmissionPort,
+    BoundOwnerAuthorizationBackend,
+)
+
+
+BASE_OID = "9" * 40
+EXECUTION_DIGEST = "a" * 64
+TASK_ID = "repository-discovery-p0-20260903"
+BRANCH = "agent/repository-discovery-p0-20260902"
+IDEMPOTENCY = "repository-discovery-p0-admission-20260903"
+FILES = [
+    "active-runtime/app/aios_repository_discovery.py",
+    "active-runtime/app/aios_chatgpt_bridge_v2.py",
+    "docs/gitops/openapi-chatgpt-actions.yaml",
+    "tests/test_repository_discovery.py",
+]
+CAPS = [
+    "governed.task.create",
+    "agent-task:control",
+    "task.principal.issue",
+    "writer.lease",
+    "governed.write.files",
+    "git.read",
+    "testing",
+    "verification",
+]
+
+
+class FakeAuthorization:
+    def __init__(self, *, policy_digest: str, now: int = 1_800_000_000):
+        self.now = now
+        self.signature_calls = 0
+        self.epoch_calls = 0
+        self.signature_ok = True
+        self.epoch = OwnerDelegationEpoch(
+            epoch_id="epoch-repository-discovery-20260903",
+            owner_identity="owner",
+            trust_domain="aios.owner",
+            repository="aios/aios",
+            allowed_branch_prefixes=("agent/repository-discovery-",),
+            allowed_capabilities=frozenset(CAPS),
+            allowed_file_prefixes=("active-runtime", "docs", "tests"),
+            risk_ceiling="medium",
+            issued_at=now - 60,
+            not_before=now - 60,
+            expires_at=now + 3600,
+            revocation_epoch=0,
+            policy_digest=policy_digest,
+        )
+
+    def verify_delegation_epoch(self, epoch_id: str) -> OwnerDelegationEpoch:
+        self.epoch_calls += 1
+        if epoch_id != self.epoch.epoch_id:
+            raise TaskAdmissionError("UNKNOWN_EPOCH", 403)
+        return self.epoch
+
+    def verify_owner_signature(
+        self,
+        *,
+        key_id: str,
+        canonical_payload: bytes,
+        signature_b64: str,
+    ) -> str:
+        self.signature_calls += 1
+        if key_id != "owner-key-1" or not self.signature_ok:
+            raise TaskAdmissionError("TASK_ADMISSION_OWNER_SIGNATURE_INVALID", 403)
+        expected = base64.b64encode(
+            hashlib.sha256(canonical_payload + b"\0owner-key-1").digest()
+        ).decode("ascii")
+        if signature_b64 != expected:
+            raise TaskAdmissionError("TASK_ADMISSION_OWNER_SIGNATURE_INVALID", 403)
+        return "owner"
+
+
+class FakeTaskPort:
+    def __init__(self):
+        self.calls = 0
+        self.grants = []
+        self._lock = threading.Lock()
+
+    def admit(self, grant):
+        with self._lock:
+            self.calls += 1
+            self.grants.append(grant)
+        return {"task_id": grant.task_id, "state": "QUEUED"}
+
+
+@pytest.fixture
+def policy():
+    return TaskAdmissionPolicy(
+        allowed_capabilities=frozenset(CAPS),
+        max_file_count=16,
+        max_task_ttl_seconds=3600,
+        max_envelope_ttl_seconds=900,
+        max_risk="medium",
+    )
+
+
+@pytest.fixture
+def service(tmp_path: Path, policy: TaskAdmissionPolicy):
+    now = 1_800_000_000
+    auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now)
+    port = FakeTaskPort()
+    svc = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
+        authorization=auth,
+        task_port=port,
+        policy=policy,
+        clock=lambda: now,
+    )
+    return svc, auth, port
+
+
+def request(**overrides):
+    value = {
+        "idempotency_key": IDEMPOTENCY,
+        "task_id": TASK_ID,
+        "repository": "aios/aios",
+        "authority_base_oid": BASE_OID,
+        "target_branch": BRANCH,
+        "generation": 1,
+        "execution_digest": EXECUTION_DIGEST,
+        "requested_capabilities": list(CAPS),
+        "file_allowlist": list(FILES),
+        "risk": "medium",
+        "expires_in_seconds": 600,
+    }
+    value.update(overrides)
+    return value
+
+
+def test_envelope_exact_replay_returns_same_nonce_and_payload(service):
+    svc, _, _ = service
+    first = svc.create_envelope(request())
+    second = svc.create_envelope(request())
+    assert first["idempotent"] is False
+    assert second["idempotent"] is True
+    assert first["envelope"] == second["envelope"]
+    assert first["canonical_payload_base64"] == second["canonical_payload_base64"]
+
+
+def test_envelope_idempotency_conflict_fails_closed(service):
+    svc, _, _ = service
+    svc.create_envelope(request())
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_IDEMPOTENCY_CONFLICT"):
+        svc.create_envelope(request(risk="low"))
+
+
+@pytest.mark.parametrize(
+    "field,value,code",
+    [
+        ("repository", "other/repo", "TASK_ADMISSION_REPOSITORY_REJECTED"),
+        ("target_branch", "main", "TASK_ADMISSION_BRANCH_REJECTED"),
+        ("authority_base_oid", "HEAD", "TASK_ADMISSION_BASE_OID_INVALID"),
+        ("requested_capabilities", ["production.write"], "TASK_ADMISSION_PROHIBITED_CAPABILITY"),
+        ("requested_capabilities", ["git:*"], "TASK_ADMISSION_CAPABILITY_INVALID"),
+        ("file_allowlist", ["../secrets"], "TASK_ADMISSION_FILE_PATH_INVALID"),
+        ("file_allowlist", ["/etc/passwd"], "TASK_ADMISSION_FILE_PATH_INVALID"),
+        ("risk", "high", "TASK_ADMISSION_RISK_OUT_OF_POLICY"),
+    ],
+)
+def test_envelope_policy_rejects_scope_widening(service, field, value, code):
+    svc, _, _ = service
+    with pytest.raises(TaskAdmissionError, match=code):
+        svc.create_envelope(request(**{field: value}))
+
+
+def test_valid_delegation_epoch_admits_exact_task_once(service):
+    svc, auth, port = service
+    envelope = svc.create_envelope(request())["envelope"]
+    result = svc.authorize_and_admit(
+        {
+            "envelope_id": envelope["envelope_id"],
+            "authorization_kind": "delegation_epoch",
+            "delegation_epoch_id": auth.epoch.epoch_id,
+        }
+    )
+    assert result["schema"] == "aios.task-admission-receipt.v1"
+    assert result["task_id"] == TASK_ID
+    assert result["task_state"] == "QUEUED"
+    assert result["parent_epoch_id"] == auth.epoch.epoch_id
+    assert port.calls == 1
+
+    replay = svc.authorize_and_admit(
+        {
+            "envelope_id": envelope["envelope_id"],
+            "authorization_kind": "delegation_epoch",
+            "delegation_epoch_id": auth.epoch.epoch_id,
+        }
+    )
+    assert replay["task_id"] == TASK_ID
+    assert replay["idempotent"] is True
+    assert port.calls == 1
+
+
+def test_post_admission_projection_runs_after_exact_task_readback_once(
+    tmp_path: Path, policy: TaskAdmissionPolicy
+):
+    auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest())
+    port = FakeTaskPort()
+    observations = []
+
+    def project(grant, task):
+        observations.append((grant.task_id, task["task_id"], task["state"]))
+        return {"job_id": "projected-job"}
+
+    svc = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
+        authorization=auth,
+        task_port=port,
+        policy=policy,
+        clock=lambda: 1_800_000_000,
+        post_admission=project,
+    )
+    envelope_id = svc.create_envelope(request())["envelope"]["envelope_id"]
+    proof = {
+        "envelope_id": envelope_id,
+        "authorization_kind": "delegation_epoch",
+        "delegation_epoch_id": auth.epoch.epoch_id,
+    }
+
+    first = svc.authorize_and_admit(proof)
+    replay = svc.authorize_and_admit(proof)
+
+    assert first["task_id"] == TASK_ID
+    assert replay["idempotent"] is True
+    assert observations == [(TASK_ID, TASK_ID, "QUEUED")]
+
+
+def test_epoch_capability_widening_rejected_before_task_port(service):
+    svc, auth, port = service
+    # Envelope policy allows the capability, parent epoch deliberately does not.
+    auth.epoch = OwnerDelegationEpoch(
+        **{
+            **auth.epoch.as_public_dict(),
+            "allowed_capabilities": frozenset(
+                item for item in CAPS if item != "writer.lease"
+            ),
+            "allowed_branch_prefixes": auth.epoch.allowed_branch_prefixes,
+            "allowed_file_prefixes": auth.epoch.allowed_file_prefixes,
+        }
+    )
+    envelope = svc.create_envelope(request())["envelope"]
+    with pytest.raises(
+        TaskAdmissionError, match="TASK_ADMISSION_EPOCH_CAPABILITY_WIDENING"
+    ):
+        svc.authorize_and_admit(
+            {
+                "envelope_id": envelope["envelope_id"],
+                "authorization_kind": "delegation_epoch",
+                "delegation_epoch_id": auth.epoch.epoch_id,
+            }
+        )
+    assert port.calls == 0
+
+
+def test_epoch_file_scope_widening_rejected(service):
+    svc, auth, port = service
+    auth.epoch = OwnerDelegationEpoch(
+        **{
+            **auth.epoch.as_public_dict(),
+            "allowed_capabilities": auth.epoch.allowed_capabilities,
+            "allowed_branch_prefixes": auth.epoch.allowed_branch_prefixes,
+            "allowed_file_prefixes": ("docs",),
+        }
+    )
+    envelope = svc.create_envelope(request())["envelope"]
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_EPOCH_FILE_WIDENING"):
+        svc.authorize_and_admit(
+            {
+                "envelope_id": envelope["envelope_id"],
+                "authorization_kind": "delegation_epoch",
+                "delegation_epoch_id": auth.epoch.epoch_id,
+            }
+        )
+    assert port.calls == 0
+
+
+def test_revoked_or_expired_epoch_rejected(service):
+    svc, auth, port = service
+    auth.epoch = OwnerDelegationEpoch(
+        **{
+            **auth.epoch.as_public_dict(),
+            "allowed_capabilities": auth.epoch.allowed_capabilities,
+            "allowed_branch_prefixes": auth.epoch.allowed_branch_prefixes,
+            "allowed_file_prefixes": auth.epoch.allowed_file_prefixes,
+            "revocation_epoch": 1,
+        }
+    )
+    envelope = svc.create_envelope(request())["envelope"]
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_EPOCH_REVOKED"):
+        svc.authorize_and_admit(
+            {
+                "envelope_id": envelope["envelope_id"],
+                "authorization_kind": "delegation_epoch",
+                "delegation_epoch_id": auth.epoch.epoch_id,
+            }
+        )
+    assert port.calls == 0
+
+
+def test_owner_signature_path_verifies_canonical_payload(service):
+    svc, auth, port = service
+    created = svc.create_envelope(request())
+    payload = canonical_json_bytes(created["envelope"])
+    signature = base64.b64encode(
+        hashlib.sha256(payload + b"\0owner-key-1").digest()
+    ).decode("ascii")
+    result = svc.authorize_and_admit(
+        {
+            "envelope_id": created["envelope"]["envelope_id"],
+            "authorization_kind": "owner_signature",
+            "owner_key_id": "owner-key-1",
+            "signature_base64": signature,
+        }
+    )
+    assert result["authorization_kind"] == "owner_signature"
+    assert result["owner_identity"] == "owner"
+    assert auth.signature_calls == 1
+    assert port.calls == 1
+
+
+def test_bad_owner_signature_is_rejected(service):
+    svc, _, port = service
+    envelope = svc.create_envelope(request())["envelope"]
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_OWNER_SIGNATURE_INVALID"):
+        svc.authorize_and_admit(
+            {
+                "envelope_id": envelope["envelope_id"],
+                "authorization_kind": "owner_signature",
+                "owner_key_id": "owner-key-1",
+                "signature_base64": base64.b64encode(b"bad").decode("ascii"),
+            }
+        )
+    assert port.calls == 0
+
+
+def test_expired_envelope_cannot_admit(tmp_path: Path, policy: TaskAdmissionPolicy):
+    now = [1_800_000_000]
+    auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now[0])
+    port = FakeTaskPort()
+    svc = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
+        authorization=auth,
+        task_port=port,
+        policy=policy,
+        clock=lambda: now[0],
+    )
+    envelope = svc.create_envelope(request(expires_in_seconds=60))["envelope"]
+    now[0] += 61
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_ENVELOPE_EXPIRED"):
+        svc.authorize_and_admit(
+            {
+                "envelope_id": envelope["envelope_id"],
+                "authorization_kind": "delegation_epoch",
+                "delegation_epoch_id": auth.epoch.epoch_id,
+            }
+        )
+    assert port.calls == 0
+
+
+def test_concurrent_exact_authorization_calls_admit_once(service):
+    svc, auth, port = service
+    envelope_id = svc.create_envelope(request())["envelope"]["envelope_id"]
+    body = {
+        "envelope_id": envelope_id,
+        "authorization_kind": "delegation_epoch",
+        "delegation_epoch_id": auth.epoch.epoch_id,
+    }
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        results = list(pool.map(lambda _: svc.authorize_and_admit(body), range(16)))
+    assert {item["task_id"] for item in results} == {TASK_ID}
+    assert port.calls == 1
+
+
+def test_cross_instance_envelope_winner_is_the_durable_value(tmp_path: Path, policy: TaskAdmissionPolicy):
+    auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest())
+    port = FakeTaskPort()
+    stores = [TaskAdmissionStore(tmp_path / "shared.sqlite3") for _ in range(16)]
+    services = [
+        TaskAdmissionBootstrapService(
+            store=store, authorization=auth, task_port=port, policy=policy,
+            clock=lambda: 1_800_000_000,
+        )
+        for store in stores
+    ]
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        values = list(pool.map(lambda svc: svc.create_envelope(request()), services))
+    assert len({item["envelope"]["envelope_id"] for item in values}) == 1
+    assert sum(not item["idempotent"] for item in values) == 1
+
+
+def test_cross_instance_authorization_calls_create_one_task(tmp_path: Path, policy: TaskAdmissionPolicy):
+    auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest())
+    port = FakeTaskPort()
+    database = tmp_path / "shared.sqlite3"
+    first = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(database), authorization=auth, task_port=port,
+        policy=policy, clock=lambda: 1_800_000_000,
+    )
+    envelope_id = first.create_envelope(request())["envelope"]["envelope_id"]
+    services = [
+        TaskAdmissionBootstrapService(
+            store=TaskAdmissionStore(database), authorization=auth, task_port=port,
+            policy=policy, clock=lambda: 1_800_000_000,
+        )
+        for _ in range(16)
+    ]
+    body = {
+        "envelope_id": envelope_id,
+        "authorization_kind": "delegation_epoch",
+        "delegation_epoch_id": auth.epoch.epoch_id,
+    }
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        values = list(pool.map(lambda svc: svc.authorize_and_admit(body), services))
+    assert {item["task_id"] for item in values} == {TASK_ID}
+    assert port.calls == 1
+
+
+@pytest.mark.parametrize("file_name", ["docs//gitops/a.yaml", "docs/./gitops/a.yaml", "docs/a\x00.yaml"])
+def test_file_paths_must_already_be_canonical(service, file_name):
+    svc, _, _ = service
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_FILE_PATH_INVALID"):
+        svc.create_envelope(request(file_allowlist=[file_name]))
+
+
+@pytest.mark.parametrize("generation", [True, "1", 1.0, 0])
+def test_generation_is_a_strict_positive_integer(service, generation):
+    svc, _, _ = service
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_GENERATION_INVALID"):
+        svc.create_envelope(request(generation=generation))
+
+
+def test_policy_digest_drift_and_changed_proof_cannot_replay(service):
+    svc, auth, _ = service
+    envelope_id = svc.create_envelope(request())["envelope"]["envelope_id"]
+    body = {
+        "envelope_id": envelope_id,
+        "authorization_kind": "delegation_epoch",
+        "delegation_epoch_id": auth.epoch.epoch_id,
+    }
+    svc.authorize_and_admit(body)
+    svc.policy = TaskAdmissionPolicy(
+        allowed_capabilities=frozenset(CAPS), policy_digest="policy-has-changed"
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_POLICY_DIGEST_DRIFT"):
+        svc.authorize_and_admit(body)
+
+
+def test_grant_expiry_is_clipped_to_parent_epoch(tmp_path: Path, policy: TaskAdmissionPolicy):
+    now = 1_800_000_000
+    auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest(), now=now)
+    auth.epoch = OwnerDelegationEpoch(
+        **{**auth.epoch.as_public_dict(), "allowed_capabilities": auth.epoch.allowed_capabilities,
+           "allowed_branch_prefixes": auth.epoch.allowed_branch_prefixes,
+           "allowed_file_prefixes": auth.epoch.allowed_file_prefixes, "expires_at": now + 120}
+    )
+    port = FakeTaskPort()
+    svc = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(tmp_path / "admission.sqlite3"), authorization=auth,
+        task_port=port, policy=policy, clock=lambda: now,
+    )
+    result = svc.authorize_and_admit({
+        "envelope_id": svc.create_envelope(request())["envelope"]["envelope_id"],
+        "authorization_kind": "delegation_epoch", "delegation_epoch_id": auth.epoch.epoch_id,
+    })
+    assert port.grants[0].expires_at == now + 120
+    assert result["task_id"] == TASK_ID
+
+
+def test_owner_backend_uses_bound_caller_and_decoded_armored_signature(policy):
+    captured = {}
+
+    class Verifier:
+        def verify(self, **kwargs):
+            captured.update(kwargs)
+            return True
+
+    epoch = FakeAuthorization(policy_digest=policy.normalized_policy_digest()).epoch
+    backend = BoundOwnerAuthorizationBackend(
+        authenticated_caller={"uid": "bridge-service"},
+        verified_epoch_provider=lambda **kwargs: captured.update(epoch_lookup=kwargs) or epoch,
+        signature_verifier=Verifier(),
+    )
+    assert backend.verify_delegation_epoch(epoch.epoch_id) == epoch
+    assert captured["epoch_lookup"]["caller"] == {"uid": "bridge-service"}
+    assert backend.verify_owner_signature(
+        key_id="owner-key-1", canonical_payload=b"exact", signature_b64=base64.b64encode(b"-----BEGIN SSH SIGNATURE-----").decode(),
+    ) == "aios-task-grant-signer"
+    assert captured["namespace"] == "aios-owner-approval-v1"
+    assert captured["payload"] == b"exact"
+
+
+def test_official_task_port_reconciles_ambiguous_create(tmp_path: Path, policy):
+    grant = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
+        authorization=FakeAuthorization(policy_digest=policy.normalized_policy_digest()),
+        task_port=FakeTaskPort(), policy=policy, clock=lambda: 1_800_000_000,
+    )
+    # Construct via the public flow so the port sees a production-shaped grant.
+    created = grant.create_envelope(request())["envelope"]
+    envelope = grant._load_envelope(created["envelope_id"])
+    task_grant = TaskAdmissionGrant(
+        grant_id="task-grant-adapter-test", envelope_id=envelope.envelope_id,
+        task_id=envelope.task_id, repository=envelope.repository,
+        authority_base_oid=envelope.authority_base_oid, target_branch=envelope.target_branch,
+        generation=envelope.generation, execution_digest=envelope.execution_digest,
+        capabilities=envelope.requested_capabilities, file_allowlist=envelope.file_allowlist,
+        risk=envelope.risk, trust_domain=envelope.trust_domain,
+        policy_digest=envelope.policy_digest, authorization_kind="owner_signature",
+        owner_identity="aios-task-grant-signer", parent_epoch_id=None,
+        issued_at=1_800_000_000, expires_at=1_800_000_600,
+    )
+
+    class NotFound(Exception):
+        status = 404
+
+    class Transport:
+        def __init__(self): self.calls = []; self.task = None
+        def request(self, method, path, body=None):
+            self.calls.append((method, path, body))
+            if method == "GET" and self.task is None: raise NotFound()
+            if method == "POST":
+                self.task = {key: value for key, value in body.items() if key != "idempotency_key"}
+                self.task["status"] = "QUEUED"
+                raise TimeoutError("response lost after side effect")
+            return self.task
+
+    transport = Transport()
+    plan = {
+        "external_job_id": "fixture-job-adapter-01", "repository": task_grant.repository,
+        "target_branch": task_grant.target_branch, "generation": task_grant.generation,
+        "execution_digest": task_grant.execution_digest,
+    }
+    assert AgentTaskAdmissionPort(transport, execution_plan_provider=lambda _: plan).admit(task_grant) == {"task_id": TASK_ID, "state": "QUEUED"}
+    assert [call[:2] for call in transport.calls] == [
+        ("GET", "/v1/tasks/" + TASK_ID), ("POST", "/v1/tasks"), ("GET", "/v1/tasks/" + TASK_ID),
+    ]
+
+
+def test_execution_identity_is_available_only_to_trusted_post_admission(
+    tmp_path: Path,
+    policy: TaskAdmissionPolicy,
+):
+    class NotFound(Exception):
+        status = 404
+
+    class Transport:
+        def __init__(self):
+            self.task = None
+
+        def request(self, method, path, body=None):
+            if method == "GET" and self.task is None:
+                raise NotFound()
+            if method == "POST":
+                self.task = {
+                    key: value for key, value in body.items()
+                    if key != "idempotency_key"
+                }
+                self.task["status"] = "QUEUED"
+            return self.task
+
+    plan = {
+        "external_job_id": "job-0123456789abcdef0123456789abcdef",
+        "repository": "aios/aios",
+        "target_branch": BRANCH,
+        "generation": 1,
+        "execution_digest": EXECUTION_DIGEST,
+    }
+    projected: dict[str, object] = {}
+    auth = FakeAuthorization(policy_digest=policy.normalized_policy_digest())
+    service = TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(tmp_path / "admission.sqlite3"),
+        authorization=auth,
+        task_port=AgentTaskAdmissionPort(
+            Transport(), execution_plan_provider=lambda _: plan
+        ),
+        policy=policy,
+        clock=lambda: 1_800_000_000,
+        post_admission=lambda _grant, result: projected.update(result),
+    )
+    envelope_id = service.create_envelope(request())["envelope"]["envelope_id"]
+    receipt = service.authorize_and_admit(
+        {
+            "envelope_id": envelope_id,
+            "authorization_kind": "delegation_epoch",
+            "delegation_epoch_id": auth.epoch.epoch_id,
+        }
+    )
+    assert receipt["task_id"] == TASK_ID
+    assert "external_job_id" not in receipt
+    assert projected == {
+        "task_id": TASK_ID,
+        "state": "QUEUED",
+        "external_job_id": plan["external_job_id"],
+    }
+
+
+def test_status_is_read_only_and_reports_admitted_task(service):
+    svc, auth, _ = service
+    envelope_id = svc.create_envelope(request())["envelope"]["envelope_id"]
+    before = svc.status(envelope_id)
+    assert before["status"] == "PENDING_AUTHORIZATION"
+    assert before["task"] is None
+    svc.authorize_and_admit(
+        {
+            "envelope_id": envelope_id,
+            "authorization_kind": "delegation_epoch",
+            "delegation_epoch_id": auth.epoch.epoch_id,
+        }
+    )
+    after = svc.status(envelope_id)
+    assert after["status"] == "ADMITTED"
+    assert after["task"]["task_id"] == TASK_ID

tests/test_task_admission_bootstrap_contract.py

@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime" / "app"
+if str(APP) not in sys.path:
+    sys.path.insert(0, str(APP))
+
+from aios_task_admission_bridge_contract import (  # noqa: E402
+    TASK_ADMISSION_OPERATION_REQUIRED_FIELDS,
+    TASK_ADMISSION_ROUTES,
+)
+
+
+def test_bootstrap_route_operation_mapping_is_exact():
+    assert TASK_ADMISSION_ROUTES == {
+        "/v2/task-admission/envelope": "getTaskAdmissionEnvelope",
+        "/v2/task-admission/authorize": "submitTaskAdmissionAuthorization",
+        "/v2/task-admission/status": "getTaskAdmissionStatus",
+    }
+
+
+def test_bootstrap_operations_do_not_replace_writer_or_principal_gates():
+    operations = set(TASK_ADMISSION_OPERATION_REQUIRED_FIELDS)
+    assert "acquireWriterLease" not in operations
+    assert "commitFilesGoverned" not in operations
+    assert "issueTaskPrincipal" not in operations
+    assert operations == {
+        "getTaskAdmissionEnvelope",
+        "submitTaskAdmissionAuthorization",
+        "getTaskAdmissionStatus",
+    }
+
+
+def test_openapi_fragment_matches_runtime_contract():
+    text = (
+        ROOT / "docs" / "gitops" / "openapi-task-admission-bootstrap-fragment.yaml"
+    ).read_text(encoding="utf-8")
+    for route, operation in TASK_ADMISSION_ROUTES.items():
+        assert f"  {route}:" in text
+        assert f"operationId: {operation}" in text
+    assert "global write" not in text.lower()
+    assert "runShell" not in text
+    assert "force push" not in text.lower()

tests/test_task_admission_bootstrap_http.py

@@ -0,0 +1,406 @@
+from __future__ import annotations
+
+import ast
+from contextlib import contextmanager
+import json
+import os
+from pathlib import Path
+import sys
+import threading
+from types import SimpleNamespace
+from http.server import ThreadingHTTPServer
+from urllib.error import HTTPError
+from urllib.parse import quote
+from urllib.request import Request, urlopen
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime" / "app"
+if str(APP) not in sys.path:
+    sys.path.insert(0, str(APP))
+
+from aios_chatgpt_bridge_v2 import (  # noqa: E402
+    BridgeV2Adapter,
+    TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS,
+    TASK_ADMISSION_OPERATION_REQUIRED_FIELDS,
+    TASK_ADMISSION_OPERATIONS,
+)
+from aios_chatgpt_bridge_v2_server import (  # noqa: E402
+    BridgeHandler,
+    GovernedRuntime,
+    build_bridge_adapter,
+)
+from aios_governed_task_bootstrap import BootstrapError  # noqa: E402
+from aios_task_admission_bootstrap import TaskAdmissionError  # noqa: E402
+
+
+class StubRuntime:
+    def __init__(self, status: dict[str, object], handlers=None) -> None:
+        self.status = status
+        self._handlers = dict(handlers or {})
+
+    def runtime_status(self) -> dict[str, object]:
+        return dict(self.status)
+
+    def handlers(self) -> dict[str, object]:
+        return dict(self._handlers)
+
+
+class FakeOwnerPort:
+    def __init__(self) -> None:
+        self.envelopes: list[dict[str, object]] = []
+        self.authorizations: list[dict[str, object]] = []
+
+    def create_envelope(self, request):
+        self.envelopes.append(dict(request))
+        return {"envelope_id": "env-1", "status": "PENDING", "http_status": 201}
+
+    def authorize(self, request, task_port):
+        self.authorizations.append(dict(request))
+        kind = request.get("authorization_kind")
+        if kind == "unsigned":
+            raise TaskAdmissionError("TASK_ADMISSION_AUTHORIZATION_REQUIRED", 400)
+        if kind == "bad_proof":
+            raise TaskAdmissionError("TASK_ADMISSION_OWNER_SIGNATURE_INVALID", 403)
+        return task_port.admit(request)
+
+
+class FakeTaskPort:
+    def __init__(self) -> None:
+        self.calls: list[dict[str, object]] = []
+
+    def admit(self, request):
+        self.calls.append(dict(request))
+        return {"task_id": request.get("task_id", "task-1"), "state": "QUEUED"}
+
+
+def _status(**overrides: object) -> dict[str, object]:
+    value: dict[str, object] = {
+        "status": "active",
+        "health": 200,
+        "governed_writes_enabled": False,
+        "legacy_global_writes_enabled": False,
+        "task_principal_writes_enabled": False,
+    }
+    value.update(overrides)
+    return value
+
+
+def _admission_handlers(owner: FakeOwnerPort, task: FakeTaskPort):
+    return {
+        "getTaskAdmissionEnvelope": owner.create_envelope,
+        "submitTaskAdmissionAuthorization": lambda request: owner.authorize(
+            request, task
+        ),
+        "getTaskAdmissionStatus": lambda request: {
+            "envelope_id": request["envelope_id"],
+            "status": "PENDING",
+        },
+    }
+
+
+def _post(base_url: str, path: str, body: dict[str, object]):
+    request = Request(
+        base_url + path,
+        data=json.dumps(body).encode("utf-8"),
+        headers={"Content-Type": "application/json"},
+        method="POST",
+    )
+    try:
+        with urlopen(request, timeout=3) as response:
+            return response.status, json.loads(response.read())
+    except HTTPError as exc:
+        return exc.code, json.loads(exc.read())
+
+
+def _get(base_url: str, path: str):
+    try:
+        with urlopen(base_url + path, timeout=3) as response:
+            return response.status, json.loads(response.read())
+    except HTTPError as exc:
+        return exc.code, json.loads(exc.read())
+
+
+@contextmanager
+def _serve(adapter: BridgeV2Adapter):
+    class TestBridgeHandler(BridgeHandler):
+        pass
+
+    TestBridgeHandler.adapter = adapter
+    TestBridgeHandler.principal_verifier = None
+    server = ThreadingHTTPServer(("127.0.0.1", 0), TestBridgeHandler)
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    try:
+        yield f"http://127.0.0.1:{server.server_port}"
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=3)
+
+
+def test_capabilities_expose_admission_only_for_complete_callable_handlers():
+    runtime = StubRuntime(_status(), {"getRuntimeStatus": lambda request: request})
+    without_backend = build_bridge_adapter(runtime)
+    assert not set(TASK_ADMISSION_OPERATIONS) & set(
+        without_backend.getGovernedWriteCapabilities()["operations"]
+    )
+    assert not set(TASK_ADMISSION_OPERATIONS) & set(
+        without_backend.getGovernedWriteCapabilities()["operation_contracts"]
+    )
+
+    handlers = _admission_handlers(FakeOwnerPort(), FakeTaskPort())
+    complete = build_bridge_adapter(runtime, task_admission_handlers=handlers)
+    capabilities = complete.getGovernedWriteCapabilities()
+    assert set(TASK_ADMISSION_OPERATIONS) <= set(capabilities["operations"])
+    expected_routes = {
+        "getTaskAdmissionEnvelope": ("POST", "/v2/task-admission/envelope"),
+        "submitTaskAdmissionAuthorization": ("POST", "/v2/task-admission/authorize"),
+        "getTaskAdmissionStatus": ("GET", "/v2/task-admission/status"),
+    }
+    for operation in TASK_ADMISSION_OPERATIONS:
+        contract = capabilities["operation_contracts"][operation]
+        assert (contract["method"], contract["path"]) == expected_routes[operation]
+        assert contract["required_fields"] == list(
+            TASK_ADMISSION_OPERATION_REQUIRED_FIELDS[operation]
+        )
+        assert contract["optional_fields"] == list(
+            TASK_ADMISSION_OPERATION_OPTIONAL_FIELDS[operation]
+        )
+
+
+def test_partial_task_admission_handlers_are_rejected():
+    runtime = StubRuntime(_status())
+    try:
+        build_bridge_adapter(
+            runtime,
+            task_admission_handlers={
+                "getTaskAdmissionEnvelope": lambda request: request,
+            },
+        )
+    except ValueError as exc:
+        assert str(exc) == "TASK_ADMISSION_HANDLERS_INCOMPLETE"
+    else:
+        raise AssertionError("partial Task Admission handlers must be rejected")
+
+
+def test_http_admission_backend_missing_returns_404():
+    adapter = build_bridge_adapter(StubRuntime(_status()))
+    with _serve(adapter) as base_url:
+        status, body = _post(
+            base_url,
+            "/v2/task-admission/envelope",
+            {"unexpected": True},
+        )
+    assert status == 404
+    assert body == {"error": "NOT_FOUND"}
+
+
+def test_http_admission_envelope_and_status_work_without_task_principal():
+    owner = FakeOwnerPort()
+    task = FakeTaskPort()
+    adapter = build_bridge_adapter(
+        StubRuntime(_status()),
+        task_admission_handlers=_admission_handlers(owner, task),
+    )
+    envelope = {
+        "idempotency_key": "test",
+        "task_id": "task-1",
+        "repository": "aios/aios",
+        "authority_base_oid": "a" * 40,
+        "target_branch": "agent/task-1",
+        "generation": 1,
+        "execution_digest": "b" * 64,
+        "requested_capabilities": ["testing"],
+        "file_allowlist": ["tests/test_task.py"],
+        "risk": "low",
+        "expires_in_seconds": 60,
+    }
+    with _serve(adapter) as base_url:
+        status, body = _post(base_url, "/v2/task-admission/envelope", envelope)
+        assert status == 201
+        assert body["envelope_id"] == "env-1"
+        status, body = _get(
+            base_url, "/v2/task-admission/status?envelope_id=" + quote("env-1")
+        )
+    assert status == 200
+    assert body == {"envelope_id": "env-1", "status": "PENDING"}
+    assert len(owner.envelopes) == 1
+    assert task.calls == []
+
+
+def test_http_admission_rejects_bad_queries_and_caller_principal():
+    owner = FakeOwnerPort()
+    task = FakeTaskPort()
+    adapter = build_bridge_adapter(
+        StubRuntime(_status()),
+        task_admission_handlers=_admission_handlers(owner, task),
+    )
+    envelope = {
+        "idempotency_key": "test",
+        "task_id": "task-1",
+        "repository": "aios/aios",
+        "authority_base_oid": "a" * 40,
+        "target_branch": "agent/task-1",
+        "generation": 1,
+        "execution_digest": "b" * 64,
+        "requested_capabilities": ["testing"],
+        "file_allowlist": ["tests/test_task.py"],
+        "risk": "low",
+    }
+    with _serve(adapter) as base_url:
+        for query in (
+            "unknown=x",
+            "envelope_id=env-1&envelope_id=env-2",
+            "envelope_id=",
+        ):
+            status, body = _get(base_url, "/v2/task-admission/status?" + query)
+            assert status == 400
+            assert body["error"] == "TASK_ADMISSION_QUERY_INVALID"
+
+        status, body = _post(
+            base_url,
+            "/v2/task-admission/envelope",
+            {**envelope, "principal": {"task_id": "caller-controlled"}},
+        )
+    assert status == 400
+    assert body == {"error": "CALLER_PRINCIPAL_FORBIDDEN"}
+    assert owner.envelopes == []
+    assert task.calls == []
+
+
+def test_http_admission_proof_failures_use_task_admission_status_and_do_not_mutate():
+    owner = FakeOwnerPort()
+    task = FakeTaskPort()
+    adapter = build_bridge_adapter(
+        StubRuntime(_status()),
+        task_admission_handlers=_admission_handlers(owner, task),
+    )
+    with _serve(adapter) as base_url:
+        for kind, expected_status, expected_error in (
+            ("unsigned", 400, "TASK_ADMISSION_AUTHORIZATION_REQUIRED"),
+            ("bad_proof", 403, "TASK_ADMISSION_OWNER_SIGNATURE_INVALID"),
+        ):
+            status, body = _post(
+                base_url,
+                "/v2/task-admission/authorize",
+                {"envelope_id": "env-1", "authorization_kind": kind},
+            )
+            assert status == expected_status
+            assert body == {"error": expected_error}
+    assert task.calls == []
+
+
+def test_existing_governed_mutation_gates_are_unchanged():
+    request = {
+        "schema": "aios.writer-lease-request.v2",
+        "task_id": "task-1",
+        "job_id": "job-1",
+        "repository": "aios/aios",
+        "branch_mode": "CREATE_NEW",
+        "base_oid": "a" * 40,
+        "expected_branch_oid": "a" * 40,
+        "target_branch": "agent/task-1",
+        "generation": 1,
+        "idempotency_key": "test",
+        "lease_seconds": 60,
+    }
+    runtime = StubRuntime(_status(), {"acquireWriterLease": lambda body: body})
+    with _serve(build_bridge_adapter(runtime)) as base_url:
+        status, body = _post(base_url, "/v2/writer-lease", request)
+        assert status == 423
+        assert body == {"error": "TASK_PRINCIPAL_WRITES_DISABLED"}
+
+    enabled_runtime = StubRuntime(
+        _status(
+            governed_writes_enabled=True,
+            task_principal_writes_enabled=True,
+        ),
+        {"acquireWriterLease": lambda body: body},
+    )
+    with _serve(build_bridge_adapter(enabled_runtime)) as base_url:
+        status, body = _post(base_url, "/v2/writer-lease", request)
+    assert status == 401
+    assert body == {"error": "TASK_PRINCIPAL_VERIFIER_UNAVAILABLE"}
+
+
+def test_post_admission_projection_is_exact_and_idempotent():
+    class Authority:
+        def __init__(self):
+            self.branch = None
+
+        @staticmethod
+        def commit_exists(oid):
+            return oid == "a" * 40
+
+        def read_ref(self, ref):
+            assert ref == "refs/heads/agent/admitted-task"
+            return self.branch
+
+    class Bootstrap:
+        def __init__(self):
+            self.binding = None
+
+        def get_binding(self, task_id):
+            if task_id != "admitted-task" or self.binding is None:
+                raise BootstrapError("TASK_EXECUTION_PROJECTION_MISSING")
+            return self.binding
+
+        def _write(self, binding):
+            assert self.binding is None
+            self.binding = binding
+
+    runtime = object.__new__(GovernedRuntime)
+    runtime.config = SimpleNamespace(repository="aios/aios")
+    runtime.authority = Authority()
+    runtime.bootstrap = Bootstrap()
+    runtime.bindings = {}
+    grant = SimpleNamespace(
+        task_id="admitted-task",
+        repository="aios/aios",
+        authority_base_oid="a" * 40,
+        target_branch="agent/admitted-task",
+        risk="low",
+        generation=1,
+    )
+    task_result = {"external_job_id": "job-0123456789abcdef0123456789abcdef"}
+
+    first = runtime.project_admitted_task(grant, task_result)
+    assert first["branch_mode"] == "CREATE_NEW"
+    assert first["expected_branch_oid"] == "a" * 40
+    assert first["job_id"] == task_result["external_job_id"]
+    assert runtime.project_admitted_task(grant, task_result)["idempotent"] is True
+
+    runtime.authority.branch = "b" * 40
+    changed = SimpleNamespace(**{**vars(grant), "task_id": "other-task"})
+    try:
+        runtime.project_admitted_task(changed, task_result)
+    except RuntimeError as exc:
+        assert str(exc) == "TASK_ADMISSION_PROJECTION_BRANCH_CONFLICT"
+    else:
+        raise AssertionError("drifted branch must not create a projection")
+
+
+def test_live_bridge_preservation_when_source_root_is_supplied():
+    configured_source = os.environ.get("AIOS_LIVE_BRIDGE_SOURCE")
+    if not configured_source:
+        return
+    source = Path(configured_source)
+    live_server = source / "aios_chatgpt_bridge_v2_server.py" if source.is_dir() else source
+    candidate_server = APP / "aios_chatgpt_bridge_v2_server.py"
+    live_tree = ast.parse(live_server.read_text(encoding="utf-8"))
+    candidate_tree = ast.parse(candidate_server.read_text(encoding="utf-8"))
+
+    def selected(tree):
+        return [
+            node
+            for node in ast.walk(tree)
+            if (isinstance(node, ast.ClassDef) and node.name == "GovernedRuntime")
+            or (isinstance(node, ast.FunctionDef) and node.name == "_verified_request")
+        ]
+
+    def without_attributes(nodes):
+        return [ast.dump(node, include_attributes=False) for node in nodes]
+
+    assert without_attributes(selected(candidate_tree)) == without_attributes(
+        selected(live_tree)
+    )

tests/test_task_admission_bootstrap_recovery.py

@@ -0,0 +1,474 @@
+from __future__ import annotations
+
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+import sys
+import threading
+import subprocess
+from dataclasses import replace
+
+import pytest
+
+ROOT = Path(__file__).resolve().parents[1]
+APP = ROOT / "active-runtime" / "app"
+if str(APP) not in sys.path:
+    sys.path.insert(0, str(APP))
+
+from aios_task_admission_bootstrap import (  # noqa: E402
+    OwnerDelegationEpoch,
+    TaskAdmissionBootstrapService,
+    TaskAdmissionError,
+    TaskAdmissionPolicy,
+    TaskAdmissionStore,
+)
+
+
+NOW = 1_800_000_000
+CAPS = ("governed.task.create", "agent-task:control")
+
+
+class Authorization:
+    def __init__(self, digest: str):
+        self.epoch = OwnerDelegationEpoch(
+            epoch_id="epoch-recovery", owner_identity="owner", trust_domain="aios.owner",
+            repository="aios/aios", allowed_branch_prefixes=("agent/recovery-",),
+            allowed_capabilities=frozenset(CAPS), allowed_file_prefixes=("docs",),
+            risk_ceiling="medium", issued_at=NOW - 1, not_before=NOW - 1,
+            expires_at=NOW + 600, revocation_epoch=0, policy_digest=digest,
+        )
+
+    def verify_delegation_epoch(self, epoch_id):
+        if epoch_id != self.epoch.epoch_id:
+            raise TaskAdmissionError("UNKNOWN_EPOCH", 403)
+        return self.epoch
+
+    def verify_owner_signature(self, **_kwargs):
+        return "owner"
+
+
+class RecoveringPort:
+    def __init__(self, *, fail_after_effect=False, fail_before_effect=False):
+        self.fail_after_effect = fail_after_effect
+        self.fail_before_effect = fail_before_effect
+        self.posts = 0
+        self.task_ids = set()
+        self._lock = threading.Lock()
+
+    def admit(self, grant):
+        with self._lock:
+            if grant.task_id in self.task_ids:
+                return {"task_id": grant.task_id, "state": "QUEUED"}
+            if self.fail_before_effect:
+                self.fail_before_effect = False
+                raise RuntimeError("before effect")
+            self.posts += 1
+            self.task_ids.add(grant.task_id)
+            if self.fail_after_effect:
+                self.fail_after_effect = False
+                raise RuntimeError("after effect")
+        return {"task_id": grant.task_id, "state": "QUEUED"}
+
+
+def request():
+    return {
+        "idempotency_key": "recovery-admission-key", "task_id": "recovery-task-20260903",
+        "repository": "aios/aios", "authority_base_oid": "a" * 40,
+        "target_branch": "agent/recovery-20260903", "generation": 1,
+        "execution_digest": "b" * 64, "requested_capabilities": list(CAPS),
+        "file_allowlist": ["docs/recovery.md"], "risk": "medium", "expires_in_seconds": 300,
+    }
+
+
+def make_service(path, port):
+    policy = TaskAdmissionPolicy(allowed_capabilities=frozenset(CAPS))
+    auth = Authorization(policy.normalized_policy_digest())
+    return TaskAdmissionBootstrapService(
+        store=TaskAdmissionStore(path), authorization=auth, task_port=port,
+        policy=policy, clock=lambda: NOW,
+    ), auth
+
+
+@pytest.mark.parametrize("after_effect", [True, False])
+def test_replay_recovers_crash_window_with_one_effect(tmp_path, after_effect):
+    database = tmp_path / "admission.sqlite3"
+    port = RecoveringPort(
+        fail_after_effect=after_effect, fail_before_effect=not after_effect,
+    )
+    first, auth = make_service(database, port)
+    envelope_id = first.create_envelope(request())["envelope"]["envelope_id"]
+    proof = {"envelope_id": envelope_id, "authorization_kind": "delegation_epoch", "delegation_epoch_id": auth.epoch.epoch_id}
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_TASK_STATE_UNKNOWN"):
+        first.authorize_and_admit(proof)
+
+    second, _ = make_service(database, port)
+    replay = second.authorize_and_admit(proof)
+    assert replay["task_id"] == "recovery-task-20260903"
+    assert port.posts == 1
+
+
+def test_cross_instance_recovery_lock_allows_one_effect(tmp_path):
+    database = tmp_path / "admission.sqlite3"
+    port = RecoveringPort()
+    first, auth = make_service(database, port)
+    envelope_id = first.create_envelope(request())["envelope"]["envelope_id"]
+    proof = {"envelope_id": envelope_id, "authorization_kind": "delegation_epoch", "delegation_epoch_id": auth.epoch.epoch_id}
+    services = [make_service(database, port)[0] for _ in range(16)]
+    with ThreadPoolExecutor(max_workers=16) as pool:
+        receipts = list(pool.map(lambda svc: svc.authorize_and_admit(proof), services))
+    assert {value["task_id"] for value in receipts} == {"recovery-task-20260903"}
+    assert port.posts == 1
+
+
+def test_changed_valid_epoch_proof_conflicts_after_receipt(tmp_path):
+    service, auth = make_service(tmp_path / "admission.sqlite3", RecoveringPort())
+    envelope_id = service.create_envelope(request())["envelope"]["envelope_id"]
+    first = {"envelope_id": envelope_id, "authorization_kind": "delegation_epoch", "delegation_epoch_id": auth.epoch.epoch_id}
+    service.authorize_and_admit(first)
+    auth.epoch = OwnerDelegationEpoch(
+        **{**auth.epoch.as_public_dict(), "epoch_id": "epoch-recovery-reissued",
+           "allowed_capabilities": auth.epoch.allowed_capabilities,
+           "allowed_branch_prefixes": auth.epoch.allowed_branch_prefixes,
+           "allowed_file_prefixes": auth.epoch.allowed_file_prefixes}
+    )
+    changed = {**first, "delegation_epoch_id": auth.epoch.epoch_id}
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_AUTHORIZATION_CONFLICT"):
+        service.authorize_and_admit(changed)
+
+
+@pytest.mark.parametrize("value", ["docs/a.md", b"docs/a.md", {"docs/a.md": True}, [], ["docs/a\x01.md"], ["docs\\a.md"]])
+def test_file_allowlist_requires_canonical_sequence(tmp_path, value):
+    service, _ = make_service(tmp_path / "admission.sqlite3", RecoveringPort())
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_(FILE_ALLOWLIST_REJECTED|FILE_PATH_INVALID)"):
+        service.create_envelope({**request(), "file_allowlist": value})
+
+
+@pytest.mark.parametrize("branch", ["agent/.hidden-task", "agent/recovery.lock", "agent/recovery@{x", "agent/recovery..task"])
+def test_branch_components_reject_git_ambiguous_spelling(tmp_path, branch):
+    service, _ = make_service(tmp_path / "admission.sqlite3", RecoveringPort())
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_BRANCH_REJECTED"):
+        service.create_envelope({**request(), "target_branch": branch})
+
+
+def test_process_death_releases_admission_lock(tmp_path):
+    database = tmp_path / "death.sqlite3"
+    script = (
+        "import sys,os;from pathlib import Path;"
+        "from aios_task_admission_bootstrap import TaskAdmissionStore;"
+        "s=TaskAdmissionStore(Path(sys.argv[1]));"
+        "lock=s.admission_lock('death-envelope');lock.__enter__();os._exit(17)"
+    )
+    result = subprocess.run([sys.executable, "-c", script, str(database)], check=False)
+    assert result.returncode == 17
+    with TaskAdmissionStore(database).admission_lock("death-envelope", timeout_seconds=1):
+        pass
+
+
+def test_persisted_grant_cannot_outlive_its_own_shorter_ttl(tmp_path):
+    port = RecoveringPort(fail_before_effect=True)
+    service, auth = make_service(tmp_path / "expiry.sqlite3", port)
+    service.policy = replace(service.policy, max_task_ttl_seconds=60)
+    auth.epoch = replace(auth.epoch, policy_digest=service.policy.normalized_policy_digest())
+    envelope = service.create_envelope(request())["envelope"]
+    proof = {"envelope_id": envelope["envelope_id"], "authorization_kind": "delegation_epoch", "delegation_epoch_id": auth.epoch.epoch_id}
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_TASK_STATE_UNKNOWN"):
+        service.authorize_and_admit(proof)
+    service.clock = lambda: NOW + 61
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_GRANT_EXPIRED"):
+        service.authorize_and_admit(proof)
+    assert port.posts == 0
+
+
+def _adapter_fixture(tmp_path):
+    from aios_task_admission_live_adapters import AgentTaskAdmissionPort
+    service, auth = make_service(tmp_path / "adapter.sqlite3", RecoveringPort())
+    envelope = service.create_envelope(request())["envelope"]
+    service.authorize_and_admit({"envelope_id": envelope["envelope_id"], "authorization_kind": "delegation_epoch", "delegation_epoch_id": auth.epoch.epoch_id})
+    with service.store._connection() as con:
+        import json
+        from aios_task_admission_bootstrap import TaskAdmissionGrant
+        raw = json.loads(con.execute("SELECT grant_json FROM task_admission_grants").fetchone()[0])
+    grant = TaskAdmissionGrant(**{**raw, "capabilities": tuple(raw["capabilities"]), "file_allowlist": tuple(raw["file_allowlist"])})
+    plan = {"external_job_id": "fixture-exact-job", "repository": grant.repository,
+            "target_branch": grant.target_branch, "generation": grant.generation,
+            "execution_digest": grant.execution_digest, "steps": [{"operation": "fixture-only"}]}
+    class Transport:
+        posts = 0
+        task = None
+        error = None
+        def request(self, method, path, body=None):
+            if method == "POST": self.posts += 1; raise AssertionError("unexpected create")
+            if self.error: raise self.error
+            return self.task
+    transport = Transport()
+    port = AgentTaskAdmissionPort(transport, execution_plan_provider=lambda _: plan)
+    transport.task = {**port._body(grant, plan), "status": "QUEUED"}
+    return port, transport, grant, plan
+
+
+@pytest.mark.parametrize("field", ["metadata", "instructions", "required_capabilities", "constraints", "deadline", "priority"])
+def test_existing_task_conflict_never_becomes_absence(tmp_path, field):
+    port, transport, grant, _ = _adapter_fixture(tmp_path)
+    transport.task[field] = None
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_TASK_IDENTITY_CONFLICT"):
+        port.admit(grant)
+    assert transport.posts == 0
+
+
+def test_unknown_get_does_not_create_task(tmp_path):
+    port, transport, grant, _ = _adapter_fixture(tmp_path)
+    transport.error = TimeoutError("fixture readback unavailable")
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_TASK_STATE_UNKNOWN"):
+        port.admit(grant)
+    assert transport.posts == 0
+
+
+def test_execution_provider_keeps_full_payload_and_rejects_type_drift(tmp_path):
+    port, transport, grant, plan = _adapter_fixture(tmp_path)
+    assert port._execution_plan(grant) == plan
+    plan["generation"] = True
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_EXECUTION_PLAN_CONFLICT"):
+        port.admit(grant)
+    assert transport.posts == 0
+
+
+def test_signed_epoch_host_provider_reverifies_signature_and_policy(tmp_path):
+    import hashlib
+    import json
+    import time
+    from aios_task_admission_bootstrap import canonical_json_bytes
+    from aios_task_admission_live_adapters import (
+        SignedEpochDirectoryProvider,
+        task_admission_policy_from_epoch,
+    )
+
+    class Verifier:
+        def __init__(self):
+            self.calls = []
+
+        def verify(self, **kwargs):
+            self.calls.append(kwargs)
+            return True
+
+    now = int(time.time())
+    epoch_value = {
+        "schema": "aios.owner-delegation-epoch.v1",
+        "epoch_id": "owner-epoch-host-test-20260903",
+        "owner_identity": "aios-task-grant-signer",
+        "trust_domain": "aios.owner",
+        "repository": "aios/aios",
+        "allowed_branch_prefixes": ["agent/p0-test"],
+        "allowed_capabilities": ["git.read", "governed.write.file"],
+        "allowed_file_prefixes": ["active-runtime/app/example.py"],
+        "risk_ceiling": "medium",
+        "issued_at": now - 1,
+        "not_before": now - 1,
+        "expires_at": now + 600,
+        "revocation_epoch": 0,
+        "policy_digest": "a" * 64,
+    }
+    root = tmp_path / "epoch"
+    root.mkdir(mode=0o700)
+    (root / "owner-delegation-epoch.canonical.json").write_bytes(
+        canonical_json_bytes(epoch_value)
+    )
+    (root / "owner-delegation-epoch.sig").write_text(
+        "-----BEGIN SSH SIGNATURE-----\nabc\n-----END SSH SIGNATURE-----\n",
+        encoding="utf-8",
+    )
+    (root / "owner-delegation-epoch.key-id").write_text(
+        "SHA256:test-key\n",
+        encoding="utf-8",
+    )
+    revocations = tmp_path / "revocations.json"
+    revocations.write_bytes(
+        canonical_json_bytes({
+            "schema": "aios.owner-delegation-revocations.v1",
+            "revoked_epoch_ids": [],
+        })
+    )
+    for path in (
+        root / "owner-delegation-epoch.canonical.json",
+        root / "owner-delegation-epoch.sig",
+        root / "owner-delegation-epoch.key-id",
+        revocations,
+    ):
+        path.chmod(0o600)
+
+    verifier = Verifier()
+    caller = {"service": "test", "trust_domain": "aios.owner"}
+    provider = SignedEpochDirectoryProvider(
+        root=root,
+        verifier=verifier,
+        revocation_file=revocations,
+        expected_caller=caller,
+        require_root_owner=False,
+        clock=lambda: now,
+    )
+    epoch = provider(epoch_id=epoch_value["epoch_id"], caller=caller)
+    policy = task_admission_policy_from_epoch(epoch)
+    assert verifier.calls[0]["namespace"] == "aios-owner-approval-v1"
+    assert verifier.calls[0]["payload"] == canonical_json_bytes(epoch_value)
+    assert policy.allowed_capabilities == frozenset(
+        epoch_value["allowed_capabilities"]
+    )
+    assert policy.policy_digest == epoch_value["policy_digest"]
+
+
+def test_signed_epoch_host_provider_rejects_revocation_and_wrong_caller(tmp_path):
+    import time
+    from aios_task_admission_bootstrap import canonical_json_bytes
+    from aios_task_admission_live_adapters import SignedEpochDirectoryProvider
+
+    class Verifier:
+        def verify(self, **_kwargs):
+            return True
+
+    now = int(time.time())
+    epoch_id = "owner-epoch-host-revoked-20260903"
+    root = tmp_path / "epoch"
+    root.mkdir(mode=0o700)
+    value = {
+        "schema": "aios.owner-delegation-epoch.v1",
+        "epoch_id": epoch_id,
+        "owner_identity": "aios-task-grant-signer",
+        "trust_domain": "aios.owner",
+        "repository": "aios/aios",
+        "allowed_branch_prefixes": ["agent/p0-test"],
+        "allowed_capabilities": ["git.read"],
+        "allowed_file_prefixes": ["docs/test.md"],
+        "risk_ceiling": "medium",
+        "issued_at": now - 1,
+        "not_before": now - 1,
+        "expires_at": now + 600,
+        "revocation_epoch": 0,
+        "policy_digest": "b" * 64,
+    }
+    (root / "owner-delegation-epoch.canonical.json").write_bytes(
+        canonical_json_bytes(value)
+    )
+    (root / "owner-delegation-epoch.sig").write_text(
+        "sig", encoding="utf-8"
+    )
+    (root / "owner-delegation-epoch.key-id").write_text(
+        "SHA256:test-key", encoding="utf-8"
+    )
+    revocations = tmp_path / "revocations.json"
+    revocations.write_bytes(
+        canonical_json_bytes({
+            "schema": "aios.owner-delegation-revocations.v1",
+            "revoked_epoch_ids": [epoch_id],
+        })
+    )
+    for path in list(root.iterdir()) + [revocations]:
+        path.chmod(0o600)
+
+    caller = {"service": "test", "trust_domain": "aios.owner"}
+    provider = SignedEpochDirectoryProvider(
+        root=root,
+        verifier=Verifier(),
+        revocation_file=revocations,
+        expected_caller=caller,
+        require_root_owner=False,
+        clock=lambda: now,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_CALLER_BINDING_MISMATCH"):
+        provider(epoch_id=epoch_id, caller={"service": "wrong"})
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_EPOCH_REVOKED"):
+        provider(epoch_id=epoch_id, caller=caller)
+
+
+def test_execution_plan_host_provider_is_content_addressed(tmp_path):
+    import hashlib
+    from types import SimpleNamespace
+    from aios_task_admission_bootstrap import canonical_json_bytes
+    from aios_task_admission_live_adapters import ExecutionPlanDirectoryProvider
+
+    plan = {
+        "task_id": "p0-task-admission-bootstrap-20260903",
+        "repository": "aios/aios",
+        "target_branch": "agent/p0-task-admission-bootstrap-repair-20260903",
+        "generation": 1,
+        "external_job_id": "job-host-test",
+        "files": [],
+    }
+    digest = hashlib.sha256(canonical_json_bytes(plan)).hexdigest()
+    root = tmp_path / "plans"
+    root.mkdir(mode=0o700)
+    path = root / (digest + ".json")
+    path.write_bytes(canonical_json_bytes(plan))
+    path.chmod(0o600)
+    grant = SimpleNamespace(
+        execution_digest=digest,
+        task_id=plan["task_id"],
+        repository=plan["repository"],
+        target_branch=plan["target_branch"],
+        generation=1,
+    )
+    provider = ExecutionPlanDirectoryProvider(
+        root=root,
+        require_root_owner=False,
+    )
+    value = provider(grant)
+    assert value["execution_digest"] == digest
+    assert value["external_job_id"] == "job-host-test"
+
+
+def test_bridge_persistent_consumer_is_opt_in_and_default_fail_closed():
+    source = (
+        ROOT / "active-runtime" / "app" / "aios_chatgpt_bridge_v2_server.py"
+    ).read_text(encoding="utf-8")
+    assert 'os.environ.get("AIOS_TASK_ADMISSION_EPOCH_ROOT", "").strip()' in source
+    assert "build_task_admission_handlers_from_environment" in source
+    assert "task_admission_handlers is None" in source
+
+
+def test_bridge_persistent_host_is_opt_in_and_activation_file_is_fail_closed():
+    text = (APP / "aios_chatgpt_bridge_v2_server.py").read_text(encoding="utf-8")
+    assert 'AIOS_TASK_ADMISSION_EPOCH_ROOT' in text
+    assert 'AIOS_TASK_ADMISSION_ACTIVATION_FILE' in text
+    assert '/var/lib/aios-gitops/bridge-v2-staging/task-admission-activation.json' in text
+    assert 'Path(activation_raw).is_file()' in text
+    assert 'build_task_admission_handlers_from_activation_file' in text
+
+
+def test_host_default_state_database_stays_inside_existing_bridge_state_root():
+    text = (APP / "aios_task_admission_live_adapters.py").read_text(encoding="utf-8")
+    assert '/var/lib/aios-gitops/bridge-v2-staging/task-admission.sqlite3' in text
+    assert '/var/lib/aios-task-admission/state.sqlite3' not in text
+
+def test_host_posix_permission_gate_is_skipped_when_posix_metadata_is_not_authoritative(
+    tmp_path,
+    monkeypatch,
+):
+    import aios_task_admission_live_adapters as live
+
+    root = tmp_path / "non-posix-mode"
+    root.mkdir()
+    target = root / "value.json"
+    target.write_text("{}", encoding="utf-8")
+
+    # Deliberately make the synthesized mode permissive. On Windows this is
+    # normal and must not be interpreted as a POSIX trust failure.
+    root.chmod(0o777)
+    target.chmod(0o666)
+
+    monkeypatch.setattr(live, "_host_posix_metadata_enforced", lambda: False)
+    live._host_require_directory(root, root_owned=True)
+    live._host_require_regular(target, root_owned=True)
+
+
+@pytest.mark.skipif(
+    __import__("os").name != "posix",
+    reason="POSIX mode-bit enforcement is a Linux/Unix trust boundary",
+)
+def test_host_posix_permission_gate_rejects_group_or_other_writable(tmp_path):
+    import aios_task_admission_live_adapters as live
+
+    root = tmp_path / "posix-mode"
+    root.mkdir(mode=0o700)
+    root.chmod(0o777)
+    with pytest.raises(
+        live.TaskAdmissionHostError,
+        match="SECURE_DIRECTORY_WRITABLE_BY_GROUP_OR_OTHER",
+    ):
+        live._host_require_directory(root, root_owned=False)

tests/test_trusted_builder_signer_provisioning.py

@@ -112,3 +112,180 @@ def test_package_contract_declares_socket_activation_no_credentials_and_0755_lau
 def test_provisioner_self_test_has_no_mutation(capsys):
     assert provisioning.main(["--self-test"]) == 0
     assert capsys.readouterr().out.strip() == "PASS"
+
+
+def _replay_fixture(tmp_path: Path):
+    _source_root, payload_path, payload = _payload(tmp_path)
+    raw = payload_path.read_bytes()
+    signature = tmp_path / "owner.sig"
+    signature.write_bytes(b"owner-signature-test")
+    os.chmod(signature, 0o600)
+
+    config = provisioning.ProvisionerConfig()
+    config.credential_store = tmp_path / "credstore"
+    config.public_keyring = tmp_path / "state" / "public-keyring.json"
+    config.binding_file = tmp_path / "state" / "service-principal-binding.json"
+    config.receipt_log = tmp_path / "state" / "provisioning-receipts.jsonl"
+    config.credential_store.mkdir(mode=0o700)
+    os.chmod(config.credential_store, 0o700)
+    config.public_keyring.parent.mkdir(mode=0o700)
+
+    public = b"p" * 32
+    now = provisioning.utc_now()
+    keyring = provisioning._keyring(payload["signing_identity"], public, now)
+    config.public_keyring.write_bytes(provisioning.pretty_json(keyring))
+    os.chmod(config.public_keyring, 0o600)
+    binding = {
+        "schema": "aios.trusted-builder.service-principal-binding.v1",
+        "purpose": provisioning.PURPOSE,
+        "service_principal": payload["signing_identity"]["service_principal"],
+        "key_id": payload["signing_identity"]["key_id"],
+        "generation": payload["signing_identity"]["key_generation"],
+        "algorithm": "Ed25519",
+        "private_key_export": False,
+        "public_key_sha256": provisioning.sha256_bytes(public),
+        "source_commit": payload["source"]["gitea_main_oid"],
+        "source_tree": payload["source"]["gitea_main_tree_oid"],
+        "payload_sha256": provisioning.sha256_bytes(raw),
+    }
+    config.binding_file.write_bytes(provisioning.pretty_json(binding))
+    os.chmod(config.binding_file, 0o600)
+    for name in ("signing_private_key", "builder_hmac"):
+        target = config.credential_store / name
+        target.write_bytes(("encrypted-" + name).encode())
+        os.chmod(target, 0o600)
+    signature_bytes = signature.read_bytes()
+    receipt = {
+        "schema": "aios.trusted-builder-signer-provisioning-receipt.v1",
+        "status": "COMPLETED",
+        "created_at": provisioning.iso(now),
+        "operation_key": "payload:" + provisioning.sha256_bytes(raw) + ":signature:" + provisioning.sha256_bytes(signature_bytes),
+        "payload_sha256": provisioning.sha256_bytes(raw),
+        "signature_sha256": provisioning.sha256_bytes(signature_bytes),
+        "source_commit": payload["source"]["gitea_main_oid"],
+        "source_tree": payload["source"]["gitea_main_tree_oid"],
+        "key_id": payload["signing_identity"]["key_id"],
+        "generation": payload["signing_identity"]["key_generation"],
+        "algorithm": "Ed25519",
+        "purpose": provisioning.PURPOSE,
+        "service_principal": payload["signing_identity"]["service_principal"],
+        "service_principal_bound": True,
+        "public_key_sha256": provisioning.sha256_bytes(public),
+        "private_key_exportable": False,
+        "credential_projection": "systemd-encrypted-host-credential",
+        "negative_capability_tests": {name: True for name in sorted(provisioning.DENIED_CAPABILITIES)},
+        "production_mutation": False,
+        "protected_ref_mutation": False,
+        "tag_mutation": False,
+        "baseline_mutation": False,
+    }
+    return config, payload_path, payload, signature, receipt
+
+
+def test_legacy_multiline_receipt_parse(tmp_path: Path):
+    path = tmp_path / "receipts.jsonl"
+    receipt = {"schema": "legacy", "status": "COMPLETED", "nested": {"value": True}}
+    path.write_bytes(provisioning.pretty_json(receipt))
+    assert provisioning._receipts(path) == [receipt]
+
+
+def test_new_canonical_jsonl_receipt_parse(tmp_path: Path):
+    path = tmp_path / "receipts.jsonl"
+    rows = [{"id": 1}, {"id": 2}]
+    for row in rows:
+        provisioning._append(path, row)
+    assert path.read_bytes() == b"".join(provisioning.canonical_json(row) + b"\n" for row in rows)
+    assert provisioning._receipts(path) == rows
+
+
+@pytest.mark.parametrize(
+    "content",
+    [b'{"status":"COMPLETED"}\nnot-json\n', b'{"status":"COMPLETED"'],
+    ids=["malformed", "truncated"],
+)
+def test_malformed_or_truncated_receipt_fails_closed(tmp_path: Path, content: bytes):
+    path = tmp_path / "receipts.jsonl"
+    path.write_bytes(content)
+    with pytest.raises(provisioning.ProvisioningError, match="RECEIPT_LOG_INVALID"):
+        provisioning._receipts(path)
+
+
+def test_preexisting_valid_state_is_adopted_without_mutation(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    os.chmod(config.receipt_log, 0o600)
+    before = {
+        path: path.read_bytes()
+        for path in (
+            config.receipt_log, config.public_keyring, config.binding_file,
+            config.credential_store / "signing_private_key",
+            config.credential_store / "builder_hmac",
+        )
+    }
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    result = provisioning.provision(config, payload_path, signature)
+    assert result["status"] == "IDEMPOTENT_REPLAY_PASS"
+    assert result["mutation_executed"] is False
+    assert result["existing_signer_state_adopted"] is True
+    assert result["private_key_exportable"] is False
+    assert before == {path: path.read_bytes() for path in before}
+
+
+def test_duplicate_replay_is_idempotent(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    os.chmod(config.receipt_log, 0o600)
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    first = provisioning.provision(config, payload_path, signature)
+    receipt_bytes = config.receipt_log.read_bytes()
+    second = provisioning.provision(config, payload_path, signature)
+    assert first == second
+    assert config.receipt_log.read_bytes() == receipt_bytes
+
+
+def test_wrong_payload_is_rejected_against_preexisting_state(tmp_path: Path, monkeypatch):
+    config, payload_path, payload, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    os.chmod(config.receipt_log, 0o600)
+    payload["nonce"] = "wrong-payload-nonce-0002"
+    payload_path.write_bytes(provisioning.pretty_json(payload))
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    with pytest.raises(provisioning.ProvisioningError, match="PREEXISTING_STATE_CONFLICT"):
+        provisioning.provision(config, payload_path, signature)
+
+
+def test_wrong_signature_is_rejected(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    monkeypatch.setattr(
+        provisioning,
+        "_run_signature_verify",
+        lambda *_args: (_ for _ in ()).throw(provisioning.ProvisioningError("OWNER_SIGNATURE_VERIFY_FAILED")),
+    )
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    with pytest.raises(provisioning.ProvisioningError, match="OWNER_SIGNATURE_VERIFY_FAILED"):
+        provisioning.provision(config, payload_path, signature)
+
+
+def test_wrong_purpose_is_rejected(tmp_path: Path):
+    _source_root, _payload_path, payload = _payload(tmp_path)
+    payload["signing_identity"]["purpose"] = "WRONG_PURPOSE"
+    with pytest.raises(provisioning.ProvisioningError, match="SIGNING_IDENTITY_INVALID"):
+        provisioning._validate_payload(payload)
+
+
+def test_replay_rejects_state_binding_tampering(tmp_path: Path, monkeypatch):
+    config, payload_path, _payload_value, signature, receipt = _replay_fixture(tmp_path)
+    receipt["service_principal"] = "wrong.service"
+    config.receipt_log.write_bytes(provisioning.pretty_json(receipt))
+    monkeypatch.setattr(provisioning, "_run_signature_verify", lambda *_args: None)
+    monkeypatch.setattr(provisioning, "_verify_replay_source", lambda *_args: None)
+    monkeypatch.setattr(provisioning.os, "geteuid", lambda: 0)
+    with pytest.raises(provisioning.ProvisioningError, match="RECEIPT_CONFLICT"):
+        provisioning.provision(config, payload_path, signature)