feat(governed): TaskPrincipal auto-recovery (AIOS-BASELINE-1.1-PRODUCTION-CLOSURE)

Integrate the TaskPrincipal auto-recovery patch from the Agent Execution
Handoff 2026-09-08 onto the Baseline 1.1 mainline candidate.

Server-side execution principal recovery:
- TaskAdmissionStore.get_grant_by_id() / get_grant_by_task_id(): read-only
  immutable grant lookup from durable admission grant/receipt state.
- TaskAdmissionPrincipalProvider: recover short-lived execution principal
  from durable admission grant + Owner epoch + authoritative official Task;
  only accepts CLAIMED/RUNNING tasks; re-verifies epoch/capability/task/job/
  repository/branch/generation before every mutation; caller JSON cannot
  supply principal; all miss-scopes fail closed.
- Bridge main() wires the admission runtime verifier back into
  BridgeHandler.principal_verifier; trusted hosts may still inject.
- staging unit: AIOS_P0_TASK_PRINCIPAL_WRITES_ENABLED=true while
  governed_writes_enabled still fail-closed on verifier readiness + legacy
  writes=false.
- OpenAPI contract updated to server-side principal recovery.
- New negative tests + adjusted recovery tests.

Verified before applying: all 8 modified files matched manifest before_sha256
(LF-normalized), so applied after-state bytes directly (no 3-way merge needed).
Index blobs match manifest after_sha256 for all 9 files.
Targeted suite: 81 passed / 1 skipped (POSIX mode-bit, Linux env) on CPython 3.13.

Base: 6ffd7fe (Baseline 1.1 port); authoritative main 0136a534.

active-runtime/app/aios_chatgpt_bridge_v2.py

@@ -157,7 +157,7 @@ def render_openapi_contract_fragment() -> str:
         "      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"
+        "      x-aios-task-principal: Bridge resolves a fresh server-side TaskPrincipal from the authoritative admitted+claimed Task and re-verifies its Owner epoch, then checks exact task/job/repository/branch/generation/capability binding; caller JSON cannot provide a principal and missing verification is denied. An explicitly injected legacy bearer verifier may still be used by a trusted host.\n"
         "      requestBody:\n"
         "        required: true\n"
         "        content:\n"

active-runtime/app/aios_chatgpt_bridge_v2_server.py

@@ -1279,6 +1279,7 @@ def serve(adapter: BridgeV2Adapter, host: str = "127.0.0.1", port: int = 18892)
 def main(
     *,
     task_admission_handlers: Mapping[str, Callable[[Mapping[str, Any]], Any]] | None = None,
+    task_principal_verifier: TaskPrincipalVerifier | None = None,
 ) -> None:
     activation_raw = os.environ.get(
         "AIOS_TASK_ADMISSION_ACTIVATION_FILE",
@@ -1333,27 +1334,33 @@ def main(
         repository_discovery_cursor_secret=discovery_cursor_secret,
     )
     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
+    # Prefer a server-side TaskPrincipal provider recovered from the same
+    # verified Task-admission runtime.  This removes the caller-side credential
+    # bootstrap loop while keeping every mutation dependent on a currently
+    # claimed Task and a freshly re-verified Owner epoch.  Hosts may still
+    # inject an independently authenticated verifier explicitly.
+    BridgeHandler.principal_verifier = task_principal_verifier
     if task_admission_handlers is None:
+        admission_runtime = None
         if os.environ.get("AIOS_TASK_ADMISSION_EPOCH_ROOT", "").strip():
             from aios_task_admission_live_adapters import (
-                build_task_admission_handlers_from_environment,
+                build_task_admission_runtime_from_environment,
             )
-            task_admission_handlers = build_task_admission_handlers_from_environment(
+            admission_runtime = build_task_admission_runtime_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,
+                build_task_admission_runtime_from_activation_file,
             )
-            task_admission_handlers = build_task_admission_handlers_from_activation_file(
+            admission_runtime = build_task_admission_runtime_from_activation_file(
                 Path(activation_raw),
                 post_admission=runtime.project_admitted_task,
             )
+        if admission_runtime is not None:
+            task_admission_handlers = admission_runtime.handlers
+            if BridgeHandler.principal_verifier is None:
+                BridgeHandler.principal_verifier = admission_runtime.principal_verifier
     serve(
         build_bridge_adapter(
             runtime,

active-runtime/app/aios_task_admission_bootstrap.py

@@ -438,6 +438,71 @@ class TaskAdmissionStore:
             ).fetchone()
         return dict(row) if row else None
 
+    def get_grant_by_id(self, grant_id: str) -> TaskAdmissionGrant | None:
+        """Return one immutable admitted grant by its stable grant identity.
+
+        This is a read-only lookup used by the server-side TaskPrincipal
+        recovery path.  It never trusts caller-supplied grant content.
+        """
+        if not isinstance(grant_id, str) or not grant_id:
+            raise TaskAdmissionError("TASK_ADMISSION_GRANT_ID_INVALID", 400)
+        with self._connection() as con:
+            row = con.execute(
+                "SELECT grant_json FROM task_admission_grants WHERE grant_id=?",
+                (grant_id,),
+            ).fetchone()
+        if row is None:
+            return None
+        try:
+            raw = json.loads(row["grant_json"])
+            return TaskAdmissionGrant(
+                **{
+                    **raw,
+                    "capabilities": tuple(raw["capabilities"]),
+                    "file_allowlist": tuple(raw["file_allowlist"]),
+                }
+            )
+        except (TypeError, ValueError, KeyError, json.JSONDecodeError) as exc:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_DURABLE_GRANT_INVALID", 503
+            ) from exc
+
+    def get_grant_by_task_id(self, task_id: str) -> TaskAdmissionGrant | None:
+        """Return the immutable admitted grant bound to an authoritative Task."""
+        if not isinstance(task_id, str) or not task_id:
+            raise TaskAdmissionError("TASK_ADMISSION_TASK_ID_INVALID", 400)
+        with self._connection() as con:
+            row = con.execute(
+                """
+                SELECT g.grant_json
+                FROM task_admission_grants AS g
+                JOIN task_admission_receipts AS r
+                  ON r.envelope_id = g.envelope_id
+                WHERE r.task_id = ?
+                """,
+                (task_id,),
+            ).fetchone()
+        if row is None:
+            return None
+        try:
+            raw = json.loads(row["grant_json"])
+            grant = TaskAdmissionGrant(
+                **{
+                    **raw,
+                    "capabilities": tuple(raw["capabilities"]),
+                    "file_allowlist": tuple(raw["file_allowlist"]),
+                }
+            )
+        except (TypeError, ValueError, KeyError, json.JSONDecodeError) as exc:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_DURABLE_GRANT_INVALID", 503
+            ) from exc
+        if grant.task_id != task_id:
+            raise TaskAdmissionError(
+                "TASK_ADMISSION_DURABLE_GRANT_TASK_MISMATCH", 503
+            )
+        return grant
+
     def save_envelope(
         self,
         *,

active-runtime/app/aios_task_admission_live_adapters.py

@@ -257,6 +257,28 @@ class AgentTaskAdmissionPort:
             raise TaskAdmissionError("TASK_ADMISSION_TASK_STATE_UNKNOWN", 503)
         return recovered
 
+    def verify_active_task(
+        self, grant: TaskAdmissionGrant
+    ) -> tuple[Mapping[str, Any], Mapping[str, Any]]:
+        """Re-read and validate the authoritative Task for execution use.
+
+        Admission may complete while the Task is still QUEUED.  A TaskPrincipal
+        may be recovered only after the Task authority reports CLAIMED/RUNNING.
+        The exact immutable metadata is revalidated on every call.
+        """
+        execution = self._execution_plan(grant)
+        task = self._read(grant, execution)
+        if task is None:
+            raise TaskAdmissionError("TASK_PRINCIPAL_TASK_NOT_FOUND", 403)
+        state = str(task.get("state") or task.get("status") or "")
+        if state not in {"CLAIMED", "RUNNING"}:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_TASK_NOT_CLAIMED",
+                409,
+                details={"state": state},
+            )
+        return task, execution
+
     def projection_result(
         self,
         grant: TaskAdmissionGrant,
@@ -286,6 +308,152 @@ class AgentTaskAdmissionPort:
             "external_job_id": external_job_id,
         }
 
+_BRIDGE_PRINCIPAL_CAPABILITIES = {
+    "acquireWriterLease": "writer.lease",
+    "commitFileGoverned": "governed.write.file",
+    "commitFilesGoverned": "governed.write.files",
+    "getGitOpsJob": "writer.lease",
+    "resumeGitOpsJob": "writer.lease",
+    "reconcileGitOpsJob": "writer.lease",
+}
+
+
+@dataclass(frozen=True, slots=True)
+class TaskAdmissionPrincipalProvider:
+    """Recover a short-lived execution principal from durable admission state.
+
+    This provider is server-side only.  It never accepts an Owner grant,
+    TaskPrincipal, or capability from caller JSON.  Every use re-reads the
+    admitted grant, re-verifies its parent Owner epoch (including revocation),
+    re-reads the authoritative Task, and requires CLAIMED/RUNNING before a
+    mutating Bridge operation can continue.
+    """
+
+    store: Any
+    authorization: Any
+    task_port: AgentTaskAdmissionPort
+    epoch_id: str
+    clock: Any = time.time
+
+    def readiness(self) -> Mapping[str, Any]:
+        try:
+            epoch = self.authorization.verify_delegation_epoch(self.epoch_id)
+        except Exception as exc:
+            return {
+                "ready": False,
+                "mode": "server-side-task-admission",
+                "error": str(exc) or type(exc).__name__,
+            }
+        return {
+            "ready": True,
+            "mode": "server-side-task-admission",
+            "epoch_id": epoch.epoch_id,
+            "repository": epoch.repository,
+        }
+
+    @staticmethod
+    def _request_value(
+        request: Mapping[str, Any], name: str, expected: object
+    ) -> None:
+        if name in request and request[name] != expected:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_REQUEST_BINDING_MISMATCH",
+                403,
+                details={"field": name},
+            )
+
+    def __call__(
+        self,
+        transport: Mapping[str, Any],
+        request: Mapping[str, Any],
+    ) -> Mapping[str, Any]:
+        operation = str(transport.get("operation") or "")
+        capability = _BRIDGE_PRINCIPAL_CAPABILITIES.get(operation)
+        if capability is None:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_BOOTSTRAP_ROUTE_REQUIRED", 403
+            )
+        task_id = request.get("task_id")
+        if not isinstance(task_id, str) or not task_id:
+            raise TaskAdmissionError("TASK_PRINCIPAL_TASK_ID_INVALID", 403)
+
+        grant = self.store.get_grant_by_task_id(task_id)
+        if grant is None:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_AUTHORIZATION_REQUIRED", 403
+            )
+        if grant.parent_epoch_id is None:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_PARENT_EPOCH_REQUIRED", 403
+            )
+
+        # Re-verify the exact parent on every use so revocation/expiry cannot be
+        # bypassed by a previously admitted Task.
+        epoch = self.authorization.verify_delegation_epoch(grant.parent_epoch_id)
+        if (
+            epoch.epoch_id != self.epoch_id
+            or epoch.repository != grant.repository
+            or epoch.trust_domain != grant.trust_domain
+            or not set(grant.capabilities).issubset(epoch.allowed_capabilities)
+            or not any(
+                grant.target_branch.startswith(prefix)
+                for prefix in epoch.allowed_branch_prefixes
+            )
+        ):
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_PARENT_EPOCH_SCOPE_MISMATCH", 403
+            )
+
+        now = int(self.clock())
+        if not grant.issued_at <= now < grant.expires_at:
+            raise TaskAdmissionError("TASK_PRINCIPAL_GRANT_EXPIRED", 403)
+        if (
+            "task.principal.issue" not in grant.capabilities
+            or capability not in grant.capabilities
+        ):
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_CAPABILITY_DENIED", 403
+            )
+
+        task, execution = self.task_port.verify_active_task(grant)
+        external_job_id = execution.get("external_job_id")
+        if not isinstance(external_job_id, str) or not external_job_id:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_EXECUTION_ID_INVALID", 503
+            )
+
+        self._request_value(request, "task_id", grant.task_id)
+        self._request_value(request, "job_id", external_job_id)
+        self._request_value(request, "repository", grant.repository)
+        self._request_value(request, "target_branch", grant.target_branch)
+        self._request_value(request, "generation", grant.generation)
+        if operation == "acquireWriterLease":
+            self._request_value(
+                request, "base_oid", grant.authority_base_oid
+            )
+
+        return {
+            "schema": "aios.task-principal-server-attestation.v1",
+            "task_id": grant.task_id,
+            "grant_id": grant.grant_id,
+            "parent_epoch_id": grant.parent_epoch_id,
+            "repository": grant.repository,
+            "target_branch": grant.target_branch,
+            "generation": grant.generation,
+            "job_id": external_job_id,
+            "capability": capability,
+            "task_state": str(task.get("state") or task.get("status") or ""),
+            "request_digest": sha256_json(dict(request)),
+            "verified_at": now,
+        }
+
+
+@dataclass(frozen=True, slots=True)
+class TaskAdmissionRuntimeBundle:
+    handlers: Mapping[str, Any]
+    principal_verifier: TaskAdmissionPrincipalProvider
+
+
 class TaskAdmissionHostError(RuntimeError):
     """Stable fail-closed host-composition error."""
 
@@ -670,7 +838,7 @@ def task_admission_policy_from_epoch(
     )
 
 
-def _build_task_admission_handlers(
+def _build_task_admission_runtime(
     *,
     caller_service: str,
     epoch_root: Path,
@@ -682,8 +850,8 @@ def _build_task_admission_handlers(
     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."""
+) -> TaskAdmissionRuntimeBundle:
+    """Compose admission handlers plus automatic execution-principal recovery."""
     from aios_task_admission_bootstrap import (
         TaskAdmissionBootstrapService,
         TaskAdmissionStore,
@@ -708,34 +876,49 @@ def _build_task_admission_handlers(
         verified_epoch_provider=provider,
         signature_verifier=verifier,
     )
+    transport = LocalAgentTaskHTTPTransport(
+        base_url=task_api_url,
+        token_file=task_control_token_file,
+        require_root_owner=True,
+    )
     task_port = AgentTaskAdmissionPort(
-        LocalAgentTaskHTTPTransport(
-            base_url=task_api_url,
-            token_file=task_control_token_file,
-            require_root_owner=True,
-        ),
+        transport,
         execution_plan_provider=ExecutionPlanDirectoryProvider(
             root=execution_plan_dir,
             require_root_owner=True,
         ),
     )
+    store = TaskAdmissionStore(state_db)
     service = TaskAdmissionBootstrapService(
-        store=TaskAdmissionStore(state_db),
+        store=store,
         authorization=authorization,
         task_port=task_port,
         policy=task_admission_policy_from_epoch(epoch),
         post_admission=post_admission,
     )
-    return TaskAdmissionBridgeContract(service).handlers()
+    return TaskAdmissionRuntimeBundle(
+        handlers=TaskAdmissionBridgeContract(service).handlers(),
+        principal_verifier=TaskAdmissionPrincipalProvider(
+            store=store,
+            authorization=authorization,
+            task_port=task_port,
+            epoch_id=epoch.epoch_id,
+        ),
+    )
 
 
-def build_task_admission_handlers_from_activation_file(
+def _build_task_admission_handlers(**kwargs: Any) -> Mapping[str, Any]:
+    """Compatibility wrapper for callers that only need admission routes."""
+    return _build_task_admission_runtime(**kwargs).handlers
+
+
+def build_task_admission_runtime_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."""
+) -> TaskAdmissionRuntimeBundle:
+    """Load admission handlers plus server-side principal recovery."""
     _host_require_regular(activation_file, root_owned=True)
     value = _host_read_json_object(activation_file)
     required = {
@@ -777,7 +960,7 @@ def build_task_admission_handlers_from_activation_file(
     if task_api_url != "http://127.0.0.1:18940":
         raise TaskAdmissionHostError("TASK_ADMISSION_ACTIVATION_TASK_API_INVALID")
 
-    return _build_task_admission_handlers(
+    return _build_task_admission_runtime(
         caller_service=caller_service,
         epoch_root=paths["epoch_root"],
         epoch_id=epoch_id,
@@ -791,12 +974,25 @@ def build_task_admission_handlers_from_activation_file(
     )
 
 
-def build_task_admission_handlers_from_environment(
+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]:
-    """Compose all three handlers from server-side trusted environment only."""
+    return build_task_admission_runtime_from_activation_file(
+        activation_file,
+        caller_service=caller_service,
+        post_admission=post_admission,
+    ).handlers
+
+
+def build_task_admission_runtime_from_environment(
+    *,
+    caller_service: str = "aios-chatgpt-bridge-v2-staging.service",
+    post_admission: Callable[[TaskAdmissionGrant, Mapping[str, Any]], Any] | None = None,
+) -> TaskAdmissionRuntimeBundle:
+    """Compose admission plus automatic TaskPrincipal recovery from trusted environment."""
     required = (
         "AIOS_TASK_ADMISSION_EPOCH_ROOT",
         "AIOS_TASK_ADMISSION_EPOCH_ID",
@@ -813,7 +1009,7 @@ def build_task_admission_handlers_from_environment(
             "TASK_ADMISSION_HOST_ENV_MISSING:"
             + ",".join(sorted(missing))
         )
-    return _build_task_admission_handlers(
+    return _build_task_admission_runtime(
         caller_service=caller_service,
         epoch_root=Path(os.environ["AIOS_TASK_ADMISSION_EPOCH_ROOT"]),
         epoch_id=os.environ["AIOS_TASK_ADMISSION_EPOCH_ID"],
@@ -840,6 +1036,17 @@ def build_task_admission_handlers_from_environment(
         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]:
+    return build_task_admission_runtime_from_environment(
+        caller_service=caller_service,
+        post_admission=post_admission,
+    ).handlers
+
+
 # ---- producer phase core v12r4 (hash-pinned sibling source) ----
 import hashlib
 import json

deploy/systemd/aios-chatgpt-bridge-v2-staging.service

@@ -11,6 +11,7 @@ WorkingDirectory=@RUNTIME_ROOT@
 Environment=AIOS_BRIDGE_SCHEMA_VERSION=aios.chatgpt-gitops-bridge.v2
 Environment=AIOS_GITOPS_WRITES_ENABLED=false
 Environment=AIOS_GITOPS_DURABLE_JOB_EXECUTION_ENABLED=true
+Environment=AIOS_P0_TASK_PRINCIPAL_WRITES_ENABLED=true
 Environment=AIOS_GITOPS_WRITES_ENABLED=false
 Environment=AIOS_GITEA_API_BASE=https://g.cnnetmkt.top/git/api/v1
 Environment=AIOS_GITEA_TOKEN_FILE=%d/gitea-governance-executor-token

docs/gitops/openapi-chatgpt-actions.yaml

@@ -410,7 +410,7 @@ paths:
       operationId: commitFilesGoverned
       summary: Submit one atomic governed commit containing multiple files
       security: [{taskPrincipalAuth: []}]
-      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.
+      x-aios-task-principal: Bridge resolves a fresh server-side TaskPrincipal from the authoritative admitted+claimed Task and re-verifies its Owner epoch, then checks exact task/job/repository/branch/generation/capability binding; caller JSON cannot provide a principal and missing verification is denied. An explicitly injected legacy bearer verifier may still be used by a trusted host.
       requestBody:
         required: true
         content:

tests/test_r7_atomic_recovery_and_bridge_sandbox.py

@@ -27,6 +27,7 @@ def test_bridge_unit_keeps_strict_sandbox_and_required_projection():
     ).read_text()
     assert "ProtectSystem=strict" in text
     assert "ProtectHome=true" in text
+    assert "Environment=AIOS_P0_TASK_PRINCIPAL_WRITES_ENABLED=true" in text
     projection = re.search(
         r"^LoadCredentialEncrypted=([^:]+):/etc/credstore\.encrypted/([^\n]+)$",
         text,

tests/test_task_admission_bootstrap_recovery.py

@@ -418,8 +418,10 @@ def test_bridge_persistent_consumer_is_opt_in_and_default_fail_closed():
         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 "build_task_admission_runtime_from_environment" in source
     assert "task_admission_handlers is None" in source
+    assert "admission_runtime.principal_verifier" in source
+    assert "TaskPrincipal" in source
 
 
 def test_bridge_persistent_host_is_opt_in_and_activation_file_is_fail_closed():
@@ -428,7 +430,7 @@ def test_bridge_persistent_host_is_opt_in_and_activation_file_is_fail_closed():
     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
+    assert 'build_task_admission_runtime_from_activation_file' in text
 
 
 def test_host_default_state_database_stays_inside_existing_bridge_state_root():

tests/test_task_principal_auto_recovery.py

@@ -0,0 +1,246 @@
+from __future__ import annotations
+
+from pathlib import Path
+import sys
+
+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,
+    TaskAdmissionError,
+    TaskAdmissionGrant,
+    TaskAdmissionStore,
+)
+from aios_task_admission_live_adapters import (  # noqa: E402
+    TaskAdmissionPrincipalProvider,
+)
+
+
+def _grant(*, capabilities=None) -> TaskAdmissionGrant:
+    return TaskAdmissionGrant(
+        grant_id="grant-1",
+        envelope_id="env-1",
+        task_id="task-1",
+        repository="aios/aios",
+        authority_base_oid="a" * 40,
+        target_branch="agent/task-1",
+        generation=1,
+        execution_digest="b" * 64,
+        capabilities=tuple(
+            capabilities
+            or (
+                "task.principal.issue",
+                "writer.lease",
+                "governed.write.file",
+                "governed.write.files",
+            )
+        ),
+        file_allowlist=("active-runtime/app/example.py",),
+        risk="medium",
+        trust_domain="aios.owner",
+        policy_digest="c" * 64,
+        authorization_kind="delegation_epoch",
+        owner_identity="owner",
+        parent_epoch_id="epoch-1",
+        issued_at=100,
+        expires_at=1000,
+        one_time_use=True,
+    )
+
+
+def _epoch(*, capabilities=None) -> OwnerDelegationEpoch:
+    return OwnerDelegationEpoch(
+        epoch_id="epoch-1",
+        owner_identity="owner",
+        trust_domain="aios.owner",
+        repository="aios/aios",
+        allowed_branch_prefixes=("agent/",),
+        allowed_capabilities=frozenset(
+            capabilities
+            or {
+                "task.principal.issue",
+                "writer.lease",
+                "governed.write.file",
+                "governed.write.files",
+            }
+        ),
+        allowed_file_prefixes=("active-runtime/app/",),
+        risk_ceiling="medium",
+        issued_at=1,
+        not_before=1,
+        expires_at=2000,
+        revocation_epoch=0,
+        policy_digest="c" * 64,
+    )
+
+
+class Store:
+    def __init__(self, grant):
+        self.grant = grant
+
+    def get_grant_by_task_id(self, task_id):
+        return self.grant if self.grant and self.grant.task_id == task_id else None
+
+
+class Authorization:
+    def __init__(self, epoch=None, error=None):
+        self.epoch = epoch or _epoch()
+        self.error = error
+        self.calls = 0
+
+    def verify_delegation_epoch(self, epoch_id):
+        self.calls += 1
+        if self.error is not None:
+            raise self.error
+        assert epoch_id == self.epoch.epoch_id
+        return self.epoch
+
+
+class Port:
+    def __init__(self, *, state="CLAIMED", job_id="job-1"):
+        self.state = state
+        self.job_id = job_id
+        self.calls = 0
+
+    def verify_active_task(self, grant):
+        self.calls += 1
+        if self.state not in {"CLAIMED", "RUNNING"}:
+            raise TaskAdmissionError(
+                "TASK_PRINCIPAL_TASK_NOT_CLAIMED", 409, details={"state": self.state}
+            )
+        return (
+            {"task_id": grant.task_id, "state": self.state},
+            {
+                "task_id": grant.task_id,
+                "repository": grant.repository,
+                "target_branch": grant.target_branch,
+                "generation": grant.generation,
+                "external_job_id": self.job_id,
+            },
+        )
+
+
+def _request(**updates):
+    value = {
+        "task_id": "task-1",
+        "job_id": "job-1",
+        "repository": "aios/aios",
+        "base_oid": "a" * 40,
+        "target_branch": "agent/task-1",
+        "generation": 1,
+    }
+    value.update(updates)
+    return value
+
+
+def test_provider_recovers_principal_from_claimed_task_without_caller_bearer():
+    auth = Authorization()
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=auth,
+        task_port=Port(state="CLAIMED"),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    result = provider({"operation": "acquireWriterLease", "headers": {}}, _request())
+    assert result["schema"] == "aios.task-principal-server-attestation.v1"
+    assert result["task_id"] == "task-1"
+    assert result["job_id"] == "job-1"
+    assert result["capability"] == "writer.lease"
+    assert result["task_state"] == "CLAIMED"
+    assert auth.calls == 1
+    assert provider.readiness()["ready"] is True
+
+
+def test_provider_requires_claim_before_writer_lease():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=Authorization(),
+        task_port=Port(state="QUEUED"),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_TASK_NOT_CLAIMED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+@pytest.mark.parametrize(
+    ("field", "value"),
+    [
+        ("job_id", "job-other"),
+        ("repository", "other/repo"),
+        ("target_branch", "agent/other"),
+        ("generation", 2),
+        ("base_oid", "d" * 40),
+    ],
+)
+def test_provider_rejects_request_binding_drift(field, value):
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=Authorization(),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_REQUEST_BINDING_MISMATCH"):
+        provider(
+            {"operation": "acquireWriterLease"},
+            _request(**{field: value}),
+        )
+
+
+def test_provider_reverifies_revocation_on_every_use():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant()),
+        authorization=Authorization(
+            error=TaskAdmissionError("TASK_ADMISSION_EPOCH_REVOKED", 403)
+        ),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_ADMISSION_EPOCH_REVOKED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+def test_provider_rejects_missing_execution_capability():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(_grant(capabilities=("task.principal.issue",))),
+        authorization=Authorization(),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_CAPABILITY_DENIED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+def test_provider_does_not_treat_missing_admission_as_owner_reauthorization():
+    provider = TaskAdmissionPrincipalProvider(
+        store=Store(None),
+        authorization=Authorization(),
+        task_port=Port(),
+        epoch_id="epoch-1",
+        clock=lambda: 200,
+    )
+    with pytest.raises(TaskAdmissionError, match="TASK_PRINCIPAL_AUTHORIZATION_REQUIRED"):
+        provider({"operation": "acquireWriterLease"}, _request())
+
+
+def test_store_recovers_exact_grant_by_task_id(tmp_path):
+    store = TaskAdmissionStore(tmp_path / "admission.sqlite3")
+    grant = _grant()
+    store.save_grant_and_receipt(
+        envelope_id=grant.envelope_id,
+        grant=grant,
+        receipt={"schema": "fixture", "task_id": grant.task_id},
+        created_at=200,
+    )
+    recovered = store.get_grant_by_task_id(grant.task_id)
+    assert recovered == grant
+    assert store.get_grant_by_task_id("task-missing") is None