governance: scope Gitea 1.22 token-row revocation

approvals/governance-policy/AIOS-V2R1C4-TASK-STATUS-MERGE-BROKER-BOOTSTRAP-20260801T004650Z.json

@@ -0,0 +1,113 @@
+{
+  "approval_type": "TASK_STATUS_MERGE_BROKER_BOOTSTRAP",
+  "schema_version": 1,
+  "repository": "aios/aios",
+  "expected_governance_base_oid": "8814dff20222625b60454c613b61ccf15fa20e51",
+  "request_branch": "governance/bootstrap/task-status-merge-broker-20260801t0036z",
+  "issued_at": "2026-08-01T00:46:50Z",
+  "expires_at": "2026-08-01T12:46:50Z",
+  "required_human_approvals": 1,
+  "allowed_human_approvers": ["owner"],
+  "require_latest_head_approval": true,
+  "purpose": "Activate a task-scoped credential, status, and fast-forward-only merge broker so a new V2R1C4 Phase B renewal can satisfy existing branch protection without impersonation or long-lived credentials.",
+  "phase_b_transition": {
+    "old_pr_number": 36,
+    "old_head_oid": "a8644fc172bd8ebbba17a16e6bf586c93fd1e793",
+    "reuse_after_governance_base_change": false,
+    "renewal_required_after_bootstrap_merge": true
+  },
+  "allowed_status_contexts": [
+    "aios/branch-protection",
+    "aios/evidence-closure",
+    "aios/secret-scan",
+    "aios/state-machine",
+    "aios/validator"
+  ],
+  "allowed_identities": ["codex", "gpt"],
+  "allowed_operations": [
+    "issue_task_status_credential",
+    "issue_task_merge_credential",
+    "sign_task_grant_after_owner_exact_head_review",
+    "publish_exact_five_statuses",
+    "fast_forward_only_merge",
+    "readback_status_and_ref",
+    "revoke_task_credentials",
+    "remove_credential_projection"
+    ,"sweep_completed_or_expired_task_credentials"
+  ],
+  "operation_limits": {
+    "maximum_status_writes_per_grant": 5,
+    "maximum_merges_per_grant": 1,
+    "maximum_grant_lifetime_seconds": 43200,
+    "automatic_revocation_required": true
+  },
+  "denied_operations": [
+    "branch_push",
+    "main_push",
+    "tag_write",
+    "registry_write",
+    "baseline_write",
+    "production_write",
+    "branch_protection_change",
+    "force_push",
+    "squash",
+    "rebase",
+    "merge_commit",
+    "approval_impersonation",
+    "plaintext_secret_persistence"
+    ,"sign_unapproved_grant"
+    ,"reuse_superseded_pr"
+    ,"grant_signing_private_key_projection_to_broker"
+    ,"commit_status_database_write"
+    ,"database_write_outside_exact_access_token_revocation"
+  ],
+  "runtime_controls": {
+    "global_gitops_writes_enabled": false,
+    "runtime_secret_only": true,
+    "systemd_encrypted_credentials_required": true,
+    "token_in_argv_allowed": false,
+    "token_in_log_allowed": false,
+    "separate_non_root_status_identity": true,
+    "separate_non_root_merge_identity": true,
+    "root_credential_broker_only": true,
+    "sqlite_quota_and_hash_chain_audit": true,
+    "hmac_audit_signature": true,
+    "detached_ssh_grant_signature_required": true,
+    "broker_receives_public_verification_key_only": true,
+    "grant_signing_requires_live_owner_exact_head_review": true,
+    "grant_signing_requires_open_pr_exact_head_base_tree_manifest": true,
+    "status_evidence_digest_bound_in_signed_grant": true,
+    "credential_issue_rechecks_local_pr_review_ref_tree_manifest": true,
+    "gitea_write_repository_scope_compensated_by_locked_broker_sandbox": true,
+    "token_revocation_exact_row_and_http_401_required": true,
+    "gitea_1_22_token_delete_requires_basic_auth": true,
+    "exact_access_token_row_revocation_exception": true,
+    "access_token_revocation_binding_fields": ["id", "uid", "name", "scope"],
+    "commit_status_database_write_allowed": false,
+    "credential_completion_and_expiry_sweep_interval_seconds": 60,
+    "sweeper_uses_same_per_task_audit_credential_as_issuer": true,
+    "gitea_api_origin": "http://127.0.0.1:3000/api/v1",
+    "merge_method": "fast-forward-only"
+  },
+  "approved_files": {
+    "services/aios-single-owner-governance/aios_single_owner_governance/task_status_merge_broker.py": "79a7b6807dc9e6c167c94adbd3bb4a1b9ae17e54d2d6fd19bd6a38ce2f8b8e3f",
+    "services/aios-single-owner-governance/aios_single_owner_governance/task_credential_broker.py": "678fccd8e61fa1806d44fbb81fe7f6e2a7b63bbd99d1323516073100b6e48ab5",
+    "services/aios-single-owner-governance/aios_single_owner_governance/task_credential_expiry_sweeper.py": "8d383768eb0322c84a35282c126bee26a3e6419b125780424ffc7793a504efdf",
+    "services/aios-single-owner-governance/task-status-merge-grant-v1.schema.json": "4528aa6dfeaa151deadb54bf1c462e1ff3e9526e0da3eb20144e6c01c19998f0",
+    "tests/single_owner_governance/test_task_status_merge_broker.py": "65a285b29798b72d13e30562ce97529780643f3739e5d07c283dfa5995035dc4",
+    "deploy/systemd/aios-task-status-broker@.service": "378a2e8f658cb56cac5dcc151772ec6a00d86f6c1ce3106509b9aa87549dc9da",
+    "deploy/systemd/aios-task-merge-broker@.service": "4e6b8fcf541f225e58ea080107fc30deb0c4ac9eac70217ff9e57081245c106a",
+    "deploy/systemd/aios-task-credential-broker@.service": "ec5dd2be87f3e02af89a40a615ccbd76a277f46aa29cc1a92fee5496d750cbce",
+    "deploy/systemd/aios-task-credential-expiry-sweeper.service": "39726537a3b4f57614eba678309fd3009c8fa25448458827aef7ece0015be10d",
+    "deploy/systemd/aios-task-credential-expiry-sweeper.timer": "49ff9c80cbd0b44fd584dc7d29f4db0d042e33f325b1ed7af81a5be42f1cd011"
+  },
+  "machine_gate_requirements": {
+    "compileall": "PASS",
+    "unit_tests": "PASS",
+    "json_schema": "PASS",
+    "systemd_validation": "PASS",
+    "secret_scan": "PASS",
+    "fault_injection": "PASS",
+    "scope_boundary_tests": "PASS"
+  }
+}

deploy/systemd/aios-task-credential-broker@.service

@@ -0,0 +1,24 @@
+[Unit]
+Description=AIOS root task credential broker (%i)
+ConditionPathExists=/var/lib/aios-gitops/runtime/task-status-merge/%i/grant.json
+
+[Service]
+Type=oneshot
+User=root
+Group=root
+LoadCredentialEncrypted=audit-key:/run/aios-task-credential-broker/%i-audit.cred
+ExecStart=/usr/bin/python3 -m aios_single_owner_governance.task_credential_broker --grant /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.json --grant-signature /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.sig --allowed-signers /etc/aios-gitops/task-status-merge/allowed_signers --audit-key %d/audit-key --audit-dir /var/lib/aios-gitops/runtime/task-status-merge/%i/audit --request /var/lib/aios-gitops/runtime/task-status-merge/%i/credential-request.json
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+NoNewPrivileges=false
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ReadOnlyPaths=/data/aios-gitops/runtime/bin/gitea /data/aios-gitops/runtime/gitea-bin/custom/conf/app.ini /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.json /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.sig /var/lib/aios-gitops/runtime/task-status-merge/%i/credential-request.json /etc/aios-gitops/task-status-merge/allowed_signers
+ReadWritePaths=/data/aios-gitops/runtime/gitea-bin/data/gitea.db /run/aios-task-credential-broker /var/lib/aios-gitops/runtime/task-status-merge/%i/audit
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=localhost
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictSUIDSGID=true

deploy/systemd/aios-task-credential-expiry-sweeper.service

@@ -0,0 +1,23 @@
+[Unit]
+Description=AIOS task credential expiry and completion sweeper
+After=aios-gitea.service
+Requires=aios-gitea.service
+
+[Service]
+Type=oneshot
+User=root
+Group=root
+ExecStart=/usr/bin/python3 -m aios_single_owner_governance.task_credential_expiry_sweeper --runtime-root /var/lib/aios-gitops/runtime/task-status-merge --allowed-signers /etc/aios-gitops/task-status-merge/allowed_signers
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+ReadOnlyPaths=/var/lib/aios-gitops/runtime/task-status-merge /etc/aios-gitops/task-status-merge/allowed_signers
+ReadWritePaths=/data/aios-gitops/runtime/gitea-bin/data/gitea.db /run/aios-task-credential-broker /var/lib/aios-gitops/runtime/task-status-merge
+RestrictAddressFamilies=AF_UNIX AF_INET
+IPAddressDeny=any
+IPAddressAllow=localhost
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictSUIDSGID=true

deploy/systemd/aios-task-credential-expiry-sweeper.timer

@@ -0,0 +1,12 @@
+[Unit]
+Description=Run AIOS task credential expiry sweeper every minute
+
+[Timer]
+OnBootSec=1min
+OnUnitActiveSec=1min
+AccuracySec=5s
+Persistent=true
+Unit=aios-task-credential-expiry-sweeper.service
+
+[Install]
+WantedBy=timers.target

deploy/systemd/aios-task-merge-broker@.service

@@ -0,0 +1,26 @@
+[Unit]
+Description=AIOS task-scoped Gitea fast-forward-only merge broker (%i)
+After=network.target aios-gitea.service aios-task-status-broker@%i.service
+Requires=aios-gitea.service aios-task-status-broker@%i.service
+
+[Service]
+Type=oneshot
+User=aios-task-merge-bot
+Group=aios-task-merge-bot
+LoadCredentialEncrypted=gitea-token:/run/aios-task-credential-broker/%i-merge.cred
+LoadCredentialEncrypted=audit-key:/run/aios-task-credential-broker/%i-audit.cred
+ExecStart=/usr/bin/python3 -m aios_single_owner_governance.task_status_merge_broker --mode merge --grant /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.json --grant-signature /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.sig --allowed-signers /etc/aios-gitops/task-status-merge/allowed_signers --token %d/gitea-token --audit-key %d/audit-key --journal /var/lib/aios-gitops/runtime/task-status-merge/%i/merge-journal/journal.sqlite3
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+RestrictAddressFamilies=AF_UNIX AF_INET
+ReadOnlyPaths=/var/lib/aios-gitops/runtime/task-status-merge/%i /etc/aios-gitops/task-status-merge/allowed_signers
+ReadWritePaths=/var/lib/aios-gitops/runtime/task-status-merge/%i/merge-journal
+IPAddressDeny=any
+IPAddressAllow=localhost
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictSUIDSGID=true

deploy/systemd/aios-task-status-broker@.service

@@ -0,0 +1,26 @@
+[Unit]
+Description=AIOS task-scoped Gitea status broker (%i)
+After=network.target aios-gitea.service
+Requires=aios-gitea.service
+
+[Service]
+Type=oneshot
+User=aios-task-status-bot
+Group=aios-task-status-bot
+LoadCredentialEncrypted=gitea-token:/run/aios-task-credential-broker/%i-status.cred
+LoadCredentialEncrypted=audit-key:/run/aios-task-credential-broker/%i-audit.cred
+ExecStart=/usr/bin/python3 -m aios_single_owner_governance.task_status_merge_broker --mode status --grant /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.json --grant-signature /var/lib/aios-gitops/runtime/task-status-merge/%i/grant.sig --allowed-signers /etc/aios-gitops/task-status-merge/allowed_signers --evidence /var/lib/aios-gitops/runtime/task-status-merge/%i/status-evidence.json --token %d/gitea-token --audit-key %d/audit-key --journal /var/lib/aios-gitops/runtime/task-status-merge/%i/status-journal/journal.sqlite3
+Environment=AIOS_GITOPS_WRITES_ENABLED=false
+NoNewPrivileges=true
+PrivateTmp=true
+PrivateDevices=true
+ProtectSystem=strict
+ProtectHome=true
+RestrictAddressFamilies=AF_UNIX AF_INET
+ReadOnlyPaths=/var/lib/aios-gitops/runtime/task-status-merge/%i /etc/aios-gitops/task-status-merge/allowed_signers
+ReadWritePaths=/var/lib/aios-gitops/runtime/task-status-merge/%i/status-journal
+IPAddressDeny=any
+IPAddressAllow=localhost
+LockPersonality=true
+MemoryDenyWriteExecute=true
+RestrictSUIDSGID=true

services/aios-single-owner-governance/aios_single_owner_governance/task_credential_broker.py

@@ -0,0 +1,237 @@
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import os
+from pathlib import Path
+import sqlite3
+import subprocess
+from datetime import datetime, timezone
+import urllib.error
+import urllib.request
+
+from .task_status_merge_broker import ContractDenied, TaskGrant, canonical, load_signed_task_grant, read_runtime_secret
+
+
+GITEA_BIN = Path("/data/aios-gitops/runtime/bin/gitea")
+GITEA_CONFIG = Path("/data/aios-gitops/runtime/gitea-bin/custom/conf/app.ini")
+GITEA_DB = Path("/data/aios-gitops/runtime/gitea-bin/data/gitea.db")
+RUNUSER = Path("/usr/sbin/runuser")
+SYSTEMD_CREDS = Path("/usr/bin/systemd-creds")
+GIT = Path("/usr/bin/git")
+GITEA_REPOSITORY = Path("/data/aios-gitops/runtime/gitea-bin/data/repositories/aios/aios.git")
+
+
+def _git(*arguments: str, input_bytes: bytes | None = None) -> bytes:
+    return subprocess.run(
+        [str(GIT), f"--git-dir={GITEA_REPOSITORY}", *arguments], input=input_bytes,
+        check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+    ).stdout
+
+
+def _verify_local_governance_binding(grant: TaskGrant) -> None:
+    """Verify the exact open PR and latest Owner review without issuing a token."""
+    connection = sqlite3.connect(f"file:{GITEA_DB}?mode=ro", uri=True)
+    connection.row_factory = sqlite3.Row
+    try:
+        pr = connection.execute(
+            "SELECT pr.head_branch,pr.base_branch,pr.merge_base,pr.has_merged,i.is_closed "
+            "FROM pull_request pr JOIN issue i ON i.id=pr.issue_id JOIN repository repo ON repo.id=i.repo_id "
+            "WHERE repo.owner_name='aios' AND repo.lower_name='aios' AND pr.[index]=?",
+            (grant.raw["pr_number"],),
+        ).fetchone()
+        review = connection.execute(
+            "SELECT r.type,r.commit_id,r.official,r.stale,r.dismissed,r.created_unix,u.id,u.lower_name "
+            "FROM review r JOIN user u ON u.id=r.reviewer_id JOIN issue i ON i.id=r.issue_id "
+            "JOIN repository repo ON repo.id=i.repo_id "
+            "WHERE repo.owner_name='aios' AND repo.lower_name='aios' AND i.[index]=? "
+            "AND u.id=2 AND u.lower_name='owner' ORDER BY r.id DESC LIMIT 1",
+            (grant.raw["pr_number"],),
+        ).fetchone()
+    finally:
+        connection.close()
+    if pr is None or pr["base_branch"] != "governance/approvals" or pr["merge_base"] != grant.raw["base_oid"] or pr["has_merged"] != 0 or pr["is_closed"] != 0:
+        raise ContractDenied("LOCAL_PR_BINDING_INVALID")
+    head = _git("rev-parse", f"refs/heads/{pr['head_branch']}").decode().strip()
+    base = _git("rev-parse", "refs/heads/governance/approvals").decode().strip()
+    if head != grant.raw["target_commit"] or base != grant.raw["base_oid"]:
+        raise ContractDenied("LOCAL_REF_BINDING_INVALID")
+    if review is None or review["type"] != 1 or review["commit_id"] != head or review["official"] != 1 or review["stale"] != 0 or review["dismissed"] != 0:
+        raise ContractDenied("LOCAL_OWNER_APPROVAL_MISSING")
+    submitted = datetime.fromtimestamp(review["created_unix"], tz=timezone.utc)
+    if submitted < grant.issued_at or submitted >= grant.expires_at:
+        raise ContractDenied("LOCAL_OWNER_APPROVAL_EXPIRED")
+    tree = _git("show", "-s", "--format=%T", head).decode().strip()
+    if tree != grant.raw["approved_tree_oid"]:
+        raise ContractDenied("APPROVED_TREE_MISMATCH")
+    entry = _git("ls-tree", head, "--", grant.raw["manifest_path"]).decode().strip().split()
+    if len(entry) < 3 or entry[1] != "blob" or entry[2] != grant.raw["manifest_blob_oid"]:
+        raise ContractDenied("MANIFEST_BLOB_BINDING_INVALID")
+    manifest = _git("cat-file", "blob", grant.raw["manifest_blob_oid"])
+    if hashlib.sha256(manifest).hexdigest() != grant.raw["manifest_sha256"]:
+        raise ContractDenied("MANIFEST_SHA256_MISMATCH")
+
+
+def _safe_absolute(path: Path, expected_parent: Path) -> None:
+    if not path.is_absolute() or path.is_symlink() or path.parent != expected_parent:
+        raise ContractDenied("CREDENTIAL_PATH_UNSAFE")
+
+
+class TaskCredentialBroker:
+    def __init__(self, grant: TaskGrant, audit_dir: Path, audit_key: bytes):
+        self.grant = grant
+        self.audit_key = bytes(audit_key)
+        if not self.audit_key:
+            raise ContractDenied("CREDENTIAL_AUDIT_KEY_EMPTY")
+        self.audit_dir = Path(audit_dir)
+        if not self.audit_dir.is_absolute() or self.audit_dir.is_symlink():
+            raise ContractDenied("AUDIT_DIR_UNSAFE")
+
+    def provision(self, *, identity: str, token_name: str, encrypted_path: Path) -> dict:
+        if os.geteuid() != 0:
+            raise ContractDenied("ROOT_BROKER_REQUIRED")
+        expected = self.grant.raw["status_identity"] if token_name.endswith("-status") else self.grant.raw["merge_identity"]
+        if identity != expected or identity not in {"codex", "gpt"}:
+            raise ContractDenied("IDENTITY_BINDING_INVALID")
+        prefix = f"pr{self.grant.raw['pr_number']}-{self.grant.raw['target_commit'][:12]}"
+        if token_name not in {f"{prefix}-status", f"{prefix}-merge"}:
+            raise ContractDenied("TOKEN_NAME_INVALID")
+        _safe_absolute(encrypted_path, Path("/run/aios-task-credential-broker"))
+        if encrypted_path.exists():
+            raise ContractDenied("CREDENTIAL_PATH_EXISTS")
+        _verify_local_governance_binding(self.grant)
+        command = [str(RUNUSER), "-u", "aiman", "--", str(GITEA_BIN), "admin", "user", "generate-access-token", "--config", str(GITEA_CONFIG), "--username", identity, "--token-name", token_name, "--scopes", "write:repository", "--raw"]
+        issued = subprocess.run(command, check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).stdout.strip()
+        if len(issued) < 32 or b"\n" in issued or b"\x00" in issued:
+            raise ContractDenied("ISSUED_TOKEN_INVALID")
+        subprocess.run([str(SYSTEMD_CREDS), "encrypt", "-", str(encrypted_path)], input=issued, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+        os.chmod(encrypted_path, 0o400)
+        connection = sqlite3.connect(f"file:{GITEA_DB}?mode=ro", uri=True)
+        try:
+            row = connection.execute("SELECT t.id,u.id FROM access_token t JOIN user u ON u.id=t.uid WHERE u.lower_name=? AND t.name=? AND t.scope='write:repository'", (identity, token_name)).fetchone()
+        finally:
+            connection.close()
+        if row is None:
+            raise ContractDenied("TOKEN_DATABASE_BINDING_MISSING")
+        receipt = {"status": "PASS", "operation": "credential_provision", "identity": identity, "token_name": token_name, "token_id": int(row[0]), "token_uid": int(row[1]), "grant_sha256": self.grant.digest, "credential_ref": f"systemd-credential-encrypted://{encrypted_path.name}"}
+        return self._audit(receipt)
+
+    def revoke(self, *, receipt: dict, encrypted_path: Path) -> dict:
+        if os.geteuid() != 0:
+            raise ContractDenied("ROOT_BROKER_REQUIRED")
+        _safe_absolute(encrypted_path, Path("/run/aios-task-credential-broker"))
+        self.validate_provision_receipt(receipt, encrypted_path)
+        decrypted = subprocess.run([str(SYSTEMD_CREDS), "decrypt", str(encrypted_path), "-"], check=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE).stdout.strip()
+        if len(decrypted) < 32 or b"\n" in decrypted or b"\x00" in decrypted:
+            raise ContractDenied("PROJECTED_TOKEN_INVALID")
+        token_header = {"Authorization": "token " + decrypted.decode("utf-8")}
+        preflight = urllib.request.Request("http://127.0.0.1:3000/api/v1/repos/aios/aios", headers=token_header)
+        try:
+            with urllib.request.urlopen(preflight, timeout=10) as response:
+                if response.status != 200:
+                    raise ContractDenied("TOKEN_PREFLIGHT_STATUS_INVALID")
+        except urllib.error.HTTPError as exc:
+            raise ContractDenied(f"TOKEN_PREFLIGHT_FAILED:{exc.code}") from exc
+        connection = sqlite3.connect(GITEA_DB)
+        try:
+            connection.execute("BEGIN IMMEDIATE")
+            cursor = connection.execute(
+                "DELETE FROM access_token WHERE id=? AND uid=? AND name=? AND scope='write:repository'",
+                (receipt["token_id"], receipt["token_uid"], receipt["token_name"]),
+            )
+            if cursor.rowcount != 1:
+                raise ContractDenied("TOKEN_REVOCATION_MISMATCH")
+            connection.commit()
+        except Exception:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+        connection = sqlite3.connect(f"file:{GITEA_DB}?mode=ro", uri=True)
+        try:
+            remaining = connection.execute("SELECT COUNT(*) FROM access_token WHERE id=?", (receipt["token_id"],)).fetchone()[0]
+        finally:
+            connection.close()
+        if remaining != 0:
+            raise ContractDenied("TOKEN_REVOCATION_READBACK_MISMATCH")
+        request = urllib.request.Request("http://127.0.0.1:3000/api/v1/repos/aios/aios", headers=token_header)
+        try:
+            urllib.request.urlopen(request, timeout=10)
+            raise ContractDenied("REVOKED_TOKEN_STILL_ACCEPTED")
+        except urllib.error.HTTPError as exc:
+            if exc.code != 401:
+                raise ContractDenied("REVOCATION_HTTP_STATUS_INVALID") from exc
+        encrypted_path.unlink(missing_ok=True)
+        result = {"status": "PASS", "operation": "credential_revoke", "identity": receipt["identity"], "token_name": receipt["token_name"], "token_id": receipt["token_id"], "grant_sha256": self.grant.digest, "token_rows_remaining": 0, "credential_projection_remaining": False, "old_token_http_status": 401}
+        return self._audit(result)
+
+    def validate_provision_receipt(self, receipt: dict, encrypted_path: Path) -> None:
+        required = {"status", "operation", "identity", "token_name", "token_id", "token_uid", "grant_sha256", "credential_ref", "audit_sha256", "audit_hmac_signature"}
+        expected_credential_ref = f"systemd-credential-encrypted://{encrypted_path.name}"
+        unsigned_receipt = {key: value for key, value in receipt.items() if key not in {"audit_sha256", "audit_hmac_signature"}}
+        expected_name_prefix = f"pr{self.grant.raw['pr_number']}-{self.grant.raw['target_commit'][:12]}"
+        if (set(receipt) != required or receipt["status"] != "PASS" or receipt["operation"] != "credential_provision"
+                or receipt["grant_sha256"] != self.grant.digest or receipt["identity"] not in {self.grant.raw["status_identity"], self.grant.raw["merge_identity"]}
+                or receipt["token_name"] not in {f"{expected_name_prefix}-status", f"{expected_name_prefix}-merge"}
+                or receipt["credential_ref"] != expected_credential_ref
+                or hashlib.sha256(canonical(unsigned_receipt)).hexdigest() != receipt["audit_sha256"]
+                or not hmac.compare_digest(receipt["audit_hmac_signature"], hmac.new(self.audit_key, canonical(unsigned_receipt), hashlib.sha256).hexdigest())):
+            raise ContractDenied("PROVISION_RECEIPT_INVALID")
+
+    def _audit(self, value: dict) -> dict:
+        raw = canonical(value)
+        result = dict(value)
+        result["audit_sha256"] = hashlib.sha256(raw).hexdigest()
+        result["audit_hmac_signature"] = hmac.new(self.audit_key, raw, hashlib.sha256).hexdigest()
+        self.audit_dir.mkdir(mode=0o750, parents=True, exist_ok=True)
+        destination = self.audit_dir / f"{value['operation']}-{value['token_name']}.json"
+        descriptor = os.open(destination, os.O_WRONLY | os.O_CREAT | os.O_EXCL | getattr(os, "O_NOFOLLOW", 0), 0o640)
+        try:
+            os.write(descriptor, canonical(result) + b"\n")
+            os.fsync(descriptor)
+        finally:
+            os.close(descriptor)
+        return result
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="aios-task-credential-broker")
+    parser.add_argument("--grant", required=True, type=Path)
+    parser.add_argument("--audit-dir", required=True, type=Path)
+    parser.add_argument("--operation", choices=("provision", "revoke"))
+    parser.add_argument("--identity")
+    parser.add_argument("--token-name")
+    parser.add_argument("--encrypted-path", type=Path)
+    parser.add_argument("--receipt", type=Path)
+    parser.add_argument("--request", type=Path)
+    parser.add_argument("--grant-signature", required=True, type=Path)
+    parser.add_argument("--allowed-signers", required=True, type=Path)
+    parser.add_argument("--audit-key", required=True, type=Path)
+    args = parser.parse_args(argv)
+    grant = load_signed_task_grant(args.grant, args.grant_signature, args.allowed_signers)
+    broker = TaskCredentialBroker(grant, args.audit_dir, read_runtime_secret(args.audit_key))
+    if args.request is not None:
+        request = json.loads(args.request.read_text(encoding="utf-8"))
+        if not isinstance(request, dict) or set(request) not in ({"operation", "identity", "token_name", "encrypted_path"}, {"operation", "identity", "token_name", "encrypted_path", "receipt"}):
+            raise ContractDenied("CREDENTIAL_REQUEST_INVALID")
+        args.operation = request["operation"]
+        args.identity = request["identity"]
+        args.token_name = request["token_name"]
+        args.encrypted_path = Path(request["encrypted_path"])
+        args.receipt = Path(request["receipt"]) if "receipt" in request else None
+    if args.operation is None or args.identity is None or args.token_name is None or args.encrypted_path is None:
+        raise ContractDenied("CREDENTIAL_REQUEST_INCOMPLETE")
+    if args.operation == "provision":
+        result = broker.provision(identity=args.identity, token_name=args.token_name, encrypted_path=args.encrypted_path)
+    else:
+        if args.receipt is None:
+            raise ContractDenied("RECEIPT_REQUIRED")
+        result = broker.revoke(receipt=json.loads(args.receipt.read_text(encoding="utf-8")), encrypted_path=args.encrypted_path)
+    print(json.dumps(result, sort_keys=True, separators=(",", ":")))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-single-owner-governance/aios_single_owner_governance/task_credential_expiry_sweeper.py

@@ -0,0 +1,94 @@
+from __future__ import annotations
+
+import argparse
+from datetime import datetime, timezone
+import json
+import os
+from pathlib import Path
+import sqlite3
+import subprocess
+
+from .task_credential_broker import SYSTEMD_CREDS, TaskCredentialBroker
+from .task_status_merge_broker import ContractDenied, load_signed_task_grant
+
+
+RUNTIME_ROOT = Path("/var/lib/aios-gitops/runtime/task-status-merge")
+CREDENTIAL_ROOT = Path("/run/aios-task-credential-broker")
+
+
+def _operation_complete(task_dir: Path, token_name: str) -> bool:
+    if token_name.endswith("-status"):
+        journal, operation = task_dir / "status-journal" / "journal.sqlite3", "statuses"
+    elif token_name.endswith("-merge"):
+        journal, operation = task_dir / "merge-journal" / "journal.sqlite3", "merge"
+    else:
+        raise ContractDenied("SWEEPER_TOKEN_NAME_INVALID")
+    if not journal.is_file() or journal.is_symlink():
+        return False
+    connection = sqlite3.connect(f"file:{journal}?mode=ro", uri=True)
+    try:
+        row = connection.execute("SELECT state FROM quota WHERE operation=?", (operation,)).fetchone()
+    finally:
+        connection.close()
+    return row == ("COMMITTED",)
+
+
+def sweep(runtime_root: Path, allowed_signers: Path, *, now: datetime | None = None) -> list[dict]:
+    if os.geteuid() != 0:
+        raise ContractDenied("ROOT_SWEEPER_REQUIRED")
+    if runtime_root != RUNTIME_ROOT or runtime_root.is_symlink():
+        raise ContractDenied("SWEEPER_ROOT_INVALID")
+    current = now or datetime.now(timezone.utc)
+    results: list[dict] = []
+    for task_dir in sorted(runtime_root.iterdir() if runtime_root.exists() else []):
+        if not task_dir.is_dir() or task_dir.is_symlink():
+            raise ContractDenied("SWEEPER_TASK_DIR_INVALID")
+        grant_path, signature_path = task_dir / "grant.json", task_dir / "grant.sig"
+        if not grant_path.is_file() or not signature_path.is_file():
+            raise ContractDenied("SWEEPER_GRANT_INCOMPLETE")
+        grant = load_signed_task_grant(grant_path, signature_path, allowed_signers, now=current, allow_expired=True)
+        audit_credential = CREDENTIAL_ROOT / f"{task_dir.name}-audit.cred"
+        if not audit_credential.is_file() or audit_credential.is_symlink():
+            raise ContractDenied("SWEEPER_TASK_AUDIT_CREDENTIAL_MISSING")
+        audit_key = subprocess.run(
+            [str(SYSTEMD_CREDS), "decrypt", str(audit_credential), "-"], check=True,
+            stdout=subprocess.PIPE, stderr=subprocess.PIPE,
+        ).stdout.strip()
+        if not audit_key or b"\n" in audit_key or b"\x00" in audit_key:
+            raise ContractDenied("SWEEPER_TASK_AUDIT_KEY_INVALID")
+        broker = TaskCredentialBroker(grant, task_dir / "audit", audit_key)
+        for receipt_path in sorted((task_dir / "audit").glob("credential_provision-*.json")):
+            if receipt_path.is_symlink():
+                raise ContractDenied("SWEEPER_RECEIPT_UNSAFE")
+            receipt = json.loads(receipt_path.read_text(encoding="utf-8"))
+            token_name = str(receipt.get("token_name") or "")
+            credential_ref = str(receipt.get("credential_ref") or "")
+            prefix = "systemd-credential-encrypted://"
+            if not credential_ref.startswith(prefix):
+                raise ContractDenied("SWEEPER_CREDENTIAL_REF_INVALID")
+            encrypted_path = CREDENTIAL_ROOT / credential_ref.removeprefix(prefix)
+            revoked_receipt = task_dir / "audit" / f"credential_revoke-{token_name}.json"
+            if revoked_receipt.exists():
+                if encrypted_path.exists():
+                    raise ContractDenied("SWEEPER_REVOKED_PROJECTION_REAPPEARED")
+                continue
+            if current < grant.expires_at and not _operation_complete(task_dir, token_name):
+                continue
+            if not encrypted_path.is_file() or encrypted_path.is_symlink():
+                raise ContractDenied("SWEEPER_CREDENTIAL_PROJECTION_MISSING")
+            results.append(broker.revoke(receipt=receipt, encrypted_path=encrypted_path))
+    return results
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="aios-task-credential-expiry-sweeper")
+    parser.add_argument("--runtime-root", required=True, type=Path)
+    parser.add_argument("--allowed-signers", required=True, type=Path)
+    args = parser.parse_args(argv)
+    result = sweep(args.runtime_root, args.allowed_signers)
+    print(json.dumps({"status": "PASS", "revoked": result}, sort_keys=True, separators=(",", ":")))
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-single-owner-governance/aios_single_owner_governance/task_status_merge_broker.py

@@ -0,0 +1,436 @@
+from __future__ import annotations
+
+import argparse
+from dataclasses import dataclass
+from datetime import datetime, timezone
+import hashlib
+import hmac
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+import stat
+import subprocess
+from typing import Any, Protocol
+import urllib.error
+import urllib.parse
+import urllib.request
+
+
+ALLOWED_CONTEXTS = frozenset({
+    "aios/branch-protection",
+    "aios/evidence-closure",
+    "aios/secret-scan",
+    "aios/state-machine",
+    "aios/validator",
+})
+ALLOWED_IDENTITIES = frozenset({"codex", "gpt"})
+OID = re.compile(r"^[0-9a-f]{40}$")
+SHA256 = re.compile(r"^[0-9a-f]{64}$")
+SECRET_REF = re.compile(r"^systemd-credential://[A-Za-z0-9_.-]+$")
+SUPERSEDED_PR_NUMBER = 36
+SUPERSEDED_HEAD_OID = "a8644fc172bd8ebbba17a16e6bf586c93fd1e793"
+FIELDS = frozenset({
+    "schema_id", "schema_version", "repository", "pr_number", "base_oid",
+    "target_commit", "approved_tree_oid", "manifest_path", "manifest_blob_oid", "manifest_sha256",
+    "payload_sha256", "status_identity", "merge_identity", "status_contexts",
+    "status_write_quota", "merge_quota", "merge_method", "branch_push_allowed",
+    "tag_write_allowed", "main_write_allowed", "registry_write_allowed",
+    "baseline_write_allowed", "production_write_allowed", "issued_at", "expires_at",
+    "audit_key_ref", "grant_signer_identity", "grant_signature_namespace",
+    "status_evidence_sha256", "supersedes_pr_number", "supersedes_head_oid",
+    "renewal_required",
+})
+
+
+class ContractDenied(RuntimeError):
+    pass
+
+
+def canonical(value: Any) -> bytes:
+    return json.dumps(value, sort_keys=True, separators=(",", ":")).encode()
+
+
+def _time(value: str) -> datetime:
+    if not isinstance(value, str) or not value.endswith("Z"):
+        raise ContractDenied("TIMESTAMP_INVALID")
+    try:
+        return datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ContractDenied("TIMESTAMP_INVALID") from exc
+
+
+@dataclass(frozen=True)
+class TaskGrant:
+    raw: dict[str, Any]
+    issued_at: datetime
+    expires_at: datetime
+
+    @classmethod
+    def from_dict(cls, value: dict[str, Any], *, now: datetime | None = None,
+                  allow_expired: bool = False) -> "TaskGrant":
+        if not isinstance(value, dict) or set(value) != FIELDS:
+            raise ContractDenied("GRANT_FIELDS_INVALID")
+        if value["schema_id"] != "AIOS_TASK_STATUS_MERGE_GRANT" or value["schema_version"] != 1:
+            raise ContractDenied("GRANT_SCHEMA_INVALID")
+        if value["repository"] != "aios/aios" or type(value["pr_number"]) is not int or value["pr_number"] <= 0:
+            raise ContractDenied("REPOSITORY_OR_PR_MISMATCH")
+        for field in ("base_oid", "target_commit", "approved_tree_oid", "manifest_blob_oid"):
+            if not isinstance(value[field], str) or not OID.fullmatch(value[field]):
+                raise ContractDenied("OID_INVALID")
+        if not isinstance(value["manifest_path"], str) or not re.fullmatch(r"approvals/[A-Za-z0-9._/-]+\.json", value["manifest_path"]) or ".." in value["manifest_path"].split("/"):
+            raise ContractDenied("MANIFEST_PATH_INVALID")
+        for field in ("manifest_sha256", "payload_sha256"):
+            if not isinstance(value[field], str) or not SHA256.fullmatch(value[field]):
+                raise ContractDenied("SHA256_INVALID")
+        evidence = value["status_evidence_sha256"]
+        if not isinstance(evidence, dict) or set(evidence) != ALLOWED_CONTEXTS or any(not isinstance(item, str) or not SHA256.fullmatch(item) for item in evidence.values()):
+            raise ContractDenied("STATUS_EVIDENCE_BINDING_INVALID")
+        if value["supersedes_pr_number"] != SUPERSEDED_PR_NUMBER or value["pr_number"] == SUPERSEDED_PR_NUMBER:
+            raise ContractDenied("RENEWAL_PR_INVALID")
+        if value["supersedes_head_oid"] != SUPERSEDED_HEAD_OID or value["supersedes_head_oid"] == value["target_commit"] or value["renewal_required"] is not True:
+            raise ContractDenied("RENEWAL_BINDING_INVALID")
+        if value["status_identity"] not in ALLOWED_IDENTITIES or value["merge_identity"] not in ALLOWED_IDENTITIES:
+            raise ContractDenied("IDENTITY_NOT_ALLOWED")
+        if set(value["status_contexts"]) != ALLOWED_CONTEXTS or len(value["status_contexts"]) != 5:
+            raise ContractDenied("STATUS_CONTEXTS_INVALID")
+        if value["status_write_quota"] != 5 or value["merge_quota"] != 1:
+            raise ContractDenied("QUOTA_INVALID")
+        if value["merge_method"] != "fast-forward-only":
+            raise ContractDenied("MERGE_METHOD_INVALID")
+        denied = (
+            "branch_push_allowed", "tag_write_allowed", "main_write_allowed",
+            "registry_write_allowed", "baseline_write_allowed", "production_write_allowed",
+        )
+        if any(value[field] is not False for field in denied):
+            raise ContractDenied("WRITE_SCOPE_EXPANDED")
+        if not isinstance(value["audit_key_ref"], str) or not SECRET_REF.fullmatch(value["audit_key_ref"]):
+            raise ContractDenied("AUDIT_KEY_REFERENCE_INVALID")
+        if value["grant_signer_identity"] != "aios-task-grant-signer" or value["grant_signature_namespace"] != "aios-task-status-merge-grant-v1":
+            raise ContractDenied("GRANT_SIGNATURE_POLICY_INVALID")
+        issued, expires = _time(value["issued_at"]), _time(value["expires_at"])
+        current = now or datetime.now(timezone.utc)
+        lifetime = (expires - issued).total_seconds()
+        if lifetime <= 0 or lifetime > 12 * 3600 or (not allow_expired and current >= expires):
+            raise ContractDenied("GRANT_EXPIRED_OR_LIFETIME_INVALID")
+        return cls(dict(value), issued, expires)
+
+    @property
+    def digest(self) -> str:
+        return hashlib.sha256(canonical(self.raw)).hexdigest()
+
+
+def load_signed_task_grant(grant_path: Path, signature_path: Path, allowed_signers: Path,
+                           *, now: datetime | None = None, allow_expired: bool = False) -> TaskGrant:
+    for path in (grant_path, signature_path, allowed_signers):
+        if not path.is_absolute() or path.is_symlink():
+            raise ContractDenied("GRANT_SIGNATURE_PATH_UNSAFE")
+    raw = grant_path.read_bytes()
+    if len(raw) > 1024 * 1024:
+        raise ContractDenied("GRANT_TOO_LARGE")
+    try:
+        value = json.loads(raw)
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ContractDenied("GRANT_JSON_INVALID") from exc
+    if raw != canonical(value) + b"\n":
+        raise ContractDenied("GRANT_NOT_CANONICAL")
+    grant = TaskGrant.from_dict(value, now=now, allow_expired=allow_expired)
+    command = [
+        "/usr/bin/ssh-keygen", "-Y", "verify", "-f", str(allowed_signers),
+        "-I", "aios-task-grant-signer", "-n", "aios-task-status-merge-grant-v1",
+        "-s", str(signature_path),
+    ]
+    try:
+        subprocess.run(command, input=raw, check=True, stdout=subprocess.DEVNULL, stderr=subprocess.PIPE)
+    except (OSError, subprocess.CalledProcessError) as exc:
+        raise ContractDenied("GRANT_SIGNATURE_INVALID") from exc
+    return grant
+
+
+class API(Protocol):
+    def get(self, path: str) -> Any: ...
+    def post(self, path: str, payload: dict[str, Any]) -> Any: ...
+
+
+def read_runtime_secret(path: Path) -> bytes:
+    flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
+    descriptor = os.open(path, flags)
+    try:
+        metadata = os.fstat(descriptor)
+        if not stat.S_ISREG(metadata.st_mode) or stat.S_IMODE(metadata.st_mode) & 0o077:
+            raise ContractDenied("CREDENTIAL_FILE_UNSAFE")
+        value = os.read(descriptor, 16 * 1024 + 1)
+    finally:
+        os.close(descriptor)
+    if not value or len(value) > 16 * 1024 or b"\x00" in value or b"\n" in value.strip():
+        raise ContractDenied("CREDENTIAL_CONTENT_INVALID")
+    return value.strip()
+
+
+class _NoRedirect(urllib.request.HTTPRedirectHandler):
+    def redirect_request(self, req, fp, code, msg, headers, newurl):
+        return None
+
+
+class GiteaClient:
+    def __init__(self, token_path: Path, api_base: str = "http://127.0.0.1:3000/api/v1"):
+        parsed = urllib.parse.urlparse(api_base)
+        if parsed.scheme != "http" or parsed.hostname != "127.0.0.1" or parsed.path.rstrip("/") != "/api/v1" or parsed.username or parsed.password or parsed.query or parsed.fragment:
+            raise ContractDenied("GITEA_API_ORIGIN_INVALID")
+        self.token_path = Path(token_path)
+        self.api_base = api_base.rstrip("/")
+        self.opener = urllib.request.build_opener(_NoRedirect)
+
+    def _request(self, method: str, path: str, payload: dict[str, Any] | None = None) -> Any:
+        if not path.startswith("/") or ".." in path.split("/"):
+            raise ContractDenied("GITEA_API_PATH_INVALID")
+        token = read_runtime_secret(self.token_path).decode("utf-8")
+        body = canonical(payload) if payload is not None else None
+        request = urllib.request.Request(self.api_base + path, data=body, method=method, headers={"Authorization": f"token {token}", "Content-Type": "application/json", "Accept": "application/json"})
+        try:
+            with self.opener.open(request, timeout=20) as response:
+                raw = response.read(2 * 1024 * 1024 + 1)
+        except (urllib.error.URLError, OSError) as exc:
+            raise ContractDenied(f"GITEA_API_FAILED:{type(exc).__name__}") from exc
+        if len(raw) > 2 * 1024 * 1024:
+            raise ContractDenied("GITEA_API_RESPONSE_TOO_LARGE")
+        return json.loads(raw.decode("utf-8")) if raw else None
+
+    def get(self, path: str) -> Any:
+        return self._request("GET", path)
+
+    def post(self, path: str, payload: dict[str, Any]) -> Any:
+        return self._request("POST", path, payload)
+
+
+class QuotaJournal:
+    def __init__(self, path: Path, grant: TaskGrant):
+        self.path = Path(path)
+        self.grant = grant
+        self.connection = sqlite3.connect(self.path, isolation_level=None)
+        self.connection.execute("PRAGMA journal_mode=WAL")
+        self.connection.execute("PRAGMA synchronous=FULL")
+        self.connection.execute("CREATE TABLE IF NOT EXISTS quota(operation TEXT PRIMARY KEY, state TEXT NOT NULL, effect_sha256 TEXT)")
+        self.connection.execute("CREATE TABLE IF NOT EXISTS audit(seq INTEGER PRIMARY KEY AUTOINCREMENT, event BLOB NOT NULL, previous_hash TEXT NOT NULL, event_hash TEXT NOT NULL UNIQUE)")
+
+    def close(self) -> None:
+        self.connection.close()
+
+    def reserve(self, operation: str) -> str:
+        limit = {"statuses": 1, "merge": 1}.get(operation)
+        if limit is None:
+            raise ContractDenied("OPERATION_INVALID")
+        self.connection.execute("BEGIN IMMEDIATE")
+        try:
+            row = self.connection.execute("SELECT state FROM quota WHERE operation=?", (operation,)).fetchone()
+            if row and row[0] == "COMMITTED":
+                raise ContractDenied("STATUS_QUOTA_EXHAUSTED" if operation == "statuses" else "MERGE_QUOTA_EXHAUSTED")
+            if row and row[0] != "RESERVED":
+                raise ContractDenied("JOURNAL_STATE_INVALID")
+            if not row:
+                self.connection.execute("INSERT INTO quota(operation,state,effect_sha256) VALUES(?,'RESERVED',NULL)", (operation,))
+            self.connection.execute("COMMIT")
+            return "RESERVED"
+        except Exception:
+            self.connection.execute("ROLLBACK")
+            raise
+
+    def state(self, operation: str) -> str | None:
+        row = self.connection.execute("SELECT state FROM quota WHERE operation=?", (operation,)).fetchone()
+        return str(row[0]) if row else None
+
+    def commit(self, operation: str, effect: dict[str, Any]) -> None:
+        digest = hashlib.sha256(canonical(effect)).hexdigest()
+        self.connection.execute("BEGIN IMMEDIATE")
+        try:
+            row = self.connection.execute("SELECT state,effect_sha256 FROM quota WHERE operation=?", (operation,)).fetchone()
+            if row is None or row[0] not in {"RESERVED", "COMMITTED"}:
+                raise ContractDenied("RESERVATION_MISSING")
+            if row[0] == "COMMITTED" and row[1] != digest:
+                raise ContractDenied("COMMITTED_EFFECT_MISMATCH")
+            self.connection.execute("UPDATE quota SET state='COMMITTED',effect_sha256=? WHERE operation=?", (digest, operation))
+            self.connection.execute("COMMIT")
+        except Exception:
+            self.connection.execute("ROLLBACK")
+            raise
+
+    def append(self, event: dict[str, Any]) -> str:
+        row = self.connection.execute("SELECT event_hash FROM audit ORDER BY seq DESC LIMIT 1").fetchone()
+        previous = row[0] if row else "0" * 64
+        raw = canonical(event)
+        digest = hashlib.sha256(previous.encode() + raw).hexdigest()
+        self.connection.execute("INSERT INTO audit(event,previous_hash,event_hash) VALUES(?,?,?)", (raw, previous, digest))
+        return digest
+
+    def verify_chain(self) -> bool:
+        previous = "0" * 64
+        for raw, observed_previous, observed_hash in self.connection.execute("SELECT event,previous_hash,event_hash FROM audit ORDER BY seq"):
+            if observed_previous != previous or hashlib.sha256(previous.encode() + raw).hexdigest() != observed_hash:
+                return False
+            previous = observed_hash
+        return True
+
+
+def _latest_statuses(values: Any) -> dict[str, dict[str, Any]]:
+    if not isinstance(values, list):
+        raise ContractDenied("STATUS_RESPONSE_INVALID")
+    latest: dict[str, tuple[tuple[str, int], dict[str, Any]]] = {}
+    for item in values:
+        if not isinstance(item, dict) or item.get("context") not in ALLOWED_CONTEXTS:
+            continue
+        key = (str(item.get("created_at") or ""), int(item.get("id") or 0))
+        context = item["context"]
+        if context not in latest or key > latest[context][0]:
+            latest[context] = (key, item)
+    return {context: item for context, (_, item) in latest.items()}
+
+
+def _verify_owner_binding(api: API, grant: TaskGrant) -> tuple[dict[str, Any], dict[str, Any]]:
+    pr_number = grant.raw["pr_number"]
+    head, base = grant.raw["target_commit"], grant.raw["base_oid"]
+    pr = api.get(f"/repos/aios/aios/pulls/{pr_number}")
+    reviews = api.get(f"/repos/aios/aios/pulls/{pr_number}/reviews")
+    if not isinstance(pr, dict) or pr.get("state") != "open" or pr.get("mergeable") is not True or (pr.get("head") or {}).get("sha") != head or (pr.get("base") or {}).get("sha") != base or (pr.get("base") or {}).get("ref") != "governance/approvals":
+        raise ContractDenied("PR_BINDING_INVALID")
+    if not isinstance(reviews, list):
+        raise ContractDenied("REVIEW_RESPONSE_INVALID")
+    owners = [item for item in reviews if isinstance(item, dict) and (item.get("user") or {}).get("login") == "owner" and (item.get("user") or {}).get("id") == 2]
+    if not owners:
+        raise ContractDenied("OWNER_EXACT_HEAD_APPROVAL_MISSING")
+    latest = max(owners, key=lambda item: int(item.get("id") or -1))
+    if latest.get("state") != "APPROVED" or latest.get("commit_id") != head or latest.get("official") is not True or latest.get("stale") is not False or latest.get("dismissed") is not False:
+        raise ContractDenied("OWNER_EXACT_HEAD_APPROVAL_MISSING")
+    submitted = _time(str(latest.get("submitted_at") or ""))
+    if submitted < grant.issued_at or submitted >= grant.expires_at:
+        raise ContractDenied("OWNER_APPROVAL_EXPIRED")
+    return pr, latest
+
+
+def _signed_receipt(value: dict[str, Any], key: bytes) -> dict[str, Any]:
+    if not key:
+        raise ContractDenied("AUDIT_KEY_EMPTY")
+    result = dict(value)
+    result["audit_signature"] = hmac.new(key, canonical(value), hashlib.sha256).hexdigest()
+    return result
+
+
+class StatusBroker:
+    def __init__(self, grant: TaskGrant, journal: QuotaJournal, api: API, audit_key: bytes):
+        self.grant, self.journal, self.api, self.audit_key = grant, journal, api, audit_key
+
+    def publish(self, evidence: list[dict[str, Any]]) -> dict[str, Any]:
+        if not isinstance(evidence, list) or len(evidence) != 5:
+            raise ContractDenied("EVIDENCE_SET_INVALID")
+        mapped = {item.get("context"): item for item in evidence if isinstance(item, dict)}
+        if set(mapped) != ALLOWED_CONTEXTS or any(set(item) != {"context", "evidence_sha256"} or not SHA256.fullmatch(str(item["evidence_sha256"])) or item["evidence_sha256"] != self.grant.raw["status_evidence_sha256"].get(context) for context, item in mapped.items()):
+            raise ContractDenied("EVIDENCE_SET_INVALID")
+        _verify_owner_binding(self.api, self.grant)
+        self.journal.reserve("statuses")
+        published = []
+        head = self.grant.raw["target_commit"]
+        pr_number = self.grant.raw["pr_number"]
+        observed_before = self.api.get(f"/repos/aios/aios/commits/{head}/statuses")
+        already = _latest_statuses(observed_before)
+        already = {context: item for context, item in already.items()
+                   if item.get("status", item.get("state")) == "success"
+                   and item.get("sha", head) == head
+                   and (item.get("creator") or {}).get("login") == self.grant.raw["status_identity"]
+                   and str(item.get("description") or "").endswith(mapped[context]["evidence_sha256"])}
+        for context in sorted(ALLOWED_CONTEXTS):
+            digest = mapped[context]["evidence_sha256"]
+            if context in already:
+                published.append({"context": context, "evidence_sha256": digest, "status_id": already[context].get("id"), "reconciled": True})
+                continue
+            response = self.api.post(f"/repos/aios/aios/statuses/{head}", {
+                "state": "success", "context": context,
+                "description": f"task-scoped PASS evidence_sha256={digest}",
+                "target_url": f"https://g.cnnetmkt.top/git/aios/aios/pulls/{pr_number}",
+            })
+            if not isinstance(response, dict) or response.get("context") != context or response.get("status", response.get("state")) != "success":
+                raise ContractDenied("STATUS_PUBLICATION_FAILED")
+            published.append({"context": context, "evidence_sha256": digest, "status_id": response.get("id")})
+        observed = self.api.get(f"/repos/aios/aios/commits/{head}/statuses")
+        latest = _latest_statuses(observed)
+        if set(latest) != ALLOWED_CONTEXTS or any(
+            item.get("status", item.get("state")) != "success"
+            or item.get("sha", head) != head
+            or (item.get("creator") or {}).get("login") != self.grant.raw["status_identity"]
+            or not str(item.get("description") or "").endswith(mapped[context]["evidence_sha256"])
+            for context, item in latest.items()
+        ):
+            raise ContractDenied("STATUS_READBACK_INVALID")
+        receipt = {"status": "PASS", "operation": "publish_statuses", "grant_sha256": self.grant.digest, "target_commit": head, "published_count": 5, "published": published}
+        receipt["audit_signature"] = hmac.new(self.audit_key, canonical(receipt), hashlib.sha256).hexdigest()
+        receipt["audit_chain_hash"] = self.journal.append(receipt)
+        self.journal.commit("statuses", receipt)
+        return receipt
+
+
+class MergeBroker:
+    def __init__(self, grant: TaskGrant, journal: QuotaJournal, api: API, audit_key: bytes):
+        self.grant, self.journal, self.api, self.audit_key = grant, journal, api, audit_key
+
+    def merge(self) -> dict[str, Any]:
+        head, base = self.grant.raw["target_commit"], self.grant.raw["base_oid"]
+        pr_number = self.grant.raw["pr_number"]
+        pr = self.api.get(f"/repos/aios/aios/pulls/{pr_number}")
+        if isinstance(pr, dict) and pr.get("merged") is True:
+            observed_branch = self.api.get("/repos/aios/aios/branches/governance%2Fapprovals")
+            if pr.get("merge_commit_sha") != head or (observed_branch.get("commit") or {}).get("id") != head or self.journal.state("merge") != "RESERVED":
+                raise ContractDenied("FAST_FORWARD_RESULT_MISMATCH")
+            receipt = {"status": "PASS", "operation": "fast_forward_merge", "grant_sha256": self.grant.digest, "pr_number": pr_number, "governance_oid": head, "merge_method": "fast-forward-only", "reconciled": True}
+            receipt["audit_signature"] = hmac.new(self.audit_key, canonical(receipt), hashlib.sha256).hexdigest()
+            receipt["audit_chain_hash"] = self.journal.append(receipt)
+            self.journal.commit("merge", receipt)
+            return receipt
+        pr, _ = _verify_owner_binding(self.api, self.grant)
+        statuses = self.api.get(f"/repos/aios/aios/commits/{head}/statuses")
+        latest = _latest_statuses(statuses)
+        if set(latest) != ALLOWED_CONTEXTS or any(item.get("status", item.get("state")) != "success" or item.get("sha", head) != head or (item.get("creator") or {}).get("login") != self.grant.raw["status_identity"] or not str(item.get("description") or "").endswith(self.grant.raw["status_evidence_sha256"][context]) for context, item in latest.items()):
+            raise ContractDenied("MACHINE_STATUS_INVALID")
+        self.journal.reserve("merge")
+        self.api.post(f"/repos/aios/aios/pulls/{pr_number}/merge", {"Do": "fast-forward-only", "head_commit_id": head, "force_merge": False, "delete_branch_after_merge": False})
+        observed_pr = self.api.get(f"/repos/aios/aios/pulls/{pr_number}")
+        observed_branch = self.api.get("/repos/aios/aios/branches/governance%2Fapprovals")
+        if observed_pr.get("merged") is not True or observed_pr.get("merge_commit_sha") != head or (observed_branch.get("commit") or {}).get("id") != head:
+            raise ContractDenied("FAST_FORWARD_RESULT_MISMATCH")
+        receipt = {"status": "PASS", "operation": "fast_forward_merge", "grant_sha256": self.grant.digest, "pr_number": pr_number, "governance_oid": head, "merge_method": "fast-forward-only"}
+        receipt["audit_signature"] = hmac.new(self.audit_key, canonical(receipt), hashlib.sha256).hexdigest()
+        receipt["audit_chain_hash"] = self.journal.append(receipt)
+        self.journal.commit("merge", receipt)
+        return receipt
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(prog="aios-task-status-merge-broker")
+    parser.add_argument("--mode", required=True, choices=("status", "merge"))
+    parser.add_argument("--grant", required=True, type=Path)
+    parser.add_argument("--token", required=True, type=Path)
+    parser.add_argument("--audit-key", required=True, type=Path)
+    parser.add_argument("--grant-signature", required=True, type=Path)
+    parser.add_argument("--allowed-signers", required=True, type=Path)
+    parser.add_argument("--journal", required=True, type=Path)
+    parser.add_argument("--evidence", type=Path)
+    args = parser.parse_args(argv)
+    grant = load_signed_task_grant(args.grant, args.grant_signature, args.allowed_signers)
+    journal = QuotaJournal(args.journal, grant)
+    try:
+        client = GiteaClient(args.token)
+        key = read_runtime_secret(args.audit_key)
+        if args.mode == "status":
+            if args.evidence is None:
+                raise ContractDenied("EVIDENCE_REQUIRED")
+            result = StatusBroker(grant, journal, client, key).publish(json.loads(args.evidence.read_text(encoding="utf-8")))
+        else:
+            if args.evidence is not None:
+                raise ContractDenied("EVIDENCE_FORBIDDEN")
+            result = MergeBroker(grant, journal, client, key).merge()
+        print(json.dumps(result, sort_keys=True, separators=(",", ":")))
+        return 0
+    finally:
+        journal.close()
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

services/aios-single-owner-governance/task-status-merge-grant-v1.schema.json

@@ -0,0 +1,64 @@
+{
+  "$schema": "https://json-schema.org/draft/2020-12/schema",
+  "$id": "AIOS_TASK_STATUS_MERGE_GRANT_V1",
+  "type": "object",
+  "additionalProperties": false,
+  "required": [
+    "schema_id", "schema_version", "repository", "pr_number", "base_oid",
+    "target_commit", "approved_tree_oid", "manifest_path", "manifest_blob_oid", "manifest_sha256",
+    "payload_sha256", "status_identity", "merge_identity", "status_contexts",
+    "status_write_quota", "merge_quota", "merge_method", "branch_push_allowed",
+    "tag_write_allowed", "main_write_allowed", "registry_write_allowed",
+    "baseline_write_allowed", "production_write_allowed", "issued_at", "expires_at",
+    "audit_key_ref", "grant_signer_identity", "grant_signature_namespace",
+    "status_evidence_sha256", "supersedes_pr_number", "supersedes_head_oid",
+    "renewal_required"
+  ],
+  "properties": {
+    "schema_id": {"const": "AIOS_TASK_STATUS_MERGE_GRANT"},
+    "schema_version": {"const": 1},
+    "repository": {"const": "aios/aios"},
+    "pr_number": {"type": "integer", "minimum": 1},
+    "base_oid": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "target_commit": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "approved_tree_oid": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "manifest_path": {"type": "string", "pattern": "^approvals/[A-Za-z0-9._/-]+\\.json$"},
+    "manifest_blob_oid": {"type": "string", "pattern": "^[0-9a-f]{40}$"},
+    "manifest_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+    "payload_sha256": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+    "status_evidence_sha256": {
+      "type": "object", "additionalProperties": false,
+      "required": ["aios/branch-protection", "aios/evidence-closure", "aios/secret-scan", "aios/state-machine", "aios/validator"],
+      "properties": {
+        "aios/branch-protection": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+        "aios/evidence-closure": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+        "aios/secret-scan": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+        "aios/state-machine": {"type": "string", "pattern": "^[0-9a-f]{64}$"},
+        "aios/validator": {"type": "string", "pattern": "^[0-9a-f]{64}$"}
+      }
+    },
+    "supersedes_pr_number": {"const": 36},
+    "supersedes_head_oid": {"const": "a8644fc172bd8ebbba17a16e6bf586c93fd1e793"},
+    "renewal_required": {"const": true},
+    "status_identity": {"enum": ["codex", "gpt"]},
+    "merge_identity": {"enum": ["codex", "gpt"]},
+    "status_contexts": {
+      "type": "array", "minItems": 5, "maxItems": 5, "uniqueItems": true,
+      "items": {"enum": ["aios/branch-protection", "aios/evidence-closure", "aios/secret-scan", "aios/state-machine", "aios/validator"]}
+    },
+    "status_write_quota": {"const": 5},
+    "merge_quota": {"const": 1},
+    "merge_method": {"const": "fast-forward-only"},
+    "branch_push_allowed": {"const": false},
+    "tag_write_allowed": {"const": false},
+    "main_write_allowed": {"const": false},
+    "registry_write_allowed": {"const": false},
+    "baseline_write_allowed": {"const": false},
+    "production_write_allowed": {"const": false},
+    "issued_at": {"type": "string", "format": "date-time"},
+    "expires_at": {"type": "string", "format": "date-time"},
+    "audit_key_ref": {"type": "string", "pattern": "^systemd-credential://[A-Za-z0-9_.-]+$"},
+    "grant_signer_identity": {"const": "aios-task-grant-signer"},
+    "grant_signature_namespace": {"const": "aios-task-status-merge-grant-v1"}
+  }
+}

tests/single_owner_governance/test_task_status_merge_broker.py

@@ -0,0 +1,357 @@
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+import json
+from pathlib import Path
+import sys
+import subprocess
+import tempfile
+import unittest
+from unittest import mock
+import urllib.error
+
+
+ROOT = Path(__file__).resolve().parents[2]
+sys.path.insert(0, str(ROOT / "services" / "aios-single-owner-governance"))
+
+from aios_single_owner_governance.task_status_merge_broker import (  # noqa: E402
+    ALLOWED_CONTEXTS,
+    ContractDenied,
+    MergeBroker,
+    QuotaJournal,
+    StatusBroker,
+    TaskGrant,
+    canonical,
+    load_signed_task_grant,
+)
+from aios_single_owner_governance.task_credential_broker import TaskCredentialBroker  # noqa: E402
+from aios_single_owner_governance import task_credential_expiry_sweeper as sweeper  # noqa: E402
+
+
+HEAD = "d" * 40
+OLD_HEAD = "a8644fc172bd8ebbba17a16e6bf586c93fd1e793"
+BASE = "8814dff20222625b60454c613b61ccf15fa20e51"
+
+
+def grant() -> dict:
+    now = datetime.now(timezone.utc).replace(microsecond=0)
+    value = {
+        "schema_id": "AIOS_TASK_STATUS_MERGE_GRANT",
+        "schema_version": 1,
+        "repository": "aios/aios",
+        "pr_number": 36,
+        "base_oid": BASE,
+        "target_commit": HEAD,
+        "approved_tree_oid": "3649be2dac8fcea3a79cb1604b2cfb63f6cc0307",
+        "manifest_path": "approvals/release-campaign/AIOS-V2R1C4-PHASE-B-RENEWAL.json",
+        "manifest_blob_oid": "d6e4431992e4315799c7f0487f2b12a20d0e2890",
+        "manifest_sha256": "2555c5e37f77f1d0aac51864733b55ee3e91169ae76bf67e8817de6d1874eea4",
+        "payload_sha256": "4db7d3fd5aaa00a65eabfce7501f275b5f6c8a2587b6153fa8066ec0dafa5d6d",
+        "status_evidence_sha256": {context: "1" * 64 for context in ALLOWED_CONTEXTS},
+        "supersedes_pr_number": 36,
+        "supersedes_head_oid": OLD_HEAD,
+        "renewal_required": True,
+        "status_identity": "codex",
+        "merge_identity": "codex",
+        "status_contexts": sorted(ALLOWED_CONTEXTS),
+        "status_write_quota": 5,
+        "merge_quota": 1,
+        "merge_method": "fast-forward-only",
+        "branch_push_allowed": False,
+        "tag_write_allowed": False,
+        "main_write_allowed": False,
+        "registry_write_allowed": False,
+        "baseline_write_allowed": False,
+        "production_write_allowed": False,
+        "issued_at": now.isoformat().replace("+00:00", "Z"),
+        "expires_at": (now + timedelta(hours=1)).isoformat().replace("+00:00", "Z"),
+        "audit_key_ref": "systemd-credential://task-audit-hmac",
+        "grant_signer_identity": "aios-task-grant-signer",
+        "grant_signature_namespace": "aios-task-status-merge-grant-v1",
+    }
+    value["pr_number"] = 37
+    return value
+
+
+class FakeAPI:
+    def __init__(self, *, with_statuses: bool = False, fail_status_after: int | None = None,
+                 fail_after_merge: bool = False):
+        self.posts: list[tuple[str, dict]] = []
+        self.merged = False
+        self.status_rows = [{"id": index + 1, "context": value, "status": "success", "sha": HEAD,
+                             "description": "task-scoped PASS evidence_sha256=" + "1" * 64,
+                             "creator": {"login": "codex"}} for index, value in enumerate(ALLOWED_CONTEXTS)] if with_statuses else []
+        self.fail_status_after = fail_status_after
+        self.fail_after_merge = fail_after_merge
+        self.base = BASE
+
+    def get(self, path: str):
+        if path.endswith("/reviews"):
+            return [{
+                "id": 76,
+                "state": "APPROVED",
+                "commit_id": HEAD,
+                "official": True,
+                "stale": False,
+                "dismissed": False,
+                "submitted_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
+                "user": {"id": 2, "login": "owner"},
+            }]
+        if path.endswith("/statuses"):
+            return list(self.status_rows)
+        if path.endswith("/branches/governance%2Fapprovals"):
+            return {"name": "governance/approvals", "commit": {"id": HEAD if self.merged else BASE}}
+        if path.endswith("/pulls/37"):
+            return {
+                "number": 37,
+                "state": "closed" if self.merged else "open",
+                "merged": self.merged,
+                "mergeable": True,
+                "head": {"sha": HEAD},
+                "base": {"sha": self.base, "ref": "governance/approvals"},
+                "merge_commit_sha": HEAD if self.merged else None,
+            }
+        raise AssertionError(path)
+
+    def post(self, path: str, payload: dict):
+        self.posts.append((path, payload))
+        if path.endswith("/merge"):
+            assert payload == {
+                "Do": "fast-forward-only",
+                "head_commit_id": HEAD,
+                "force_merge": False,
+                "delete_branch_after_merge": False,
+            }
+            self.merged = True
+            if self.fail_after_merge:
+                raise ContractDenied("SIMULATED_POST_MERGE_LOSS")
+            return {"merged": True}
+        if self.fail_status_after is not None and len(self.status_rows) >= self.fail_status_after:
+            raise ContractDenied("SIMULATED_STATUS_FAILURE")
+        row = {"id": len(self.posts), "context": payload["context"], "status": "success", "sha": HEAD,
+               "description": payload["description"],
+               "creator": {"login": "codex"}}
+        self.status_rows.append(row)
+        return row
+
+
+def evidence() -> list[dict]:
+    return [{"context": value, "evidence_sha256": "1" * 64} for value in sorted(ALLOWED_CONTEXTS)]
+
+
+class TaskStatusMergeBrokerTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.root = Path(self.temporary.name)
+        self.journals: list[QuotaJournal] = []
+
+    def tearDown(self):
+        for journal in self.journals:
+            journal.close()
+        self.temporary.cleanup()
+
+    def journal(self, item: TaskGrant) -> QuotaJournal:
+        value = QuotaJournal(self.root / "journal.sqlite3", item)
+        self.journals.append(value)
+        return value
+
+    def test_status_broker_publishes_exact_five_and_quota_is_single_use(self):
+        item = TaskGrant.from_dict(grant())
+        api = FakeAPI()
+        journal = self.journal(item)
+        receipt = StatusBroker(item, journal, api, b"audit-key").publish(evidence())
+        self.assertEqual(receipt["status"], "PASS")
+        self.assertEqual(receipt["published_count"], 5)
+        self.assertEqual({entry[1]["context"] for entry in api.posts}, ALLOWED_CONTEXTS)
+        with self.assertRaisesRegex(ContractDenied, "STATUS_QUOTA_EXHAUSTED"):
+            StatusBroker(item, journal, api, b"audit-key").publish(evidence())
+
+    def test_merge_broker_uses_fast_forward_only_and_binds_exact_head(self):
+        item = TaskGrant.from_dict(grant())
+        api = FakeAPI(with_statuses=True)
+        journal = self.journal(item)
+        receipt = MergeBroker(item, journal, api, b"audit-key").merge()
+        self.assertEqual(receipt["status"], "PASS")
+        self.assertEqual(receipt["governance_oid"], HEAD)
+        self.assertTrue(api.merged)
+        with self.assertRaisesRegex(ContractDenied, "MERGE_QUOTA_EXHAUSTED"):
+            MergeBroker(item, journal, FakeAPI(with_statuses=True), b"audit-key").merge()
+
+    def test_contract_rejects_scope_expansion_secret_and_wrong_identity(self):
+        value = grant()
+        value["main_write_allowed"] = True
+        with self.assertRaisesRegex(ContractDenied, "WRITE_SCOPE_EXPANDED"):
+            TaskGrant.from_dict(value)
+
+    def test_unbound_evidence_fails_closed(self):
+        item = TaskGrant.from_dict(grant())
+        changed = evidence()
+        changed[0]["evidence_sha256"] = "2" * 64
+        with self.assertRaisesRegex(ContractDenied, "EVIDENCE_SET_INVALID"):
+            StatusBroker(item, self.journal(item), FakeAPI(), b"audit-key").publish(changed)
+
+    def test_grant_requires_detached_asymmetric_signature(self):
+        grant_path = self.root / "grant.json"
+        signature_path = self.root / "grant.sig"
+        allowed_signers = self.root / "allowed_signers"
+        raw = canonical(grant()) + b"\n"
+        grant_path.write_bytes(raw)
+        signature_path.write_text("ssh-signature", encoding="utf-8")
+        allowed_signers.write_text("aios-task-grant-signer ssh-ed25519 AAAA", encoding="utf-8")
+        with mock.patch("aios_single_owner_governance.task_status_merge_broker.subprocess.run") as runner:
+            item = load_signed_task_grant(grant_path, signature_path, allowed_signers)
+        self.assertEqual(item.raw["pr_number"], 37)
+        self.assertEqual(runner.call_args.kwargs["input"], raw)
+        with mock.patch(
+            "aios_single_owner_governance.task_status_merge_broker.subprocess.run",
+            side_effect=subprocess.CalledProcessError(255, ["ssh-keygen"]),
+        ):
+            with self.assertRaisesRegex(ContractDenied, "GRANT_SIGNATURE_INVALID"):
+                load_signed_task_grant(grant_path, signature_path, allowed_signers)
+
+    def test_base_movement_blocks_status_publication_before_side_effect(self):
+        item = TaskGrant.from_dict(grant())
+        api = FakeAPI()
+        api.base = "c" * 40
+        with self.assertRaisesRegex(ContractDenied, "PR_BINDING_INVALID"):
+            StatusBroker(item, self.journal(item), api, b"audit-key").publish(evidence())
+        self.assertEqual(api.posts, [])
+
+    def test_credential_provision_rechecks_local_owner_and_refs_before_issuance(self):
+        item = TaskGrant.from_dict(grant())
+        broker = TaskCredentialBroker(item, self.root / "audit", b"audit-key")
+        with mock.patch("aios_single_owner_governance.task_credential_broker.os.geteuid", return_value=0, create=True), \
+             mock.patch("aios_single_owner_governance.task_credential_broker._safe_absolute"), \
+             mock.patch("aios_single_owner_governance.task_credential_broker._verify_local_governance_binding", side_effect=ContractDenied("LOCAL_OWNER_APPROVAL_MISSING")), \
+             mock.patch("aios_single_owner_governance.task_credential_broker.subprocess.run") as runner:
+            with self.assertRaisesRegex(ContractDenied, "LOCAL_OWNER_APPROVAL_MISSING"):
+                broker.provision(
+                    identity="codex",
+                    token_name="pr37-dddddddddddd-status",
+                    encrypted_path=Path("/run/aios-task-credential-broker/status.cred"),
+                )
+        runner.assert_not_called()
+        value = grant()
+        value["status_identity"] = "gitops-bot"
+        with self.assertRaisesRegex(ContractDenied, "IDENTITY_NOT_ALLOWED"):
+            TaskGrant.from_dict(value)
+
+    def test_expiry_sweeper_revokes_projected_token_after_grant_expiry(self):
+        item = TaskGrant.from_dict(grant())
+        runtime_root = self.root / "runtime"
+        credential_root = self.root / "credentials"
+        task_dir = runtime_root / "task-37"
+        audit_dir = task_dir / "audit"
+        audit_dir.mkdir(parents=True)
+        credential_root.mkdir()
+        (task_dir / "grant.json").write_text("{}", encoding="utf-8")
+        (task_dir / "grant.sig").write_text("signature", encoding="utf-8")
+        token_name = "pr37-dddddddddddd-status"
+        encrypted_path = credential_root / "status.cred"
+        encrypted_path.write_text("encrypted", encoding="utf-8")
+        (credential_root / "task-37-audit.cred").write_text("encrypted-audit-key", encoding="utf-8")
+        real_broker = TaskCredentialBroker(item, audit_dir, b"audit-key")
+        receipt = real_broker._audit({
+            "status": "PASS", "operation": "credential_provision", "identity": "codex",
+            "token_name": token_name, "token_id": 101, "token_uid": 7,
+            "grant_sha256": item.digest,
+            "credential_ref": "systemd-credential-encrypted://status.cred",
+        })
+        real_broker.validate_provision_receipt(receipt, encrypted_path)
+        tampered = dict(receipt)
+        tampered["token_id"] = 102
+        with self.assertRaisesRegex(ContractDenied, "PROVISION_RECEIPT_INVALID"):
+            real_broker.validate_provision_receipt(tampered, encrypted_path)
+        fake_broker = mock.Mock()
+        fake_broker.revoke.return_value = {"status": "PASS", "old_token_http_status": 401}
+        with mock.patch.object(sweeper, "RUNTIME_ROOT", runtime_root), \
+             mock.patch.object(sweeper, "CREDENTIAL_ROOT", credential_root), \
+             mock.patch.object(sweeper.os, "geteuid", return_value=0, create=True), \
+             mock.patch.object(sweeper, "load_signed_task_grant", return_value=item), \
+             mock.patch.object(sweeper.subprocess, "run", return_value=mock.Mock(stdout=b"audit-key")), \
+             mock.patch.object(sweeper, "TaskCredentialBroker", return_value=fake_broker) as broker_type:
+            results = sweeper.sweep(runtime_root, self.root / "allowed_signers", now=item.expires_at)
+        self.assertEqual(results, [{"status": "PASS", "old_token_http_status": 401}])
+        broker_type.assert_called_once_with(item, audit_dir, b"audit-key")
+        fake_broker.revoke.assert_called_once()
+
+    def test_revocation_is_exact_access_token_row_only_and_proves_http_401(self):
+        item = TaskGrant.from_dict(grant())
+        broker = TaskCredentialBroker(item, self.root / "revoke-audit", b"audit-key")
+        token_name = "pr37-dddddddddddd-status"
+        receipt = broker._audit({
+            "status": "PASS", "operation": "credential_provision", "identity": "codex",
+            "token_name": token_name, "token_id": 101, "token_uid": 7,
+            "grant_sha256": item.digest,
+            "credential_ref": "systemd-credential-encrypted://status.cred",
+        })
+        response = mock.MagicMock()
+        response.__enter__.return_value.status = 200
+        unauthorized = urllib.error.HTTPError("http://127.0.0.1:3000/api/v1/repos/aios/aios", 401, "Unauthorized", {}, None)
+        write_connection = mock.MagicMock()
+        write_connection.execute.return_value.rowcount = 1
+        read_connection = mock.MagicMock()
+        read_connection.execute.return_value.fetchone.return_value = (0,)
+        with mock.patch("aios_single_owner_governance.task_credential_broker.os.geteuid", return_value=0, create=True), \
+             mock.patch("aios_single_owner_governance.task_credential_broker._safe_absolute"), \
+             mock.patch("aios_single_owner_governance.task_credential_broker.subprocess.run", return_value=mock.Mock(stdout=b"x" * 40)), \
+             mock.patch("aios_single_owner_governance.task_credential_broker.sqlite3.connect", side_effect=[write_connection, read_connection]) as connect, \
+             mock.patch("aios_single_owner_governance.task_credential_broker.urllib.request.urlopen", side_effect=[response, unauthorized]) as urlopen:
+            result = broker.revoke(receipt=receipt, encrypted_path=Path("/run/aios-task-credential-broker/status.cred"))
+        self.assertEqual(result["old_token_http_status"], 401)
+        self.assertIn("/repos/aios/aios", urlopen.call_args_list[0].args[0].full_url)
+        delete_sql = write_connection.execute.call_args_list[1].args[0]
+        self.assertIn("DELETE FROM access_token", delete_sql)
+        self.assertNotIn("commit_status", delete_sql)
+        self.assertIn("mode=ro", connect.call_args_list[1].args[0])
+        value = grant()
+        value["token"] = "plaintext"
+        with self.assertRaisesRegex(ContractDenied, "GRANT_FIELDS_INVALID"):
+            TaskGrant.from_dict(value)
+
+    def test_audit_chain_is_content_addressed_and_contains_no_secret(self):
+        item = TaskGrant.from_dict(grant())
+        journal = self.journal(item)
+        receipt = StatusBroker(item, journal, FakeAPI(), b"audit-key").publish(evidence())
+        serialized = json.dumps(receipt, sort_keys=True)
+        self.assertNotIn("audit-key", serialized)
+        self.assertEqual(len(receipt["audit_signature"]), 64)
+        self.assertTrue(journal.verify_chain())
+
+    def test_credential_broker_is_root_only_and_never_returns_a_token(self):
+        item = TaskGrant.from_dict(grant())
+        broker = TaskCredentialBroker(item, self.root / "audit", b"audit-key")
+        with mock.patch("aios_single_owner_governance.task_credential_broker.os.geteuid", return_value=1000, create=True):
+            with self.assertRaisesRegex(ContractDenied, "ROOT_BROKER_REQUIRED"):
+                broker.provision(
+                    identity="codex",
+                    token_name="pr36-a8644fc172bd-status",
+                    encrypted_path=Path("/run/aios-task-credential-broker/status.cred"),
+                )
+
+    def test_partial_status_side_effects_are_reconciled_without_duplicate_writes(self):
+        item = TaskGrant.from_dict(grant())
+        api = FakeAPI(fail_status_after=2)
+        journal = self.journal(item)
+        with self.assertRaisesRegex(ContractDenied, "SIMULATED_STATUS_FAILURE"):
+            StatusBroker(item, journal, api, b"audit-key").publish(evidence())
+        self.assertEqual(len(api.status_rows), 2)
+        api.fail_status_after = None
+        receipt = StatusBroker(item, journal, api, b"audit-key").publish(evidence())
+        self.assertEqual(receipt["published_count"], 5)
+        self.assertEqual(len(api.status_rows), 5)
+
+    def test_post_merge_response_loss_is_reconciled_from_exact_remote_ref(self):
+        item = TaskGrant.from_dict(grant())
+        api = FakeAPI(with_statuses=True, fail_after_merge=True)
+        journal = self.journal(item)
+        with self.assertRaisesRegex(ContractDenied, "SIMULATED_POST_MERGE_LOSS"):
+            MergeBroker(item, journal, api, b"audit-key").merge()
+        api.fail_after_merge = False
+        receipt = MergeBroker(item, journal, api, b"audit-key").merge()
+        self.assertTrue(receipt["reconciled"])
+        self.assertEqual(receipt["governance_oid"], HEAD)
+
+
+if __name__ == "__main__":
+    unittest.main()