governance(release): request V2R1C4 Phase B successor approval

governance: replace bootstrap with RFC3339 offset fix
2026-08-01 13:19:13 +08:00 · 2026-08-01 02:50:00 +00:00
14 changed files with 1668 additions and 0 deletions
@@ -0,0 +1,185 @@
+{
+  "approval_type": "TASK_STATUS_MERGE_BROKER_BOOTSTRAP_RFC3339_FIX",
+  "schema_version": 2,
+  "repository": "aios/aios",
+  "expected_governance_base_oid": "8814dff20222625b60454c613b61ccf15fa20e51",
+  "request_branch": "patch/codex/AIOS-V2R1C4-BOOTSTRAP-RFC3339-OFFSET-FIX",
+  "target_ref": "refs/heads/governance/approvals",
+  "issued_at": "2026-08-01T02:50:00Z",
+  "expires_at": "2026-08-01T14:50:00Z",
+  "required_human_approvals": 1,
+  "allowed_human_approvers": ["owner"],
+  "require_latest_head_approval": true,
+  "approval_scopes": [
+    "BOOTSTRAP_FIX_APPROVAL",
+    "GITEA_INTERNAL_TOKEN_ROTATION_APPROVAL"
+  ],
+  "supersedes": {
+    "pr_number": 37,
+    "head_oid": "57697d81cf3d1fef23798ec390418efe8b64f4e9",
+    "tree_oid": "b58af1925318a0eefa6cc709e8d52ba57bb20d7f",
+    "reason": "APPROVED_BOOTSTRAP_RUNTIME_CONTRACT_MISMATCH",
+    "old_review_reusable": false,
+    "old_pr_immutable": true
+  },
+  "exact_head_binding": {
+    "method": "GITEA_EXACT_HEAD_OWNER_REVIEW_AND_PR_METADATA",
+    "head_oid_resolved_after_commit": true,
+    "tree_oid_resolved_after_commit": true,
+    "manifest_blob_oid_resolved_after_commit": true,
+    "manifest_sha256_resolved_after_commit": true,
+    "head_change_invalidates_review": true
+  },
+  "timestamp_contract": {
+    "id": "RFC3339_OFFSET_AWARE_V2",
+    "canonical_timezone": "UTC",
+    "raw_timestamp_preserved": true,
+    "naive_datetime_allowed": false,
+    "accepted_timezone_forms": ["Z", "+HH:MM", "-HH:MM"],
+    "fractional_seconds_allowed": true,
+    "lowercase_z_allowed": false,
+    "expiry_boundary": "REVIEW_MUST_BE_STRICTLY_BEFORE_EXPIRY"
+  },
+  "internal_token_rotation": {
+    "secret_class": "GITEA_INTERNAL_TOKEN",
+    "old_secret_fingerprint_algorithm": "SHA256",
+    "old_secret_fingerprint": "42e2af774700405ae32d55c98ca010fdb5da890b08f9e12e77372133c68e0562",
+    "old_secret_reference": "gitea-config://security/INTERNAL_TOKEN",
+    "affected_consumers": [
+      "gitea-config-residue:/data/aios-gitops/runtime/gitea-bin/custom/conf/app.ini.bak-20260706-auth-gitops",
+      "gitea-config:/data/aios-gitops/runtime/gitea-bin/custom/conf/app.ini",
+      "systemd-service:aios-gitea.service"
+    ],
+    "replacement_secret_reference": "systemd-credential-encrypted://gitea-internal-token-v2",
+    "rotation_method": "ATOMIC_REPLACE_THEN_REVOKE",
+    "rollback_plan_path": "evidence/governance/bootstrap-rfc3339-fix-20260801/ROLLBACK_PLAN.json",
+    "rollback_plan_sha256": "3283f4c3d6bb7f34ac893789f7e9f5df6c68b7807203e6f471ad92e8e007ff3f",
+    "audit_window_start": "2026-08-01T00:00:00Z",
+    "allowed_operations": [
+      "generate_replacement_internal_token_to_encrypted_runtime_credential",
+      "validate_replacement_internal_token_without_logging_plaintext",
+      "atomically_replace_exact_gitea_internal_token_config_value",
+      "restart_exact_aios_gitea_service",
+      "verify_exact_consumers_healthy",
+      "verify_old_internal_token_authentication_rejected",
+      "remove_or_redact_exact_old_secret_config_residue",
+      "audit_exposure_window",
+      "write_immutable_security_event_receipt"
+    ],
+    "access_token_table_exception_applicable": false,
+    "plaintext_secret_in_manifest_allowed": false,
+    "plaintext_secret_in_argv_allowed": false,
+    "plaintext_secret_in_logs_allowed": false,
+    "plaintext_secret_in_git_allowed": false
+  },
+  "phase_b": {
+    "approval_mode": "TWO_STAGE_REQUIRED",
+    "reason": "CURRENT_SCHEMAS_DO_NOT_ATOMICALLY_BIND_BOOTSTRAP_ACTIVATION_AND_FULL_TASK_SCOPED_RELEASE_PAYLOAD",
+    "old_pr_number": 36,
+    "old_head_oid": "a8644fc172bd8ebbba17a16e6bf586c93fd1e793",
+    "old_approval_reusable": false,
+    "successor_required_after_bootstrap_activation": true,
+    "frozen_release_objects_unchanged": true
+  },
+  "frozen_release_objects": {
+    "source_commit_oid": "1857b88690cd1c0119df984534c260b6f217a21a",
+    "source_tree_oid": "17c153de2338f570f70cb7407e7dda3c596b524b",
+    "evidence_closure_oid": "057f5923f72cdb8d002ae40bc0cd584961825801",
+    "evidence_tree_oid": "30be66d153fe9d3758f5fdba89ca4169beab38f8",
+    "git_bundle_sha256": "762e280e240988282bc2c2e65fafecedd596400ab6829016947e84559508a015",
+    "tag_oid": "7f82d77cf7702a6d695b4f0b74da7a9a9ecd7fa3",
+    "peeled_target_oid": "057f5923f72cdb8d002ae40bc0cd584961825801",
+    "release_artifact_sha256": "39523caa48f39b33e5e5df785bb568e0381d8b20301746c280442f73f294f2b2",
+    "rollback_artifact_sha256": "dd0dbbe21db14722996636347ea1d22917436c688d5f0b7055b5bfb1776c4d95",
+    "release_root_sha256": "16e4e881287cef4a02525a6a5b337d3e12995e510832fe04fea2ac196885f3ef",
+    "evidence_manifest_sha256": "3b0b72af400997385e817e9bec677942288f422b69c2d2b52a8a00407fc395e0"
+  },
+  "allowed_status_contexts": [
+    "aios/branch-protection",
+    "aios/evidence-closure",
+    "aios/secret-scan",
+    "aios/state-machine",
+    "aios/validator"
+  ],
+  "allowed_identities": ["codex", "gpt"],
+  "allowed_operations": [
+    "install_exact_approved_bootstrap_files",
+    "sign_task_grant_after_owner_exact_head_review",
+    "issue_task_status_credential",
+    "issue_task_merge_credential",
+    "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_outside_request_branch",
+    "main_push",
+    "tag_write",
+    "registry_write",
+    "baseline_write",
+    "production_write",
+    "branch_protection_change",
+    "force_push",
+    "squash",
+    "rebase",
+    "merge_commit",
+    "approval_impersonation",
+    "commit_status_database_write",
+    "database_write_outside_exact_task_access_token_revocation",
+    "reuse_pr36_or_pr37_review",
+    "plaintext_secret_persistence"
+  ],
+  "runtime_controls": {
+    "global_gitops_writes_enabled": false,
+    "systemd_encrypted_credentials_required": true,
+    "systemd_projection_mode": "0440_ROOT_ROOT_CREDENTIALS_DIRECTORY_ONLY",
+    "systemd_gitea_token_embedded_name": "gitea-token",
+    "sqlite_parent_directory_metadata_write_required": true,
+    "sqlite_sibling_paths_read_only": true,
+    "sqlite_quota_journal_mode": "WAL",
+    "sqlite_quota_synchronous": "FULL",
+    "status_evidence_digest_bound_in_signed_grant": true,
+    "merge_method": "fast-forward-only",
+    "commit_status_database_write_allowed": false
+  },
+  "approved_files": {
+    "deploy/systemd/aios-task-credential-broker@.service": "95f8da02600b3c908013ffa723d79f6fca0c7a4332a077a7b673e97ce35bcd69",
+    "deploy/systemd/aios-task-credential-expiry-sweeper.service": "72a336523983b2486300c4e52f2f01dce392aaf6ffd953041c209d204df85717",
+    "deploy/systemd/aios-task-credential-expiry-sweeper.timer": "49ff9c80cbd0b44fd584dc7d29f4db0d042e33f325b1ed7af81a5be42f1cd011",
+    "deploy/systemd/aios-task-merge-broker@.service": "4e6b8fcf541f225e58ea080107fc30deb0c4ac9eac70217ff9e57081245c106a",
+    "deploy/systemd/aios-task-status-broker@.service": "378a2e8f658cb56cac5dcc151772ec6a00d86f6c1ce3106509b9aa87549dc9da",
+    "evidence/governance/bootstrap-rfc3339-fix-20260801/ROLLBACK_PLAN.json": "3283f4c3d6bb7f34ac893789f7e9f5df6c68b7807203e6f471ad92e8e007ff3f",
+    "evidence/governance/bootstrap-rfc3339-fix-20260801/TEST_SUMMARY.json": "c7d28139fddb9666e60fd62a9a27803dd00a1ebbccbbb75a5121cd8ed3d20fc2",
+    "services/aios-single-owner-governance/aios_single_owner_governance/task_credential_broker.py": "3b2fbbf2698a2d8385a2bf9eccdebbea629642908cbc7a1e8251476475e0d647",
+    "services/aios-single-owner-governance/aios_single_owner_governance/task_credential_expiry_sweeper.py": "8d383768eb0322c84a35282c126bee26a3e6419b125780424ffc7793a504efdf",
+    "services/aios-single-owner-governance/aios_single_owner_governance/task_status_merge_broker.py": "697d8deccdf2fec902dab38a0436075d78a95c1c5b410e115ee74cf79f477e63",
+    "services/aios-single-owner-governance/task-status-merge-grant-v1.schema.json": "34472624261203b5c1a215c141887ea3fed5d68fb3c503c7db2ad9a717862597",
+    "tests/single_owner_governance/test_task_status_merge_broker.py": "af1c5844ee0f7d6907f8cb68d5f503f3bdbd59bb05150e5364eaf7bf2b37be1a"
+  },
+  "aggregate_bindings": {
+    "broker_module_set_sha256": "c73ccaf8f60cf2446a5d4c0b4baf5456cb77961fda682285cb0a4f509ed49caa",
+    "systemd_definition_set_sha256": "3591be96f1717e7a0a57957aafbb233eec2b1617a8fb938889ccd312d812d066",
+    "grant_schema_sha256": "34472624261203b5c1a215c141887ea3fed5d68fb3c503c7db2ad9a717862597",
+    "test_summary_sha256": "c7d28139fddb9666e60fd62a9a27803dd00a1ebbccbbb75a5121cd8ed3d20fc2",
+    "python_executable_sha256": "e1efa562c2cc2e35521a5c9c9b9939921001ff8ca9708a13ef15ace68cc2ccd7"
+  },
+  "machine_gate_requirements": {
+    "compileall": "PASS",
+    "unit_tests": "PASS",
+    "rfc3339_offset_tests": "PASS",
+    "json_schema": "PASS",
+    "systemd_validation": "PASS",
+    "changed_path_secret_scan": "PASS",
+    "fault_injection": "PASS",
+    "scope_boundary_tests": "PASS"
+  }
+}
@@ -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 /data/aios-gitops/runtime/gitea-bin/data/actions_artifacts /data/aios-gitops/runtime/gitea-bin/data/actions_log /data/aios-gitops/runtime/gitea-bin/data/attachments /data/aios-gitops/runtime/gitea-bin/data/avatars /data/aios-gitops/runtime/gitea-bin/data/gitea.db.bak.20260708133830 /data/aios-gitops/runtime/gitea-bin/data/gitea.db.bak-20260710-0105 /data/aios-gitops/runtime/gitea-bin/data/home /data/aios-gitops/runtime/gitea-bin/data/indexers /data/aios-gitops/runtime/gitea-bin/data/jwt /data/aios-gitops/runtime/gitea-bin/data/packages /data/aios-gitops/runtime/gitea-bin/data/queues /data/aios-gitops/runtime/gitea-bin/data/repo-archive /data/aios-gitops/runtime/gitea-bin/data/repo-avatars /data/aios-gitops/runtime/gitea-bin/data/repositories /data/aios-gitops/runtime/gitea-bin/data/tmp
+ReadWritePaths=/data/aios-gitops/runtime/gitea-bin/data /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
@@ -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=/etc/aios-gitops/task-status-merge/allowed_signers /data/aios-gitops/runtime/gitea-bin/data/actions_artifacts /data/aios-gitops/runtime/gitea-bin/data/actions_log /data/aios-gitops/runtime/gitea-bin/data/attachments /data/aios-gitops/runtime/gitea-bin/data/avatars /data/aios-gitops/runtime/gitea-bin/data/gitea.db.bak.20260708133830 /data/aios-gitops/runtime/gitea-bin/data/gitea.db.bak-20260710-0105 /data/aios-gitops/runtime/gitea-bin/data/home /data/aios-gitops/runtime/gitea-bin/data/indexers /data/aios-gitops/runtime/gitea-bin/data/jwt /data/aios-gitops/runtime/gitea-bin/data/packages /data/aios-gitops/runtime/gitea-bin/data/queues /data/aios-gitops/runtime/gitea-bin/data/repo-archive /data/aios-gitops/runtime/gitea-bin/data/repo-avatars /data/aios-gitops/runtime/gitea-bin/data/repositories /data/aios-gitops/runtime/gitea-bin/data/tmp
+ReadWritePaths=/data/aios-gitops/runtime/gitea-bin/data /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
@@ -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
@@ -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
@@ -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
@@ -0,0 +1,28 @@
+{
+  "plan_id": "AIOS-BOOTSTRAP-RFC3339-FIX-ROLLBACK-V1",
+  "schema_version": 1,
+  "repository": "aios/aios",
+  "approved_base_oid": "8814dff20222625b60454c613b61ccf15fa20e51",
+  "governance_history_rewrite_allowed": false,
+  "steps": [
+    "stop_uncompleted_task_status_merge_units",
+    "revoke_exact_task_tokens_and_verify_http_401",
+    "remove_all_runtime_credential_projections",
+    "remove_unactivated_runtime_units_and_non_root_task_accounts",
+    "verify_governance_refs_and_production_refs_unchanged"
+  ],
+  "internal_token_rotation": {
+    "rollback_allowed_before_old_secret_revocation": true,
+    "rollback_source": "root_only_encrypted_runtime_credential",
+    "rollback_allowed_after_old_secret_revocation": false,
+    "post_revocation_recovery": "issue_another_replacement_secret_under_new_governed_operation",
+    "plaintext_secret_in_evidence_allowed": false
+  },
+  "required_closure": {
+    "temporary_accounts_remaining": 0,
+    "temporary_credentials_remaining": 0,
+    "credential_projections_remaining": 0,
+    "old_task_tokens_http_status": 401,
+    "AIOS_GITOPS_WRITES_ENABLED": false
+  }
+}
@@ -0,0 +1,30 @@
+{
+  "schema_id": "AIOS_BOOTSTRAP_RFC3339_FIX_TEST_SUMMARY",
+  "schema_version": 1,
+  "candidate_tree_before_manifest": "c72b13bfe0656de9deeff9652e3e13ada81b110e",
+  "local_unittest": {
+    "tests_passed": 19,
+    "tests_failed": 0,
+    "tests_skipped": 0
+  },
+  "remote_posix": {
+    "pytest_tests_passed": 19,
+    "pytest_subtests_passed": 16,
+    "tests_failed": 0,
+    "tests_skipped": 0,
+    "compileall": "PASS",
+    "json_schema": "PASS",
+    "systemd_validation": "PASS"
+  },
+  "secret_scan": {
+    "scanner": "gitleaks",
+    "exact_changed_paths": "PASS",
+    "exact_changed_path_findings": 0,
+    "whole_tree_existing_findings": 10,
+    "whole_tree_findings_claimed_as_new": false
+  },
+  "python_executable": {
+    "path": "/usr/bin/python3.12",
+    "sha256": "e1efa562c2cc2e35521a5c9c9b9939921001ff8ca9708a13ef15ace68cc2ccd7"
+  }
+}
@@ -0,0 +1,238 @@
+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", "--name=gitea-token", "-", 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:
+            rows = 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)).fetchall()
+        finally:
+            connection.close()
+        if len(rows) != 1:
+            raise ContractDenied("TOKEN_DATABASE_BINDING_MISSING")
+        row = rows[0]
+        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())
@@ -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())
@@ -0,0 +1,472 @@
+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", "timestamp_contract", "canonical_timezone",
+    "raw_timestamp_preserved", "naive_datetime_allowed",
+})
+RFC3339 = re.compile(
+    r"^\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:Z|[+-]\d{2}:\d{2})$"
+)
+
+
+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 value != value.strip() or not RFC3339.fullmatch(value):
+        raise ContractDenied("TIMESTAMP_INVALID")
+    try:
+        normalized = value[:-1] + "+00:00" if value.endswith("Z") else value
+        parsed = datetime.fromisoformat(normalized)
+    except ValueError as exc:
+        raise ContractDenied("TIMESTAMP_INVALID") from exc
+    if parsed.tzinfo is None or parsed.utcoffset() is None:
+        raise ContractDenied("TIMESTAMP_TIMEZONE_REQUIRED")
+    return parsed.astimezone(timezone.utc)
+
+
+def _utc_timestamp(value: datetime) -> str:
+    normalized = value.astimezone(timezone.utc).isoformat()
+    return normalized[:-6] + "Z" if normalized.endswith("+00:00") else normalized
+
+
+@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")
+        if (value["timestamp_contract"] != "RFC3339_OFFSET_AWARE_V2"
+                or value["canonical_timezone"] != "UTC"
+                or value["raw_timestamp_preserved"] is not True
+                or value["naive_datetime_allowed"] is not False):
+            raise ContractDenied("TIMESTAMP_CONTRACT_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)
+        mode = stat.S_IMODE(metadata.st_mode)
+        credential_directory = os.environ.get("CREDENTIALS_DIRECTORY")
+        systemd_projection = (
+            credential_directory is not None
+            and path.parent == Path(credential_directory)
+            and mode == 0o440
+            and metadata.st_uid == 0
+            and metadata.st_gid == 0
+        )
+        if not stat.S_ISREG(metadata.st_mode) or (mode != 0o400 and not systemd_projection):
+            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)
+        journal_mode = self.connection.execute("PRAGMA journal_mode=WAL").fetchone()
+        if journal_mode is None or str(journal_mode[0]).lower() != "wal":
+            raise ContractDenied("JOURNAL_MODE_INVALID")
+        self.connection.execute("PRAGMA synchronous=FULL")
+        synchronous = self.connection.execute("PRAGMA synchronous").fetchone()
+        if synchronous is None or int(synchronous[0]) != 2:
+            raise ContractDenied("JOURNAL_SYNCHRONOUS_INVALID")
+        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")
+    raw_timestamp = str(latest.get("submitted_at") or "")
+    submitted = _time(raw_timestamp)
+    if submitted < grant.issued_at or submitted >= grant.expires_at:
+        raise ContractDenied("OWNER_APPROVAL_EXPIRED")
+    normalized_review = dict(latest)
+    normalized_review["review_timestamp_raw"] = raw_timestamp
+    normalized_review["review_timestamp_utc"] = _utc_timestamp(submitted)
+    return pr, normalized_review
+
+
+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")
+        _, review = _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, "review_timestamp_raw": review["review_timestamp_raw"], "review_timestamp_utc": review["review_timestamp_utc"]}
+        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, review = _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", "review_timestamp_raw": review["review_timestamp_raw"], "review_timestamp_utc": review["review_timestamp_utc"]}
+        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())
@@ -0,0 +1,69 @@
+{
+  "$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", "timestamp_contract", "canonical_timezone",
+    "raw_timestamp_preserved", "naive_datetime_allowed"
+  ],
+  "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},
+    "timestamp_contract": {"const": "RFC3339_OFFSET_AWARE_V2"},
+    "canonical_timezone": {"const": "UTC"},
+    "raw_timestamp_preserved": {"const": true},
+    "naive_datetime_allowed": {"const": false},
+    "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"}
+  }
+}
@@ -0,0 +1,440 @@
+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,
+    _time,
+    _utc_timestamp,
+    canonical,
+    load_signed_task_grant,
+    read_runtime_secret,
+)
+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,
+        "timestamp_contract": "RFC3339_OFFSET_AWARE_V2",
+        "canonical_timezone": "UTC",
+        "raw_timestamp_preserved": True,
+        "naive_datetime_allowed": False,
+        "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, review_timestamp: str | None = None,
+                 reviewer: str = "owner", review_commit: str = HEAD):
+        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
+        self.review_timestamp = review_timestamp or datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")
+        self.reviewer = reviewer
+        self.review_commit = review_commit
+
+    def get(self, path: str):
+        if path.endswith("/reviews"):
+            return [{
+                "id": 76,
+                "state": "APPROVED",
+                "commit_id": self.review_commit,
+                "official": True,
+                "stale": False,
+                "dismissed": False,
+                "submitted_at": self.review_timestamp,
+                "user": {"id": 2 if self.reviewer == "owner" else 42, "login": self.reviewer},
+            }]
+        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_rfc3339_offset_parser_normalizes_to_utc(self):
+        cases = {
+            "2026-08-01T02:03:02Z": "2026-08-01T02:03:02Z",
+            "2026-08-01T10:03:02+08:00": "2026-08-01T02:03:02Z",
+            "2026-07-31T18:03:02-08:00": "2026-08-01T02:03:02Z",
+            "2026-08-01T10:03:02.123456+08:00": "2026-08-01T02:03:02.123456Z",
+        }
+        for raw, expected in cases.items():
+            with self.subTest(raw=raw):
+                self.assertEqual(_utc_timestamp(_time(raw)), expected)
+
+    def test_rfc3339_offset_parser_rejects_ambiguous_or_invalid_values(self):
+        invalid = [
+            "2026-08-01T10:03:02", "2026-08-01", "2026-08-01T10:03:02+0800",
+            "2026-08-01T10:03:02+25:00", "2026-08-01T10:03:02z",
+            " 2026-08-01T10:03:02+08:00", "2026-08-01T10:03:02+08:00 ",
+            None, 1, [],
+        ]
+        for raw in invalid:
+            with self.subTest(raw=raw):
+                with self.assertRaisesRegex(ContractDenied, "TIMESTAMP_INVALID"):
+                    _time(raw)  # type: ignore[arg-type]
+
+    def test_offset_review_preserves_raw_and_canonical_utc_in_receipt(self):
+        value = grant()
+        value["issued_at"] = "2026-08-01T00:00:00Z"
+        value["expires_at"] = "2026-08-01T12:00:00Z"
+        item = TaskGrant.from_dict(value, now=datetime(2026, 8, 1, 3, tzinfo=timezone.utc))
+        api = FakeAPI(review_timestamp="2026-08-01T10:03:02+08:00")
+        receipt = StatusBroker(item, self.journal(item), api, b"audit-key").publish(evidence())
+        self.assertEqual(receipt["review_timestamp_raw"], "2026-08-01T10:03:02+08:00")
+        self.assertEqual(receipt["review_timestamp_utc"], "2026-08-01T02:03:02Z")
+
+    def test_review_expiry_boundary_is_fail_closed(self):
+        value = grant()
+        value["issued_at"] = "2026-08-01T00:00:00Z"
+        value["expires_at"] = "2026-08-01T02:03:02Z"
+        item = TaskGrant.from_dict(value, now=datetime(2026, 8, 1, 1, tzinfo=timezone.utc))
+        for raw in ("2026-08-01T02:03:02Z", "2026-08-01T02:03:03Z"):
+            with self.subTest(raw=raw), self.assertRaisesRegex(ContractDenied, "OWNER_APPROVAL_EXPIRED"):
+                StatusBroker(item, self.journal(item), FakeAPI(review_timestamp=raw), b"audit-key").publish(evidence())
+
+    def test_wrong_reviewer_and_reviewed_head_remain_rejected(self):
+        item = TaskGrant.from_dict(grant())
+        for api in (FakeAPI(reviewer="codex"), FakeAPI(review_commit="e" * 40)):
+            with self.assertRaisesRegex(ContractDenied, "OWNER_EXACT_HEAD_APPROVAL_MISSING"):
+                StatusBroker(item, self.journal(item), 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_systemd_credential_projection_mode_and_name_are_supported_exactly(self):
+        metadata = mock.Mock(st_mode=0o100440, st_uid=0, st_gid=0)
+        credential = Path("/run/credentials/unit/gitea-token")
+        with mock.patch.dict("os.environ", {"CREDENTIALS_DIRECTORY": str(credential.parent)}), \
+             mock.patch("aios_single_owner_governance.task_status_merge_broker.os.open", return_value=7), \
+             mock.patch("aios_single_owner_governance.task_status_merge_broker.os.fstat", return_value=metadata), \
+             mock.patch("aios_single_owner_governance.task_status_merge_broker.os.read", return_value=b"x" * 40), \
+             mock.patch("aios_single_owner_governance.task_status_merge_broker.os.close"):
+            self.assertEqual(read_runtime_secret(credential), b"x" * 40)
+        item = TaskGrant.from_dict(grant())
+        broker = TaskCredentialBroker(item, self.root / "provision-audit", b"audit-key")
+        connection = mock.MagicMock()
+        connection.execute.return_value.fetchall.return_value = [(101, 42)]
+        issued = mock.Mock(stdout=b"x" * 40)
+        encrypted = mock.Mock(stdout=b"")
+        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"), \
+             mock.patch("aios_single_owner_governance.task_credential_broker.os.chmod"), \
+             mock.patch("aios_single_owner_governance.task_credential_broker.subprocess.run", side_effect=[issued, encrypted]) as runner, \
+             mock.patch("aios_single_owner_governance.task_credential_broker.sqlite3.connect", return_value=connection):
+            broker.provision(identity="codex", token_name="pr37-dddddddddddd-status", encrypted_path=Path("/run/aios-task-credential-broker/status.cred"))
+        self.assertIn("--name=gitea-token", runner.call_args_list[1].args[0])
+
+    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()
Author	SHA1	Message	Date
codex	52a60a3c41	governance(release): request V2R1C4 Phase B successor approval	2026-08-01 13:19:13 +08:00
AIOS Governance Automation	9303a5d684	governance: replace bootstrap with RFC3339 offset fix aios/branch-protection task-scoped PASS evidence_sha256=9cda4ebf561ae96e3f2f21eb4d4b89541626026fe521d089b7d5ec6d4af80787 Details aios/evidence-closure task-scoped PASS evidence_sha256=2239bd1868318c50d4e3287ce94adb92c2d13759aa8ef01cfea97142ae8a94b4 Details aios/secret-scan task-scoped PASS evidence_sha256=4ae633ace2f93b4093545108b2511c35530f1f1e89058bbb2633151f5d166248 Details aios/state-machine task-scoped PASS evidence_sha256=ce63ce87cfdf0a17d4c1f0a0a68df1be8a48596c44e914fa8604ee3d05f8ead2 Details aios/validator task-scoped PASS evidence_sha256=469ef5a951db82e7d4b65554185cd337e5c4c6b9d8dcef92c4babec6d0ff9daf Details	2026-08-01 02:50:00 +00:00