feat: close production trust and operations gates

.github/workflows/provider-promotion-canary.yml

@@ -33,6 +33,10 @@ on:
         description: "Maximum revocation or kill-switch propagation SLA"
         required: false
         default: "300"
+      max_observation_age_seconds:
+        description: "Maximum accepted age of each no-cache readiness response"
+        required: false
+        default: "120"
 
 permissions:
   contents: read
@@ -65,6 +69,7 @@ jobs:
           EXPECTED_KEY_ID: ${{ inputs.expected_key_id }}
           EXPECTED_IMAGE_DIGEST: ${{ inputs.expected_image_digest }}
           PROPAGATION_DEADLINE_SECONDS: ${{ inputs.propagation_deadline_seconds }}
+          MAX_OBSERVATION_AGE_SECONDS: ${{ inputs.max_observation_age_seconds }}
         run: |
           set -euo pipefail
           umask 077
@@ -76,6 +81,7 @@ jobs:
             --node "$NODE_A"
             --node "$NODE_B"
             --service-key-file .promotion-control-private/service-key
+            --max-observation-age-seconds "$MAX_OBSERVATION_AGE_SECONDS"
           )
           if [ "$MODE" = "canary" ]; then
             args+=(

project-docs/V2-production-rollback-capacity-audit-2026-08-12.md

@@ -0,0 +1,201 @@
+# V2 production rollback and capacity audit
+
+Date: 2026-08-12
+
+## Scope and safety boundary
+
+This audit began with repository tests and read-only SSH checks, then used an
+authorized maintenance window for backup/restore verification and a complete
+application rollback/forward rehearsal. The public edge remained stopped.
+PostgreSQL, Redis and MinIO were not rolled back, no database downgrade was run,
+and DNS/public traffic were not changed.
+
+## Result
+
+The primary-host capacity policy is currently within its configured limits,
+but it has little percentage headroom: the latest production health artifact
+reported `/` at 84.45% used with 28.90 GiB free, immediately below the 85%
+warning ceiling. `/home` was 16.94% used with 192.14 GiB free, and the off-host
+backup mount was 14.42% used with 68.24 GiB free. Both required paths were
+observed as separate mounts by the production health probe.
+
+Capacity code and repository-level alert behavior pass their automated tests.
+The production health unit wires the protected
+`PRODUCTION_ALERT_WEBHOOK_URL_FILE` into the probe, and production preflight
+fails closed unless that file contains an owner-restricted HTTPS URL. A receiver
+on the independent host `47.94.178.212` accepted the controlled warning and
+returned a receiver-side event ID, receipt time and body hash. The normal 85%
+threshold was restored and reported `/` at 84.45%, status `pass`.
+
+The production timer remains enabled while the edge is intentionally stopped,
+so the edge check is expected to fail closed. To avoid sending the identical
+known failure every minute, the probe now retains an atomic, non-secret alert
+state: the first failure, any check-set change and an hourly reminder are sent;
+failed delivery retries on the next run; a healthy run resets the state.
+
+A complete application version rollback and forward restoration was rehearsed
+from current release `d6a6752258a555f3a40894cdcf88917659538492` to approved
+previous release `c24f7a4f1ee041d92feca96bf147e423e58da17c`, then back to the
+current release. All four application services were stopped/recreated in both
+directions. The Contract API and worker digests were identical between those
+releases; the AI API image changed. This limitation is recorded rather than
+misrepresenting the exercise as four distinct image rollbacks. Health, schemas,
+worker readiness, contract/template reads and audit/outbox continuity passed in
+both directions. The alert and complete version rollback gates are **closed**.
+
+## Read-only field evidence
+
+Primary host: `192.168.3.90` (host-reported name `10.20.0.11`).
+
+Observed production services:
+
+- Contract API, Contract review worker, AI Platform API, AI worker, PostgreSQL,
+  Redis and MinIO were running and healthy.
+- Contract API was pinned to
+  `git.yuqei.com/yuqei/ai-contract-platform-contract-api@sha256:a6969d90680f866232f1ad2c09607fd63f226476028b6c0f34c3175e7407549f`.
+- AI Platform API was pinned to
+  `git.yuqei.com/yuqei/ai-contract-platform-ai-api@sha256:05faf2aa67d258708387daa7007442928b8d786814f1c5e8a619db4b1ae9d8b3`.
+- The latest normal-policy production health artifact had status `fail` only
+  because `edge_health` returned HTTP 503. Contract/AI health, schema checks,
+  Provider runtime, worker health, backup freshness and all three capacity
+  checks passed.
+- A controlled capacity warning was received as event
+  `yuqei-alert-45b05755b2fc4c498dd4eb8cd3c426ed`; its posted body SHA-256 was
+  `933d5741e600efec3190e79892a93165bb94a57f00976184fde31ec43148eeff`.
+- The receiver-owned mode `0600` receipt SHA-256 was
+  `8816696a8829215ce7e7821a0965d35d48a27651c89b1a267846d84bd664f909`.
+- Backup `backup-20260812T025650Z` completed in quiesced mode with four
+  artifacts. Disposable restoration of all three PostgreSQL databases passed;
+  summary SHA-256 was
+  `70e6c223a6870dd0d8fe28365bda7f883fc6827a77ddcc0da397dd77cbee4370`.
+- `/etc/yuqei/production.env` existed with mode `0600`.
+- `yuqei-production-health.timer` was enabled and active at a one-minute cadence.
+- `/etc/yuqei/monitoring/alert-webhook-url` is configured without exposing its
+  secret URL in repository evidence. It is mode `0600`, owned by the unprivileged
+  probe account. The receiver is loopback-bound behind HTTPS.
+- After deploying deduplication, the first expected `edge_health` failure was
+  delivered as `yuqei-alert-9014cb51c85241f5b63836bf8cb1a510`; an immediate
+  second identical run recorded `duplicate_within_resend_window`, did not call
+  the receiver, left the timer active and kept the edge replica count at `0`.
+- Current production read-only preflight passed with `failed=0`. Its retained
+  summary SHA-256 is
+  `3b1e2e3b25272adf4c1443fb98cbe7284b520d82808836f879de58c4e4240556`.
+
+## Complete rollback and forward evidence
+
+- Remote evidence directory:
+  `/opt/apps/ai-contract-platform-v2-production/project-docs/qa-artifacts/v2-production-full-rollback/20260812T030852Z`.
+- Manifest SHA-256:
+  `2ac616acc69868a8274ab0047d42dcdcd2b8ff29660ac68486fcbdc007d1d4a4`.
+- Evidence file count: `40`; manifest status: `passed`.
+- Before, rollback and forward phases all passed Contract health/schema, AI
+  health/schema, worker health, contract-list read, template-list read and
+  audit/outbox count continuity.
+- The edge replica count remained `0`. Provider readiness remained blocked by
+  the intentional kill switch and, on the previous release, missing receipt
+  binding. This is expected fail-closed behavior, not a failed recovery check.
+- An exit trap restored the current environment on any failure. Final inspection
+  confirmed the current release and all four application services healthy.
+
+## Independent public-policy trust evidence
+
+- Legal approval ID: `policy-approval-0887bbd5149bf7a1c41cb35e`.
+- Trust request SHA-256:
+  `d2d76ee0789b235af67f1df3ae3a86237467ca72bbf76ad6a613e35a0f996eee`.
+- Independent operations reviewer: Chen Yong, stable approver ID
+  `chenyong@yuqei.com`, key ID `policy-trust-ops-2026q3`.
+- Independent public-key file SHA-256:
+  `0e2b147e5da9d1084da368944c720821a053b866282f1c5f9385dbad529e4009`.
+- Exact trust payload SHA-256:
+  `1b783cb6a9f1976d50f7d8b4c77e286425d7c8bec7c597057a172b6eaa113442`.
+- Signed trust approval ID:
+  `policy-key-trust-approval-09d9860e94d6df9fe74a0d1e`; approval
+  SHA-256:
+  `5093462cefff6a75e5f9a320f8bd4e033d9327e0234f84e9bbc258c36148fbbc`.
+- Trusted policy keyring SHA-256:
+  `41f9517dcd375afbeea4aaa4dce5d8bbe07db8f71e2706d47e7aec5ccd8bd388`.
+- Strict verification at `2026-08-12T12:40:02.047339+08:00` passed for all
+  four exact policy files without candidate-keyring override.
+- The signed decision binds the out-of-band identity/authority and legal-key
+  fingerprint attestations. `policy_activation_authorized=false`; the trusted
+  keyring remains local and has not been installed in production. No policy was
+  published or activated by this approval.
+
+Host filesystem observations were also checked directly with `df -P`: root was
+85% at `df` display precision, while `/home` was 17%. The application probe's
+non-reserved-capacity calculation is the authoritative threshold comparison and
+reported root at 84.45%.
+
+## Repository verification
+
+Executed on 2026-08-12:
+
+| Command | Result |
+| --- | --- |
+| `python -m pytest yuqei-infra/tests/test_observability_capacity.py yuqei-infra/tests/test_alert_receiver.py -q` | `26 passed` |
+| `python -m pytest yuqei-infra/tests -q` | `148 passed` |
+| `python -m py_compile` on all changed Python entrypoints | passed |
+| `python -m bandit -q` on all changed Python entrypoints | passed |
+| `git diff --check` | passed; only existing line-ending warnings |
+
+The capacity tests cover warning/critical percent and free-space thresholds,
+reserved filesystem blocks, separate-mount fail-closed behavior, webhook success,
+webhook timeout, receiver authentication and immutable receipts, healthy
+no-alert behavior, changed-state and one-hour resend deduplication, production
+preflight thresholds, and the installed systemd arguments.
+
+## Capacity and alert operating policy
+
+1. Keep the HTTPS URL in its owner-restricted file; never put its secret path or
+   token in the repository, environment file or audit report.
+2. Retain the receiver-side receipt, local summary and event ID for each
+   acceptance test.
+3. Keep one-hour unchanged-alert deduplication enabled. Do not suppress the first
+   failure, a changed check set, delivery retries or hourly reminders.
+4. Treat root at or above 85%, free below 20 GiB, or more than five percentage
+   points growth in 24 hours as a release stop. Do not delete business volumes,
+   Registry data or backups to satisfy the gate. Move rebuildable caches/logs only
+   after an explicit inventory and retention approval, or extend the root LV.
+
+## Complete version rollback acceptance policy
+
+The completed exercise followed these controls. Re-run it after a material
+release topology, image or migration change; do not infer future downgrade
+compatibility from this one result.
+
+1. Record the current release ID, all image digests, receipt/key fingerprints,
+   Alembic heads, container health and golden-smoke results.
+2. Take and verify a new off-host production backup. Restore all three dumps into
+   disposable `yuqei_restore_check_*` databases before continuing.
+3. Prove migration compatibility. If the previous application cannot read the
+   current schema, stop: use forward-fix recovery unless a separately reviewed,
+   data-preserving downgrade exists. Never infer downgrade safety from an Alembic
+   command succeeding in an empty database.
+4. With edge and public traffic still disabled, switch application services only
+   to the approved previous digests. Do not roll back PostgreSQL/Redis/MinIO data
+   or delete audit/outbox records.
+5. Verify health, schema checks, immutable release fingerprints, login/workspace,
+   contract list/detail, template drafting, review read path and audit continuity.
+6. Return to the current immutable digests, re-run migrations in the forward
+   direction, and repeat the same checks.
+7. Capture start/end times, RTO, every command exit code, before/rollback/forward
+   fingerprints, database heads, release-gate summaries and operator approval.
+8. Abort immediately on migration drift, partial health, missing audit/outbox data,
+   receipt mismatch, unexpected writes, or any need to use an unpinned image.
+
+Acceptance requires both rollback and forward-restoration checks to pass, the
+database and immutable audit history to remain intact, and an operator-owned
+evidence package. The 2026-08-12 rehearsal satisfies this gate for the recorded
+release pair.
+
+## Gate status
+
+| Gate | Status | Reason |
+| --- | --- | --- |
+| Capacity thresholds and live primary-host readings | pass, narrow headroom | root 84.45% used; warning begins at 85% |
+| Capacity timer and local alert contract | pass | timer active; automated warning/failure delivery semantics pass |
+| Real external alert delivery | pass | HTTPS receiver event and immutable receipt hashes captured |
+| Backup integrity and disposable DB restore | pass from existing field evidence | three dumps and restore checks completed |
+| Complete production version rollback and forward restore | pass | 40-file evidence manifest passed for previous/current release pair |
+| Independent public-policy trust approval | pass, policy not activated | Chen Yong signed the exact trust payload as operations reviewer; strict verification issued a non-activating trusted keyring |
+| Public traffic authorization | not in scope | edge/DNS/traffic remain separately controlled |

project-docs/V2-public-policy-offline-signing-runbook-2026-08-12.md

@@ -31,13 +31,20 @@
 | `service_agreement` | `service-agreement.md` | `214f5744039dc8ebf554f6ead06e5a9fb3ec08ea973719bb59fc8e3f8b1d5f48` |
 | `ai_use_notice` | `ai-use-notice.md` | `f261df70a29109e90a537f5812d871591e4787716e008d7966ed9b5e79ebcdfd` |
 
-当前未签名材料位于 `.runtime/production-closure/policies/2026-08-12/`：
+当前签批材料位于 `.runtime/production-closure/policies/2026-08-12/`：
 
 - review package SHA-256：`0b3c105feaf5025e20b6799701ec19b8580837ff2ad71297baf96af56a1d8e62`
 - pending payload SHA-256：`7f543f8980ceedc085d451c6c8d691d5ba93fa5f11baef276f4bd004a1b4eba5`
+- final payload SHA-256：`7969de1c7c8384657b42a3fd16e70bb83a0430d0100b33088c4a094c48b17312`
+- signed approval SHA-256：`84f45dd4ec188eb29509f8ce75787862099558b24ad1bd42dc27f976c24a5719`
+- approval ID：`policy-approval-0887bbd5149bf7a1c41cb35e`
 - candidate keyring SHA-256：`1f85a758252d571aec4233c864714ad2fb637940ebb2f0d2e37d1818b09f9df4`
+- independent trust payload SHA-256：`1b783cb6a9f1976d50f7d8b4c77e286425d7c8bec7c597057a172b6eaa113442`
+- independent trust approval ID：`policy-key-trust-approval-09d9860e94d6df9fe74a0d1e`
+- independent trust approval SHA-256：`5093462cefff6a75e5f9a320f8bd4e033d9327e0234f84e9bbc258c36148fbbc`
+- trusted keyring SHA-256：`41f9517dcd375afbeea4aaa4dce5d8bbe07db8f71e2706d47e7aec5ccd8bd388`
 
-这三个文件当前状态分别是 `pending_named_legal_review` 或 `candidate_not_trusted`，不得签名、发布或安装为生产信任材料。
+review package 和 pending payload 仍是历史输入材料。陈勇已作为独立运营复核人签署精确信任 payload；工具从原始 trust request、法务批准、候选 keyring、四份政策字节和独立公钥重新验证后生成 trusted keyring，并在未启用 candidate override 的情况下完成四份政策验签。`trust_basis.policy_activation_authorized=false`，所以该结果只关闭公钥信任审批门禁，不表示政策已经发布或生效。trusted keyring 当前仅保存在本机受控 `.runtime` 目录，尚未安装到生产。
 
 ## 审批人审阅清单
 
@@ -64,7 +71,7 @@ python yuqei-infra/scripts/prepare-policy-legal-approval.py finalize-payload `
   --output .runtime/production-closure/policies/2026-08-12/policy-approval-final-payload.json
 ```
 
-记录命令输出的最终 payload SHA-256。该 hash 仍未签名，审批人必须再次核对批准时间、失效时间和 exact hash。
+记录命令输出的最终 payload SHA-256。签名前，审批人必须再次核对批准时间、失效时间和 exact hash；本次最终 payload hash 为 `7969de1c7c8384657b42a3fd16e70bb83a0430d0100b33088c4a094c48b17312`。
 
 ## 离线签名
 
@@ -81,9 +88,81 @@ python yuqei-infra/scripts/prepare-policy-legal-approval.py sign `
 
 签名命令会拒绝错误确认字符串、非最终 payload、错误密码、非 Ed25519 私钥或与 payload 公钥绑定不一致的私钥。
 
+本次签名已完成并通过验签，产物为 `policy-legal-approval.json`，`approval_id` 为 `policy-approval-0887bbd5149bf7a1c41cb35e`。该结果仅证明精确政策 payload 已由具名审批人签署，不代表政策已发布或已生效。
+
+## 独立信任申请
+
+法务签名只证明审批人签署了精确政策 payload，不会自动使候选公钥成为生产信任。签名完成后，先生成不可变的信任申请：
+
+```powershell
+python yuqei-infra/scripts/prepare-policy-legal-approval.py prepare-trust-request `
+  --approval-file .runtime/production-closure/policies/2026-08-12/policy-legal-approval.json `
+  --candidate-keyring-file .runtime/production-closure/policies/2026-08-12/policy-legal-keyring.candidate.json `
+  --policy-dir project-docs/legal-policy-candidates/2026-08-12 `
+  --requested-at <ISO-8601-with-timezone> `
+  --output .runtime/production-closure/policies/2026-08-12/policy-key-trust-request.json
+```
+
+该命令会在审批有效期内重新验签，并绑定签名文件、候选 keyring 和四份政策原始字节。输出状态固定为 `pending_independent_trust_approval`，`activation_authorized` 固定为 `false`；它不会生成 trusted keyring、发布政策、切换 `effective`、启动 edge 或修改公网流量。
+
+安全或运维负责人必须通过签名流程之外的渠道核对审批人身份、授权范围和公钥指纹，具名批准信任申请并保留证据。不得仅把候选 JSON 的 `status` 字段手工改为 `trusted` 作为信任证据。
+
+独立审批人必须与法务审批人不同，并提供：真实姓名、稳定工作邮箱或 approver ID、职务、组织全称、独立 `key_id`、本人控制的 Ed25519 公钥 PEM、公钥有效期和本次批准有效期。approver ID 及公钥都不得与法务审批人相同；本次批准最长 7 天，且不得晚于原法务批准失效时间。
+
+先生成精确待签 payload：
+
+```powershell
+python yuqei-infra/scripts/prepare-policy-legal-approval.py prepare-trust-approval `
+  --trust-request-file .runtime/production-closure/policies/2026-08-12/policy-key-trust-request.json `
+  --approval-file .runtime/production-closure/policies/2026-08-12/policy-legal-approval.json `
+  --candidate-keyring-file .runtime/production-closure/policies/2026-08-12/policy-legal-keyring.candidate.json `
+  --policy-dir project-docs/legal-policy-candidates/2026-08-12 `
+  --trust-approver-public-key-file <independent-trust-approver-public.pem> `
+  --key-id <independent-key-id> `
+  --approver-id <independent-work-id> `
+  --reviewer <real-name> `
+  --role <security-or-operations-role> `
+  --organization <organization> `
+  --responsibility <security-or-operations> `
+  --approved-at <ISO-8601-with-timezone> `
+  --expires-at <ISO-8601-with-timezone> `
+  --key-not-before <ISO-8601-with-timezone> `
+  --key-not-after <ISO-8601-with-timezone> `
+  --output .runtime/production-closure/policies/2026-08-12/policy-key-trust-approval-payload.json
+```
+
+审批人核对命令输出的 `trust_approval_payload_sha256` 后，使用其本人控制且与法务密钥不同的私钥签名：
+
+待签 payload 固定包含三项不可省略声明：已通过线下渠道核验法务审批人身份和权限、已通过线下渠道核验法务签名公钥指纹，以及本人承担独立安全或运营复核责任。审批人必须在签名前真实完成这些核验；工具不会替审批人作此事实判断。
+
+```powershell
+python yuqei-infra/scripts/prepare-policy-legal-approval.py sign-trust-approval `
+  --payload-file .runtime/production-closure/policies/2026-08-12/policy-key-trust-approval-payload.json `
+  --private-key-file <independent-trust-approver-private.pem> `
+  --private-key-password-file <protected-password-file> `
+  --confirmation I_APPROVE_THIS_EXACT_POLICY_KEY_TRUST_PAYLOAD `
+  --output .runtime/production-closure/policies/2026-08-12/policy-key-trust-approval.json
+```
+
+最后从所有原始材料重新验签并生成新 keyring 文件：
+
+```powershell
+python yuqei-infra/scripts/prepare-policy-legal-approval.py issue-trusted-keyring `
+  --trust-approval-file .runtime/production-closure/policies/2026-08-12/policy-key-trust-approval.json `
+  --trust-request-file .runtime/production-closure/policies/2026-08-12/policy-key-trust-request.json `
+  --approval-file .runtime/production-closure/policies/2026-08-12/policy-legal-approval.json `
+  --candidate-keyring-file .runtime/production-closure/policies/2026-08-12/policy-legal-keyring.candidate.json `
+  --policy-dir project-docs/legal-policy-candidates/2026-08-12 `
+  --trust-approver-public-key-file <independent-trust-approver-public.pem> `
+  --checked-at <ISO-8601-with-timezone> `
+  --output .runtime/production-closure/policies/2026-08-12/policy-legal-keyring.trusted.json
+```
+
+`issue-trusted-keyring` 会拒绝过期审批、同一 approver ID、相同公钥、手工翻转 `status`、任一 hash/文件字节漂移和签名错误。输出中的 `trust_basis.policy_activation_authorized` 固定为 `false`；trusted keyring 只建立验签公钥信任，仍不代表政策已经发布或生效。
+
 ## 验签与生产边界
 
-候选 keyring 不能直接作为生产信任。安全或运维负责人应核对公钥来源和审批人身份，将状态从 `candidate_not_trusted` 独立提升为 `trusted`，并保留安装证据。随后运行：
+候选 keyring 不能直接作为生产信任。只有上述 `issue-trusted-keyring` 生成且自带可重验独立签名证据的文件才能用于后续验签。随后运行：
 
 ```powershell
 python yuqei-infra/scripts/prepare-policy-legal-approval.py verify `

yuqei-infra/compose/.env.production.example

@@ -52,6 +52,7 @@ BACKUP_DISK_WARNING_PERCENT=85
 BACKUP_DISK_CRITICAL_PERCENT=90
 BACKUP_DISK_WARNING_FREE_GIB=20
 BACKUP_DISK_CRITICAL_FREE_GIB=10
+PRODUCTION_ALERT_WEBHOOK_URL_FILE=/etc/yuqei/monitoring/alert-webhook-url
 
 # The file is mounted as /run/secrets/ai_provider_api_key. Configure the default
 # database provider with api_key_secret_ref=file:///run/secrets/ai_provider_api_key.

yuqei-infra/deploy/README.md

@@ -17,6 +17,10 @@ The production backend runs on `192.168.3.90`; Contract Web is deployed separate
 3. Mount an off-host backup filesystem at `BACKUP_HOST_DIR`.
 4. Place `tls.crt`, `tls.key`, and the AI provider key in protected host directories. When enabling SMTP, also create the file named by `CONTRACT_API_IDENTITY_SMTP_PASSWORD_FILE_HOST`, mode `0600`.
 5. Copy `.env.production.example` to `/etc/yuqei/production.env`, replace every placeholder, and keep mode `0600`.
+   Create the file named by `PRODUCTION_ALERT_WEBHOOK_URL_FILE`, containing only
+   the approved HTTPS alert receiver URL. Keep it owner-only and readable by the
+   `nobody` account used by the production health unit (for example, owner
+   `nobody:nobody`, mode `0600`).
 6. Run the read-only gate:
 
 ```bash
@@ -71,6 +75,22 @@ api_key_secret_ref=file:///run/secrets/ai_provider_api_key
 
 Do not send the key in the model-config API payload. Run the Provider acceptance gate before any production start approval.
 
+The production Provider canary is read-only. Run the protected
+`provider-promotion-canary.yml` workflow against at least two distinct HTTPS nodes after
+the signed receipt, public keyring, immutable release and image digest are pinned. The
+gate sends `Cache-Control: no-cache, no-store` and fails closed unless every response:
+
+- declares `environment=production` and `verification_mode=offline_read_only`;
+- is fresh within `max_observation_age_seconds` (120 seconds by default);
+- reports consistent top-level, check-detail, and promotion-diagnostic state; and
+- matches the expected release, receipt, signing key and image digest.
+
+Revocation and kill-switch modes apply the same provenance and freshness checks after
+all nodes report the blocking control. The workflow performs GET requests only; it does
+not change a keyring, toggle the kill switch, import a receipt, or invoke any identity
+write/execute API. Make control changes through the independently approved operations
+procedure, then use this workflow only to observe their propagation.
+
 ## Alibaba Cloud frontend variables
 
 The frontend deployment must use server-only values:

yuqei-infra/deploy/preflight-production.py

@@ -28,6 +28,7 @@ PLACEHOLDER_MARKERS = ("change_me", "local-only", "example.com", "your-api-key",
 LOOPBACK_VALUES = {"127.0.0.1", "::1", "localhost"}
 PROVIDER_ACCEPTANCE_RECEIPT_MAX_BYTES = 1024 * 1024
 PROVIDER_ACCEPTANCE_KEYRING_MAX_BYTES = 256 * 1024
+ALERT_WEBHOOK_MAX_BYTES = 16 * 1024
 SMTP_CREDENTIAL_CONTAINER_PATH = "/run/secrets/contract_smtp_password"
 LAUNCH_SCOPE_MANIFEST_PATH = (
     REPO_ROOT / "project-docs" / "V2-production-launch-scope-2026-08-12.json"
@@ -145,6 +146,7 @@ def validate_environment(values: dict[str, str], *, template_mode: bool = False)
         "AI_PLATFORM_PROVIDER_ACCEPTANCE_RECEIPT_FILE_HOST",
         "AI_PLATFORM_PROVIDER_ACCEPTANCE_TRUSTED_KEYS_FILE_HOST",
         "AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256",
+        "PRODUCTION_ALERT_WEBHOOK_URL_FILE",
     )
     missing = [key for key in required if not values.get(key, "").strip()]
     checks.append(_result("required_values", not missing, "All required values are present." if not missing else f"Missing required keys: {', '.join(missing)}."))
@@ -375,6 +377,38 @@ def validate_files(values: dict[str, str], *, template_mode: bool = False) -> li
     require_mount = values.get("BACKUP_REQUIRE_SEPARATE_MOUNT", "true").lower() == "true"
     backup_mount_ok = backup_root.is_dir() and (not require_mount or os.path.ismount(backup_root))
     checks.append(_result("backup_mount", backup_mount_ok, "Backup destination is an available separate mount." if backup_mount_ok else "Backup destination is missing or is not a separate mount."))
+
+    webhook_path = Path(values.get("PRODUCTION_ALERT_WEBHOOK_URL_FILE", ""))
+    webhook_ok = _is_bounded_absolute_file(webhook_path, ALERT_WEBHOOK_MAX_BYTES)
+    webhook_url = ""
+    if webhook_ok:
+        try:
+            webhook_url = webhook_path.read_text(encoding="utf-8").strip()
+        except (OSError, UnicodeError):
+            webhook_ok = False
+    parsed_webhook = urlparse(webhook_url)
+    webhook_ok = webhook_ok and parsed_webhook.scheme == "https" and bool(parsed_webhook.hostname)
+    checks.append(
+        _result(
+            "alert_webhook_file",
+            webhook_ok,
+            "Production alert webhook file is an owner-only HTTPS URL."
+            if webhook_ok
+            else "Production alert webhook file must be an absolute, owner-only file containing an HTTPS URL.",
+        )
+    )
+    if webhook_ok and os.name != "nt":
+        mode = stat.S_IMODE(webhook_path.stat().st_mode)
+        secure_mode = mode & 0o077 == 0
+        checks.append(
+            _result(
+                "alert_webhook_permissions",
+                secure_mode,
+                "Production alert webhook file is owner-only."
+                if secure_mode
+                else f"Alert webhook file mode {oct(mode)} exposes group or other permissions.",
+            )
+        )
     return checks
 
 

yuqei-infra/observability/README.md

@@ -30,7 +30,11 @@ python3 yuqei-infra/observability/check-production-health.py \
   --webhook-url-file /etc/yuqei/secrets/alert-webhook-url
 ```
 
-Install the included systemd service and timer for one-minute checks. The webhook file is optional but, when configured, must contain an HTTPS URL and mode `0600`.
+Install the included systemd service and timer for one-minute checks. Production
+preflight requires `PRODUCTION_ALERT_WEBHOOK_URL_FILE` to point to an absolute,
+owner-only file containing the approved HTTPS receiver URL. The included unit
+passes that file to every one-minute probe; do not put the URL or credentials in
+the repository or directly in the environment file.
 
 The production environment template fixes warning/critical ceilings at 85%/90%
 used and free-space floors at 20/10 GiB for each monitored filesystem. A path
@@ -45,6 +49,13 @@ systemd records the alerting failure. Every run writes
 alert-delivery result, under
 `project-docs/qa-artifacts/v2-production-health/`.
 
+The installed service deduplicates an unchanged failing check set for one hour.
+The first failure, any failed/warning check-set change, and the hourly reminder
+are delivered. A failed delivery is retried on the next one-minute run, and a
+healthy run resets the state so a recurrence is delivered immediately. The
+state is written atomically beside the health artifacts and contains no webhook
+URL or credential.
+
 The production unit uses the host's `/usr/bin/python3.11` and runs as the
 unprivileged `nobody` account so NFS backup manifests remain readable under the
 export's root-squash policy.
@@ -58,4 +69,18 @@ systemd-analyze verify \
   yuqei-infra/observability/yuqei-production-health.timer
 ```
 
+After deployment, alert acceptance must use a temporary command-line threshold
+below current usage and an isolated artifact directory. Do not fill a filesystem
+or edit the installed timer to inject a warning. Retain both the generated
+summary (`attempted=true`, `delivered=true`) and the receiver-side event ID.
+
+`receive-production-alert.py` and `yuqei-production-alert-receiver.service`
+provide a minimal self-hosted receiver when an approved third-party receiver is
+not available. Run it as the dedicated `yuqei-alert` user on a separate node,
+bind it to loopback, and expose only `POST /ops/alerts/<random-token>` through an
+existing HTTPS reverse proxy. Keep the token in owner-restricted files on both
+nodes. Each accepted alert is written as a mode `0600` JSON receipt containing a
+unique event ID, receipt time, source, request-body SHA-256, and the exact health
+payload. Do not expose the receipt directory or add a public read endpoint.
+
 This probe is a minimum launch guard, not a replacement for centralized logs, long-term metrics, PostgreSQL slow-query analysis, and infrastructure dashboards.

yuqei-infra/observability/check-production-health.py

@@ -4,6 +4,7 @@
 from __future__ import annotations
 
 import argparse
+import hashlib
 import json
 import os
 import shutil
@@ -20,6 +21,7 @@ from urllib.request import Request, urlopen
 
 REPO_ROOT = Path(__file__).resolve().parents[2]
 DEFAULT_ARTIFACT_ROOT = REPO_ROOT / "project-docs" / "qa-artifacts" / "v2-production-health"
+DEFAULT_ALERT_RESEND_SECONDS = 3600
 
 
 @dataclass(frozen=True)
@@ -170,7 +172,7 @@ def check_disk(
     )
 
 
-def send_webhook(webhook_file: Path, summary: dict[str, Any], *, timeout: float) -> None:
+def send_webhook(webhook_file: Path, summary: dict[str, Any], *, timeout: float) -> dict[str, str]:
     url = webhook_file.read_text(encoding="utf-8").strip()
     _require_http_url(url, allow_http=False)
     body = json.dumps(summary, ensure_ascii=False).encode("utf-8")
@@ -178,6 +180,75 @@ def send_webhook(webhook_file: Path, summary: dict[str, Any], *, timeout: float)
     with urlopen(request, timeout=timeout) as response:  # nosec B310
         if response.status >= 400:
             raise RuntimeError(f"Alert webhook returned HTTP {response.status}.")
+        content_type = response.headers.get_content_type()
+        if content_type != "application/json":
+            return {}
+        try:
+            receipt = json.loads(response.read().decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise RuntimeError("Alert webhook returned invalid JSON.") from exc
+        if not isinstance(receipt, dict):
+            raise RuntimeError("Alert webhook receipt must be a JSON object.")
+        return {
+            key: value
+            for key in ("event_id", "received_at", "body_sha256")
+            if isinstance((value := receipt.get(key)), str) and value
+        }
+
+
+def alert_fingerprint(summary: dict[str, Any]) -> str:
+    active_state = {
+        "status": summary["status"],
+        "failed_checks": sorted(summary["failed_checks"]),
+        "warning_checks": sorted(summary["warning_checks"]),
+    }
+    encoded = json.dumps(active_state, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def should_deliver_alert(
+    state_file: Path | None,
+    *,
+    fingerprint: str,
+    now: datetime,
+    resend_seconds: int,
+) -> tuple[bool, str]:
+    if state_file is None:
+        return True, "deduplication_not_configured"
+    try:
+        state = json.loads(state_file.read_text(encoding="utf-8"))
+        previous_fingerprint = state.get("fingerprint")
+        delivered_at = datetime.fromisoformat(str(state["delivered_at"]).replace("Z", "+00:00"))
+        if delivered_at.tzinfo is None:
+            raise ValueError("delivery timestamp has no timezone")
+    except (OSError, ValueError, KeyError, TypeError, json.JSONDecodeError):
+        return True, "no_valid_previous_delivery"
+    if previous_fingerprint != fingerprint:
+        return True, "alert_state_changed"
+    elapsed = (now.astimezone(timezone.utc) - delivered_at.astimezone(timezone.utc)).total_seconds()
+    if elapsed >= resend_seconds:
+        return True, "periodic_resend_due"
+    return False, "duplicate_within_resend_window"
+
+
+def write_alert_state(
+    state_file: Path,
+    *,
+    fingerprint: str | None,
+    delivered_at: datetime | None,
+    receipt: dict[str, str] | None = None,
+) -> None:
+    state_file.parent.mkdir(parents=True, exist_ok=True)
+    state = {
+        "schema_version": "yuqei-production-alert-state.v1",
+        "fingerprint": fingerprint,
+        "delivered_at": delivered_at.isoformat() if delivered_at else None,
+        "receiver_event_id": (receipt or {}).get("event_id"),
+    }
+    temporary = state_file.with_name(f".{state_file.name}.{os.getpid()}.tmp")
+    temporary.write_text(json.dumps(state, sort_keys=True) + "\n", encoding="utf-8")
+    os.chmod(temporary, 0o600)
+    temporary.replace(state_file)
 
 
 def read_secret_file(path: Path | None) -> str | None:
@@ -243,12 +314,17 @@ def _parse_args() -> argparse.Namespace:
     )
     parser.add_argument("--timeout-seconds", type=float, default=10.0)
     parser.add_argument("--webhook-url-file", type=Path)
+    parser.add_argument("--alert-state-file", type=Path)
+    parser.add_argument("--alert-resend-seconds", type=int, default=DEFAULT_ALERT_RESEND_SECONDS)
     parser.add_argument("--artifact-dir", type=Path)
     return parser.parse_args()
 
 
 def main() -> int:
     args = _parse_args()
+    if not 60 <= args.alert_resend_seconds <= 86400:
+        print("PRODUCTION_HEALTH_FAIL invalid_alert_resend_seconds")
+        return 2
     capacity_policies = [
         {
             "key": "disk_root",
@@ -344,6 +420,11 @@ def main() -> int:
             "attempted": False,
             "delivered": False,
             "error": None,
+            "receiver_event_id": None,
+            "receiver_received_at": None,
+            "receiver_body_sha256": None,
+            "suppressed": False,
+            "decision": None,
         },
         "checks": [asdict(check) for check in checks],
     }
@@ -352,15 +433,48 @@ def main() -> int:
     summary_path = artifact_dir / "production-health-summary.json"
     alert_delivery_failed = False
     if (failed or warnings) and args.webhook_url_file:
-        summary["alert_delivery"]["attempted"] = True
-        try:
-            send_webhook(args.webhook_url_file, summary, timeout=args.timeout_seconds)
-        except Exception as exc:
-            alert_delivery_failed = True
-            summary["alert_delivery"]["error"] = type(exc).__name__
-            print(f"PRODUCTION_HEALTH_WEBHOOK_FAIL error={type(exc).__name__}", file=sys.stderr)
+        fingerprint = alert_fingerprint(summary)
+        deliver, decision = should_deliver_alert(
+            args.alert_state_file,
+            fingerprint=fingerprint,
+            now=checked_at,
+            resend_seconds=args.alert_resend_seconds,
+        )
+        summary["alert_delivery"]["decision"] = decision
+        if deliver:
+            summary["alert_delivery"]["attempted"] = True
+            try:
+                receipt = send_webhook(args.webhook_url_file, summary, timeout=args.timeout_seconds) or {}
+            except Exception as exc:
+                alert_delivery_failed = True
+                summary["alert_delivery"]["error"] = type(exc).__name__
+                print(f"PRODUCTION_HEALTH_WEBHOOK_FAIL error={type(exc).__name__}", file=sys.stderr)
+            else:
+                summary["alert_delivery"]["delivered"] = True
+                summary["alert_delivery"]["receiver_event_id"] = receipt.get("event_id")
+                summary["alert_delivery"]["receiver_received_at"] = receipt.get("received_at")
+                summary["alert_delivery"]["receiver_body_sha256"] = receipt.get("body_sha256")
+                if args.alert_state_file:
+                    try:
+                        write_alert_state(
+                            args.alert_state_file,
+                            fingerprint=fingerprint,
+                            delivered_at=checked_at,
+                            receipt=receipt,
+                        )
+                    except OSError:
+                        alert_delivery_failed = True
+                        summary["alert_delivery"]["error"] = "StateWriteError"
+                        print("PRODUCTION_HEALTH_ALERT_STATE_FAIL", file=sys.stderr)
         else:
-            summary["alert_delivery"]["delivered"] = True
+            summary["alert_delivery"]["suppressed"] = True
+    elif not (failed or warnings) and args.alert_state_file:
+        try:
+            write_alert_state(args.alert_state_file, fingerprint=None, delivered_at=None)
+        except OSError:
+            alert_delivery_failed = True
+            summary["alert_delivery"]["error"] = "StateWriteError"
+            print("PRODUCTION_HEALTH_ALERT_STATE_FAIL", file=sys.stderr)
     summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     print(
         f"PRODUCTION_HEALTH_{summary['status'].upper()} "

yuqei-infra/observability/nginx-production-alert-location.conf

@@ -0,0 +1,14 @@
+location ^~ /ops/alerts/ {
+    client_max_body_size 1m;
+    proxy_http_version 1.1;
+    proxy_set_header Host $host;
+    proxy_set_header X-Forwarded-For $remote_addr;
+    proxy_set_header X-Forwarded-Proto https;
+    proxy_connect_timeout 3s;
+    proxy_read_timeout 10s;
+    proxy_pass http://127.0.0.1:19091;
+
+    limit_except POST {
+        deny all;
+    }
+}

yuqei-infra/observability/receive-production-alert.py

@@ -0,0 +1,207 @@
+#!/usr/bin/env python3
+"""Receive authenticated production health alerts and persist immutable receipts."""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import hmac
+import json
+import os
+import re
+import sys
+import uuid
+from datetime import datetime, timezone
+from http import HTTPStatus
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from pathlib import Path
+from typing import Any
+from urllib.parse import unquote, urlparse
+
+
+TOKEN_PATTERN = re.compile(r"^[A-Za-z0-9_-]{43,128}$")
+DEFAULT_MAX_BODY_BYTES = 1024 * 1024
+
+
+def read_token(path: Path) -> str:
+    token = path.read_text(encoding="utf-8").strip()
+    if not TOKEN_PATTERN.fullmatch(token):
+        raise ValueError("alert receiver token must be 43-128 URL-safe characters")
+    return token
+
+
+def parse_alert_body(raw: bytes) -> dict[str, Any]:
+    try:
+        payload = json.loads(raw.decode("utf-8"))
+    except (UnicodeDecodeError, json.JSONDecodeError) as exc:
+        raise ValueError("alert body must be valid UTF-8 JSON") from exc
+    if not isinstance(payload, dict):
+        raise ValueError("alert body must be a JSON object")
+    if payload.get("status") not in {"warning", "fail"}:
+        raise ValueError("alert status must be warning or fail")
+    if not isinstance(payload.get("checked_at"), str):
+        raise ValueError("alert checked_at is required")
+    return payload
+
+
+def persist_event(
+    event_dir: Path,
+    *,
+    payload: dict[str, Any],
+    raw: bytes,
+    source: str,
+) -> dict[str, str]:
+    received_at = datetime.now(timezone.utc).isoformat()
+    event_id = f"yuqei-alert-{uuid.uuid4().hex}"
+    receipt = {
+        "schema_version": "yuqei-production-alert-receipt.v1",
+        "event_id": event_id,
+        "received_at": received_at,
+        "source": source,
+        "body_sha256": hashlib.sha256(raw).hexdigest(),
+        "payload": payload,
+    }
+    event_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+    target = event_dir / f"{event_id}.json"
+    encoded = (json.dumps(receipt, ensure_ascii=False, indent=2) + "\n").encode("utf-8")
+    descriptor = os.open(target, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    try:
+        with os.fdopen(descriptor, "wb") as stream:
+            stream.write(encoded)
+            stream.flush()
+            os.fsync(stream.fileno())
+    except Exception:
+        target.unlink(missing_ok=True)
+        raise
+    return {
+        "event_id": event_id,
+        "received_at": received_at,
+        "body_sha256": receipt["body_sha256"],
+    }
+
+
+class AlertReceiver(ThreadingHTTPServer):
+    daemon_threads = True
+
+    def __init__(
+        self,
+        server_address: tuple[str, int],
+        *,
+        token: str,
+        event_dir: Path,
+        max_body_bytes: int,
+    ) -> None:
+        super().__init__(server_address, AlertHandler)
+        self.expected_path = f"/ops/alerts/{token}"
+        self.event_dir = event_dir
+        self.max_body_bytes = max_body_bytes
+
+
+class AlertHandler(BaseHTTPRequestHandler):
+    server: AlertReceiver
+
+    def do_POST(self) -> None:  # noqa: N802
+        request_path = unquote(urlparse(self.path).path)
+        if not hmac.compare_digest(request_path, self.server.expected_path):
+            self._respond(HTTPStatus.NOT_FOUND, {"error": "not_found"})
+            return
+        content_type = self.headers.get_content_type()
+        if content_type != "application/json":
+            self._respond(HTTPStatus.UNSUPPORTED_MEDIA_TYPE, {"error": "application_json_required"})
+            return
+        try:
+            content_length = int(self.headers.get("Content-Length", ""))
+        except ValueError:
+            content_length = -1
+        if not 0 < content_length <= self.server.max_body_bytes:
+            self._respond(HTTPStatus.REQUEST_ENTITY_TOO_LARGE, {"error": "invalid_content_length"})
+            return
+        raw = self.rfile.read(content_length)
+        if len(raw) != content_length:
+            self._respond(HTTPStatus.BAD_REQUEST, {"error": "incomplete_body"})
+            return
+        try:
+            payload = parse_alert_body(raw)
+            receipt = persist_event(
+                self.server.event_dir,
+                payload=payload,
+                raw=raw,
+                source=self.headers.get("X-Forwarded-For", self.client_address[0]),
+            )
+        except (OSError, ValueError):
+            self._respond(HTTPStatus.BAD_REQUEST, {"error": "invalid_alert"})
+            return
+        self._respond(HTTPStatus.CREATED, receipt)
+
+    def do_GET(self) -> None:  # noqa: N802
+        self._respond(HTTPStatus.METHOD_NOT_ALLOWED, {"error": "method_not_allowed"})
+
+    def log_message(self, message: str, *args: object) -> None:
+        print(f"ALERT_RECEIVER {self.address_string()} {message % args}", file=sys.stderr)
+
+    def _respond(self, status: HTTPStatus, payload: dict[str, str]) -> None:
+        raw = json.dumps(payload, separators=(",", ":")).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json")
+        self.send_header("Content-Length", str(len(raw)))
+        self.send_header("Cache-Control", "no-store")
+        self.send_header("X-Content-Type-Options", "nosniff")
+        self.end_headers()
+        self.wfile.write(raw)
+
+
+def build_server(
+    bind: str,
+    port: int,
+    *,
+    token_file: Path,
+    event_dir: Path,
+    max_body_bytes: int,
+) -> AlertReceiver:
+    if not 0 <= port <= 65535:
+        raise ValueError("port must be between 0 and 65535")
+    if not 1024 <= max_body_bytes <= 4 * 1024 * 1024:
+        raise ValueError("maximum body size must be between 1 KiB and 4 MiB")
+    return AlertReceiver(
+        (bind, port),
+        token=read_token(token_file),
+        event_dir=event_dir.resolve(),
+        max_body_bytes=max_body_bytes,
+    )
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--bind", default="127.0.0.1")
+    parser.add_argument("--port", type=int, default=19091)
+    parser.add_argument("--token-file", type=Path, required=True)
+    parser.add_argument("--event-dir", type=Path, required=True)
+    parser.add_argument("--max-body-bytes", type=int, default=DEFAULT_MAX_BODY_BYTES)
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        server = build_server(
+            args.bind,
+            args.port,
+            token_file=args.token_file,
+            event_dir=args.event_dir,
+            max_body_bytes=args.max_body_bytes,
+        )
+    except (OSError, ValueError) as exc:
+        print(f"ALERT_RECEIVER_FAIL error={type(exc).__name__}", file=sys.stderr)
+        return 2
+    print(f"ALERT_RECEIVER_READY bind={args.bind} port={args.port}", flush=True)
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        server.server_close()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

yuqei-infra/observability/yuqei-production-alert-receiver.service

@@ -0,0 +1,28 @@
+[Unit]
+Description=Yuqei production alert receipt receiver
+After=network.target
+
+[Service]
+Type=simple
+User=yuqei-alert
+Group=yuqei-alert
+WorkingDirectory=/opt/yuqei-alert-receiver
+ExecStart=/usr/bin/python3 /opt/yuqei-alert-receiver/receive-production-alert.py --bind 127.0.0.1 --port 19091 --token-file /etc/yuqei/alert-receiver/token --event-dir /var/lib/yuqei-alert-receiver/events
+Restart=on-failure
+RestartSec=5
+UMask=0077
+NoNewPrivileges=true
+PrivateTmp=true
+ProtectSystem=strict
+ProtectHome=true
+ProtectKernelTunables=true
+ProtectKernelModules=true
+ProtectControlGroups=true
+RestrictSUIDSGID=true
+LockPersonality=true
+MemoryDenyWriteExecute=true
+ReadOnlyPaths=/etc/yuqei/alert-receiver/token
+ReadWritePaths=/var/lib/yuqei-alert-receiver/events
+
+[Install]
+WantedBy=multi-user.target

yuqei-infra/observability/yuqei-production-health.service

@@ -8,7 +8,7 @@ Type=oneshot
 User=nobody
 Group=nobody
 WorkingDirectory=/opt/apps/ai-contract-platform-v2-production
-ExecStart=/usr/bin/python3.11 yuqei-infra/observability/check-production-health.py --edge-base-url ${BACKEND_PUBLIC_BASE_URL} --contract-api-base http://127.0.0.1:${CONTRACT_API_HOST_PORT} --contract-proxy-key-file /etc/yuqei/monitoring/contract-proxy-key --ai-api-base http://127.0.0.1:${AI_PLATFORM_API_HOST_PORT} --ai-service-key-file /etc/yuqei/monitoring/ai-platform-service-key --backup-root ${BACKUP_HOST_DIR} --backup-max-age-hours ${BACKUP_MAX_AGE_HOURS} --root-disk-path / --root-disk-warning-percent ${ROOT_DISK_WARNING_PERCENT} --root-disk-critical-percent ${ROOT_DISK_CRITICAL_PERCENT} --root-disk-warning-free-gib ${ROOT_DISK_WARNING_FREE_GIB} --root-disk-critical-free-gib ${ROOT_DISK_CRITICAL_FREE_GIB} --home-disk-path /home --home-disk-warning-percent ${HOME_DISK_WARNING_PERCENT} --home-disk-critical-percent ${HOME_DISK_CRITICAL_PERCENT} --home-disk-warning-free-gib ${HOME_DISK_WARNING_FREE_GIB} --home-disk-critical-free-gib ${HOME_DISK_CRITICAL_FREE_GIB} --home-require-separate-mount --backup-disk-path ${BACKUP_HOST_DIR} --backup-disk-warning-percent ${BACKUP_DISK_WARNING_PERCENT} --backup-disk-critical-percent ${BACKUP_DISK_CRITICAL_PERCENT} --backup-disk-warning-free-gib ${BACKUP_DISK_WARNING_FREE_GIB} --backup-require-separate-mount --timeout-seconds 30
+ExecStart=/usr/bin/python3.11 yuqei-infra/observability/check-production-health.py --edge-base-url ${BACKEND_PUBLIC_BASE_URL} --contract-api-base http://127.0.0.1:${CONTRACT_API_HOST_PORT} --contract-proxy-key-file /etc/yuqei/monitoring/contract-proxy-key --ai-api-base http://127.0.0.1:${AI_PLATFORM_API_HOST_PORT} --ai-service-key-file /etc/yuqei/monitoring/ai-platform-service-key --backup-root ${BACKUP_HOST_DIR} --backup-max-age-hours ${BACKUP_MAX_AGE_HOURS} --root-disk-path / --root-disk-warning-percent ${ROOT_DISK_WARNING_PERCENT} --root-disk-critical-percent ${ROOT_DISK_CRITICAL_PERCENT} --root-disk-warning-free-gib ${ROOT_DISK_WARNING_FREE_GIB} --root-disk-critical-free-gib ${ROOT_DISK_CRITICAL_FREE_GIB} --home-disk-path /home --home-disk-warning-percent ${HOME_DISK_WARNING_PERCENT} --home-disk-critical-percent ${HOME_DISK_CRITICAL_PERCENT} --home-disk-warning-free-gib ${HOME_DISK_WARNING_FREE_GIB} --home-disk-critical-free-gib ${HOME_DISK_CRITICAL_FREE_GIB} --home-require-separate-mount --backup-disk-path ${BACKUP_HOST_DIR} --backup-disk-warning-percent ${BACKUP_DISK_WARNING_PERCENT} --backup-disk-critical-percent ${BACKUP_DISK_CRITICAL_PERCENT} --backup-disk-warning-free-gib ${BACKUP_DISK_WARNING_FREE_GIB} --backup-disk-critical-free-gib ${BACKUP_DISK_CRITICAL_FREE_GIB} --backup-require-separate-mount --webhook-url-file ${PRODUCTION_ALERT_WEBHOOK_URL_FILE} --alert-state-file /opt/apps/ai-contract-platform-v2-production/project-docs/qa-artifacts/v2-production-health/alert-state.json --alert-resend-seconds 3600 --timeout-seconds 30
 EnvironmentFile=/etc/yuqei/production.env
 NoNewPrivileges=true
 PrivateTmp=true

yuqei-infra/scripts/prepare-policy-legal-approval.py

@@ -1,5 +1,5 @@
 #!/usr/bin/env python3
-"""Prepare, finalize, sign, and verify named legal approval for public policies."""
+"""Prepare, sign, and verify named legal approval and trust requests for public policies."""
 
 from __future__ import annotations
 
@@ -23,8 +23,18 @@ PENDING_PAYLOAD_SCHEMA = "yuqei-public-policy-approval-pending-v1"
 APPROVED_PAYLOAD_SCHEMA = "yuqei-public-policy-approval-payload-v1"
 APPROVAL_SCHEMA = "yuqei-public-policy-legal-approval-v1"
 KEYRING_SCHEMA = "yuqei-public-policy-legal-keyring-v1"
+TRUST_REQUEST_SCHEMA = "yuqei-public-policy-key-trust-request-v1"
+TRUST_APPROVAL_PAYLOAD_SCHEMA = "yuqei-public-policy-key-trust-approval-payload-v1"
+TRUST_APPROVAL_SCHEMA = "yuqei-public-policy-key-trust-approval-v1"
 SIGNATURE_DOMAIN = b"YUQEI_PUBLIC_POLICY_LEGAL_APPROVAL_V1\0"
+TRUST_SIGNATURE_DOMAIN = b"YUQEI_PUBLIC_POLICY_KEY_TRUST_APPROVAL_V1\0"
 SIGN_CONFIRMATION = "I_APPROVE_THIS_EXACT_POLICY_PAYLOAD"
+TRUST_SIGN_CONFIRMATION = "I_APPROVE_THIS_EXACT_POLICY_KEY_TRUST_PAYLOAD"
+TRUST_REVIEW_ATTESTATIONS = [
+    "verified_legal_approver_identity_and_authority_out_of_band",
+    "verified_legal_signing_key_fingerprint_out_of_band",
+    "accepts_independent_security_or_operations_review_responsibility",
+]
 EXPECTED_POLICY_IDS = {
     "user_terms",
     "privacy_policy",
@@ -409,9 +419,9 @@ def verify_approval(
 
     if keyring.get("schema_version") != KEYRING_SCHEMA:
         raise PolicyApprovalError("policy keyring schema_version is invalid")
-    if keyring.get("status") != "trusted" and not (
-        allow_candidate_keyring and keyring.get("status") == "candidate_not_trusted"
-    ):
+    if keyring.get("status") == "trusted":
+        verify_trusted_keyring_basis(keyring, approval=approval)
+    elif not (allow_candidate_keyring and keyring.get("status") == "candidate_not_trusted"):
         raise PolicyApprovalError("policy keyring is not trusted")
     keys = keyring.get("keys")
     if not isinstance(keys, list):
@@ -459,6 +469,436 @@ def verify_approval(
     }
 
 
+def verify_trusted_keyring_basis(keyring: dict[str, Any], *, approval: dict[str, Any]) -> None:
+    basis = keyring.get("trust_basis")
+    if not isinstance(basis, dict):
+        raise PolicyApprovalError("trusted policy keyring lacks an independently signed trust basis")
+    if basis.get("policy_activation_authorized") is not False:
+        raise PolicyApprovalError("trusted policy keyring must not authorize policy activation")
+    trust_approval = basis.get("trust_approval")
+    if not isinstance(trust_approval, dict) or trust_approval.get("schema_version") != TRUST_APPROVAL_SCHEMA:
+        raise PolicyApprovalError("trusted policy keyring trust approval is invalid")
+    payload = trust_approval.get("payload")
+    if not isinstance(payload, dict) or payload.get("schema_version") != TRUST_APPROVAL_PAYLOAD_SCHEMA:
+        raise PolicyApprovalError("trusted policy keyring trust payload is invalid")
+    decision = payload.get("decision")
+    trust_approver = payload.get("trust_approver")
+    legal_approver = payload.get("legal_approver")
+    if not isinstance(decision, dict) or not isinstance(trust_approver, dict) or not isinstance(legal_approver, dict):
+        raise PolicyApprovalError("trusted policy keyring trust bindings are invalid")
+    if decision.get("decision") != "approve_policy_signing_key_trust":
+        raise PolicyApprovalError("trusted policy keyring trust decision is invalid")
+    if decision.get("policy_activation_authorized") is not False:
+        raise PolicyApprovalError("trusted policy keyring trust decision must not authorize policy activation")
+    if decision.get("responsibility") not in {"security", "operations"}:
+        raise PolicyApprovalError("trusted policy keyring trust responsibility is invalid")
+    if decision.get("attestations") != TRUST_REVIEW_ATTESTATIONS:
+        raise PolicyApprovalError("trusted policy keyring trust attestations are incomplete")
+    if payload.get("policy_approval_id") != approval.get("approval_id") or payload.get(
+        "policy_approval_sha256"
+    ) != sha256_bytes(canonical_json(approval)):
+        raise PolicyApprovalError("trusted policy keyring does not authorize this exact policy approval")
+    trusted_at = parse_timestamp(_required_text(basis, "trusted_at"), label="trusted_at")
+    approved = parse_timestamp(_required_text(decision, "approved_at"), label="approved_at")
+    expires = parse_timestamp(_required_text(decision, "expires_at"), label="expires_at")
+    if trusted_at < approved or trusted_at > expires:
+        raise PolicyApprovalError("trusted policy keyring was issued outside its approval window")
+    candidate = {key: value for key, value in keyring.items() if key != "trust_basis"}
+    candidate["status"] = "candidate_not_trusted"
+    candidate_sha256 = sha256_bytes(canonical_json(candidate))
+    if payload.get("candidate_keyring_sha256") != candidate_sha256:
+        raise PolicyApprovalError("trusted policy keyring candidate binding does not match")
+    if basis.get("trust_request_sha256") != payload.get("trust_request_sha256"):
+        raise PolicyApprovalError("trusted policy keyring trust request binding does not match")
+    if basis.get("trust_approval_sha256") != sha256_bytes(canonical_json(trust_approval)):
+        raise PolicyApprovalError("trusted policy keyring trust approval hash does not match")
+    keys = candidate.get("keys")
+    if not isinstance(keys, list) or len(keys) != 1 or not isinstance(keys[0], dict):
+        raise PolicyApprovalError("trusted policy keyring candidate key set is invalid")
+    legal_key = keys[0]
+    if trust_approver.get("approver_id") == legal_key.get("approver_id"):
+        raise PolicyApprovalError("trusted policy keyring approvers are not independent")
+    if trust_approver.get("public_key_raw_sha256") == legal_key.get("public_key_raw_sha256"):
+        raise PolicyApprovalError("trusted policy keyring approval keys are not independent")
+    public_pem = basis.get("trust_approver_public_key_pem")
+    if not isinstance(public_pem, str) or not public_pem.strip():
+        raise PolicyApprovalError("trusted policy keyring trust approver public key is invalid")
+    try:
+        public_key = serialization.load_pem_public_key(public_pem.encode("ascii"))
+    except (UnicodeError, ValueError) as exc:
+        raise PolicyApprovalError("trusted policy keyring trust approver public key is invalid") from exc
+    if not isinstance(public_key, Ed25519PublicKey):
+        raise PolicyApprovalError("trusted policy keyring trust approver key must be Ed25519")
+    public_raw = public_key.public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    if sha256_bytes(public_pem.encode("ascii")) != trust_approver.get("public_key_file_sha256") or sha256_bytes(
+        public_raw
+    ) != trust_approver.get("public_key_raw_sha256"):
+        raise PolicyApprovalError("trusted policy keyring trust approver key binding does not match")
+    try:
+        signature = base64.b64decode(_required_text(trust_approval, "signature"), validate=True)
+        public_key.verify(signature, TRUST_SIGNATURE_DOMAIN + canonical_json(payload))
+    except Exception as exc:
+        raise PolicyApprovalError("trusted policy keyring trust approval signature verification failed") from exc
+    expected_id = f"policy-key-trust-approval-{sha256_bytes(canonical_json(payload) + signature)[:24]}"
+    if trust_approval.get("trust_approval_id") != expected_id or basis.get("trust_approval_id") != expected_id:
+        raise PolicyApprovalError("trusted policy keyring trust approval_id does not match")
+
+
+def build_trust_request(
+    approval: dict[str, Any],
+    *,
+    candidate_keyring: dict[str, Any],
+    policy_dir: Path,
+    requested_at: str,
+) -> dict[str, Any]:
+    if candidate_keyring.get("schema_version") != KEYRING_SCHEMA:
+        raise PolicyApprovalError("candidate policy keyring schema_version is invalid")
+    if candidate_keyring.get("status") != "candidate_not_trusted":
+        raise PolicyApprovalError("trust request requires a candidate_not_trusted keyring")
+    requested_at = normalized_timestamp(requested_at, label="requested_at")
+    verified = verify_approval(
+        approval,
+        keyring=candidate_keyring,
+        policy_dir=policy_dir,
+        checked_at=requested_at,
+        allow_candidate_keyring=True,
+    )
+    payload = approval["payload"]
+    request = payload["approval_request"]
+    decision = payload["decision"]
+    return {
+        "schema_version": TRUST_REQUEST_SCHEMA,
+        "status": "pending_independent_trust_approval",
+        "requested_at": requested_at,
+        "activation_authorized": False,
+        "approval_id": verified["approval_id"],
+        "approval_sha256": sha256_bytes(canonical_json(approval)),
+        "candidate_keyring_sha256": sha256_bytes(canonical_json(candidate_keyring)),
+        "policy_version": verified["policy_version"],
+        "effective_at": verified["effective_at"],
+        "approval_expires_at": normalized_timestamp(
+            _required_text(decision, "expires_at"),
+            label="expires_at",
+        ),
+        "documents": verified["documents"],
+        "key": {
+            field: request[field]
+            for field in (
+                "key_id",
+                "algorithm",
+                "approver_id",
+                "reviewer",
+                "role",
+                "organization",
+                "public_key_file_sha256",
+                "public_key_raw_sha256",
+            )
+        },
+        "required_independent_actions": [
+            "verify_approver_identity_and_authority_out_of_band",
+            "verify_public_key_fingerprint_out_of_band",
+            "record_named_security_or_operations_approval",
+            "install_a_separately_approved_trusted_keyring",
+            "rerun_policy_approval_verification_with_candidate_override_disabled",
+        ],
+        "prohibited_automatic_actions": [
+            "mark_candidate_keyring_trusted",
+            "publish_policy_documents",
+            "change_policy_lifecycle_to_effective",
+            "start_edge",
+            "change_public_traffic",
+        ],
+    }
+
+
+def validate_trust_request(
+    trust_request: dict[str, Any],
+    *,
+    approval: dict[str, Any],
+    candidate_keyring: dict[str, Any],
+    policy_dir: Path,
+    checked_at: str,
+) -> dict[str, Any]:
+    if trust_request.get("schema_version") != TRUST_REQUEST_SCHEMA:
+        raise PolicyApprovalError("policy trust request schema_version is invalid")
+    if trust_request.get("status") != "pending_independent_trust_approval":
+        raise PolicyApprovalError("policy trust request is not pending independent approval")
+    if trust_request.get("activation_authorized") is not False:
+        raise PolicyApprovalError("policy trust request must not authorize activation")
+    requested_at = _required_text(trust_request, "requested_at")
+    expected = build_trust_request(
+        approval,
+        candidate_keyring=candidate_keyring,
+        policy_dir=policy_dir,
+        requested_at=requested_at,
+    )
+    if canonical_json(trust_request) != canonical_json(expected):
+        raise PolicyApprovalError("policy trust request does not match its signed source material")
+    checked = parse_timestamp(checked_at, label="checked_at")
+    if checked < parse_timestamp(requested_at, label="requested_at"):
+        raise PolicyApprovalError("policy trust request cannot be checked before it was requested")
+    verify_approval(
+        approval,
+        keyring=candidate_keyring,
+        policy_dir=policy_dir,
+        checked_at=checked.isoformat(),
+        allow_candidate_keyring=True,
+    )
+    return expected
+
+
+def build_trust_approval_payload(
+    trust_request: dict[str, Any],
+    *,
+    approval: dict[str, Any],
+    candidate_keyring: dict[str, Any],
+    policy_dir: Path,
+    trust_approver_public_key_file: Path,
+    key_id: str,
+    approver_id: str,
+    reviewer: str,
+    role: str,
+    organization: str,
+    responsibility: str,
+    approved_at: str,
+    expires_at: str,
+    key_not_before: str,
+    key_not_after: str,
+) -> dict[str, Any]:
+    if not KEY_ID_PATTERN.fullmatch(key_id):
+        raise PolicyApprovalError("trust approver key_id format is invalid")
+    approved_at = normalized_timestamp(approved_at, label="approved_at")
+    expires_at = normalized_timestamp(expires_at, label="expires_at")
+    key_not_before = normalized_timestamp(key_not_before, label="key_not_before")
+    key_not_after = normalized_timestamp(key_not_after, label="key_not_after")
+    approved = parse_timestamp(approved_at, label="approved_at")
+    expires = parse_timestamp(expires_at, label="expires_at")
+    if expires <= approved or expires - approved > MAX_APPROVAL_WINDOW:
+        raise PolicyApprovalError("trust approval validity must be greater than zero and no longer than seven days")
+    if approved < parse_timestamp(key_not_before, label="key_not_before") or expires > parse_timestamp(
+        key_not_after, label="key_not_after"
+    ):
+        raise PolicyApprovalError("trust approval window must remain within the trust key validity window")
+    request = validate_trust_request(
+        trust_request,
+        approval=approval,
+        candidate_keyring=candidate_keyring,
+        policy_dir=policy_dir,
+        checked_at=approved_at,
+    )
+    legal_expiry = parse_timestamp(_required_text(request, "approval_expires_at"), label="approval_expires_at")
+    if expires > legal_expiry:
+        raise PolicyApprovalError("trust approval cannot outlive the legal policy approval")
+    _, public_pem, public_raw = load_public_key(trust_approver_public_key_file)
+    identity = {
+        "approver_id": _nonempty(approver_id, "trust approver_id"),
+        "reviewer": _nonempty(reviewer, "trust reviewer"),
+        "role": _nonempty(role, "trust role"),
+        "organization": _nonempty(organization, "trust organization"),
+    }
+    responsibility = _nonempty(responsibility, "trust responsibility").lower()
+    if responsibility not in {"security", "operations"}:
+        raise PolicyApprovalError("trust responsibility must be security or operations")
+    legal_key = request.get("key")
+    if not isinstance(legal_key, dict):
+        raise PolicyApprovalError("policy trust request legal key binding is invalid")
+    public_raw_sha256 = sha256_bytes(public_raw)
+    if identity["approver_id"] == legal_key.get("approver_id"):
+        raise PolicyApprovalError("trust approver must differ from the legal approver")
+    if public_raw_sha256 == legal_key.get("public_key_raw_sha256"):
+        raise PolicyApprovalError("trust approver key must differ from the legal approval key")
+    return {
+        "schema_version": TRUST_APPROVAL_PAYLOAD_SCHEMA,
+        "status": "pending_independent_trust_signature",
+        "trust_request_sha256": sha256_bytes(canonical_json(request)),
+        "policy_approval_id": request["approval_id"],
+        "policy_approval_sha256": request["approval_sha256"],
+        "candidate_keyring_sha256": request["candidate_keyring_sha256"],
+        "policy_version": request["policy_version"],
+        "effective_at": request["effective_at"],
+        "documents": request["documents"],
+        "legal_approver": {
+            field: legal_key[field]
+            for field in ("approver_id", "reviewer", "role", "organization", "key_id", "public_key_raw_sha256")
+        },
+        "trust_approver": {
+            **identity,
+            "key_id": key_id,
+            "algorithm": "Ed25519",
+            "public_key_file_sha256": sha256_bytes(public_pem),
+            "public_key_raw_sha256": public_raw_sha256,
+            "key_not_before": key_not_before,
+            "key_not_after": key_not_after,
+        },
+        "decision": {
+            "decision": "approve_policy_signing_key_trust",
+            "responsibility": responsibility,
+            "attestations": TRUST_REVIEW_ATTESTATIONS,
+            "approved_at": approved_at,
+            "expires_at": expires_at,
+            "confirmation": TRUST_SIGN_CONFIRMATION,
+            "policy_activation_authorized": False,
+        },
+    }
+
+
+def sign_trust_approval_payload(
+    payload: dict[str, Any],
+    *,
+    private_key_file: Path,
+    password_file: Path,
+    confirmation: str,
+) -> dict[str, Any]:
+    if confirmation != TRUST_SIGN_CONFIRMATION:
+        raise PolicyApprovalError(f"trust signing requires confirmation {TRUST_SIGN_CONFIRMATION}")
+    if payload.get("schema_version") != TRUST_APPROVAL_PAYLOAD_SCHEMA:
+        raise PolicyApprovalError("trust approval payload schema_version is invalid")
+    if payload.get("status") != "pending_independent_trust_signature":
+        raise PolicyApprovalError("only a pending trust approval payload can be signed")
+    decision = payload.get("decision")
+    if not isinstance(decision, dict) or decision.get("policy_activation_authorized") is not False:
+        raise PolicyApprovalError("trust approval payload must not authorize policy activation")
+    private_raw = read_bounded(private_key_file, label="trust approver private key", max_bytes=64 * 1024)
+    password = read_bounded(password_file, label="trust approver private key password", max_bytes=4096).strip()
+    if not password:
+        raise PolicyApprovalError("trust approver private key password is empty")
+    try:
+        key = serialization.load_pem_private_key(private_raw, password=password)
+    except (TypeError, ValueError) as exc:
+        raise PolicyApprovalError("trust approver private key or password is invalid") from exc
+    if not isinstance(key, Ed25519PrivateKey):
+        raise PolicyApprovalError("trust approver private key must be Ed25519")
+    public_raw = key.public_key().public_bytes(serialization.Encoding.Raw, serialization.PublicFormat.Raw)
+    trust_approver = payload.get("trust_approver")
+    if not isinstance(trust_approver, dict) or sha256_bytes(public_raw) != trust_approver.get(
+        "public_key_raw_sha256"
+    ):
+        raise PolicyApprovalError("trust approver private key does not match the payload binding")
+    payload_raw = canonical_json(payload)
+    signature = key.sign(TRUST_SIGNATURE_DOMAIN + payload_raw)
+    approval_id = f"policy-key-trust-approval-{sha256_bytes(payload_raw + signature)[:24]}"
+    return {
+        "schema_version": TRUST_APPROVAL_SCHEMA,
+        "trust_approval_id": approval_id,
+        "payload": payload,
+        "signature": base64.b64encode(signature).decode("ascii"),
+    }
+
+
+def issue_trusted_keyring(
+    trust_approval: dict[str, Any],
+    *,
+    trust_request: dict[str, Any],
+    approval: dict[str, Any],
+    candidate_keyring: dict[str, Any],
+    policy_dir: Path,
+    trust_approver_public_key_file: Path,
+    checked_at: str,
+) -> dict[str, Any]:
+    if trust_approval.get("schema_version") != TRUST_APPROVAL_SCHEMA:
+        raise PolicyApprovalError("policy key trust approval schema_version is invalid")
+    payload = trust_approval.get("payload")
+    if not isinstance(payload, dict) or payload.get("schema_version") != TRUST_APPROVAL_PAYLOAD_SCHEMA:
+        raise PolicyApprovalError("signed policy key trust approval payload is invalid")
+    checked_at = normalized_timestamp(checked_at, label="checked_at")
+    request = validate_trust_request(
+        trust_request,
+        approval=approval,
+        candidate_keyring=candidate_keyring,
+        policy_dir=policy_dir,
+        checked_at=checked_at,
+    )
+    expected_bindings = {
+        "trust_request_sha256": sha256_bytes(canonical_json(request)),
+        "policy_approval_id": request["approval_id"],
+        "policy_approval_sha256": request["approval_sha256"],
+        "candidate_keyring_sha256": request["candidate_keyring_sha256"],
+        "policy_version": request["policy_version"],
+        "effective_at": request["effective_at"],
+        "documents": request["documents"],
+    }
+    for field, expected in expected_bindings.items():
+        if payload.get(field) != expected:
+            raise PolicyApprovalError(f"trust approval binding mismatch: {field}")
+    decision = payload.get("decision")
+    if not isinstance(decision, dict) or decision.get("decision") != "approve_policy_signing_key_trust":
+        raise PolicyApprovalError("policy key trust decision is invalid")
+    if decision.get("responsibility") not in {"security", "operations"}:
+        raise PolicyApprovalError("policy key trust responsibility is invalid")
+    if decision.get("attestations") != TRUST_REVIEW_ATTESTATIONS:
+        raise PolicyApprovalError("policy key trust attestations are incomplete")
+    if decision.get("policy_activation_authorized") is not False:
+        raise PolicyApprovalError("policy key trust approval must not authorize policy activation")
+    checked = parse_timestamp(checked_at, label="checked_at")
+    approved = parse_timestamp(_required_text(decision, "approved_at"), label="approved_at")
+    expires = parse_timestamp(_required_text(decision, "expires_at"), label="expires_at")
+    if checked < approved or checked > expires:
+        raise PolicyApprovalError("policy key trust approval is not valid at checked_at")
+    trust_approver = payload.get("trust_approver")
+    legal_approver = payload.get("legal_approver")
+    if not isinstance(trust_approver, dict) or not isinstance(legal_approver, dict):
+        raise PolicyApprovalError("policy key trust approver bindings are invalid")
+    expected_payload = build_trust_approval_payload(
+        trust_request,
+        approval=approval,
+        candidate_keyring=candidate_keyring,
+        policy_dir=policy_dir,
+        trust_approver_public_key_file=trust_approver_public_key_file,
+        key_id=_required_text(trust_approver, "key_id"),
+        approver_id=_required_text(trust_approver, "approver_id"),
+        reviewer=_required_text(trust_approver, "reviewer"),
+        role=_required_text(trust_approver, "role"),
+        organization=_required_text(trust_approver, "organization"),
+        responsibility=_required_text(decision, "responsibility"),
+        approved_at=_required_text(decision, "approved_at"),
+        expires_at=_required_text(decision, "expires_at"),
+        key_not_before=_required_text(trust_approver, "key_not_before"),
+        key_not_after=_required_text(trust_approver, "key_not_after"),
+    )
+    if canonical_json(payload) != canonical_json(expected_payload):
+        raise PolicyApprovalError("policy key trust approval payload does not match its source material")
+    public_key, public_pem, public_raw = load_public_key(trust_approver_public_key_file)
+    if sha256_bytes(public_pem) != trust_approver.get("public_key_file_sha256") or sha256_bytes(
+        public_raw
+    ) != trust_approver.get("public_key_raw_sha256"):
+        raise PolicyApprovalError("trust approver public key does not match the signed payload")
+    if sha256_bytes(public_raw) == legal_approver.get("public_key_raw_sha256"):
+        raise PolicyApprovalError("trust approver key must differ from the legal approval key")
+    try:
+        signature = base64.b64decode(_required_text(trust_approval, "signature"), validate=True)
+        public_key.verify(signature, TRUST_SIGNATURE_DOMAIN + canonical_json(payload))
+    except Exception as exc:
+        raise PolicyApprovalError("policy key trust approval signature verification failed") from exc
+    expected_id = f"policy-key-trust-approval-{sha256_bytes(canonical_json(payload) + signature)[:24]}"
+    if trust_approval.get("trust_approval_id") != expected_id:
+        raise PolicyApprovalError("policy key trust approval_id does not match its signed material")
+    return {
+        **candidate_keyring,
+        "status": "trusted",
+        "trust_basis": {
+            "trust_approval_id": expected_id,
+            "trust_approval_sha256": sha256_bytes(canonical_json(trust_approval)),
+            "trust_request_sha256": expected_bindings["trust_request_sha256"],
+            "trusted_at": checked_at,
+            "trust_approval": trust_approval,
+            "trust_approver_public_key_pem": public_pem.decode("ascii"),
+            "trust_approver": {
+                field: trust_approver[field]
+                for field in (
+                    "approver_id",
+                    "reviewer",
+                    "role",
+                    "organization",
+                    "key_id",
+                    "public_key_file_sha256",
+                    "public_key_raw_sha256",
+                )
+            },
+            "policy_activation_authorized": False,
+        },
+    }
+
+
 def _required_text(value: dict[str, Any], key: str) -> str:
     return _nonempty(value.get(key), key)
 
@@ -538,6 +978,97 @@ def _verify_command(args: argparse.Namespace) -> dict[str, Any]:
     )
 
 
+def _prepare_trust_request_command(args: argparse.Namespace) -> dict[str, Any]:
+    approval = _load_json_file(args.approval_file, label="policy approval")
+    keyring = _load_json_file(args.candidate_keyring_file, label="candidate policy keyring")
+    request = build_trust_request(
+        approval,
+        candidate_keyring=keyring,
+        policy_dir=args.policy_dir,
+        requested_at=args.requested_at or datetime.now(timezone.utc).astimezone().isoformat(),
+    )
+    raw = canonical_json(request)
+    write_atomic(args.output, raw)
+    return {
+        "status": request["status"],
+        "trust_request_sha256": sha256_bytes(raw),
+        "approval_id": request["approval_id"],
+        "activation_authorized": False,
+    }
+
+
+def _prepare_trust_approval_command(args: argparse.Namespace) -> dict[str, Any]:
+    trust_request = _load_json_file(args.trust_request_file, label="policy trust request")
+    approval = _load_json_file(args.approval_file, label="policy approval")
+    keyring = _load_json_file(args.candidate_keyring_file, label="candidate policy keyring")
+    payload = build_trust_approval_payload(
+        trust_request,
+        approval=approval,
+        candidate_keyring=keyring,
+        policy_dir=args.policy_dir,
+        trust_approver_public_key_file=args.trust_approver_public_key_file,
+        key_id=args.key_id,
+        approver_id=args.approver_id,
+        reviewer=args.reviewer,
+        role=args.role,
+        organization=args.organization,
+        responsibility=args.responsibility,
+        approved_at=args.approved_at,
+        expires_at=args.expires_at,
+        key_not_before=args.key_not_before,
+        key_not_after=args.key_not_after,
+    )
+    raw = canonical_json(payload)
+    write_atomic(args.output, raw)
+    return {
+        "status": payload["status"],
+        "trust_approval_payload_sha256": sha256_bytes(raw),
+        "activation_authorized": False,
+    }
+
+
+def _sign_trust_approval_command(args: argparse.Namespace) -> dict[str, Any]:
+    payload = _load_json_file(args.payload_file, label="policy trust approval payload")
+    trust_approval = sign_trust_approval_payload(
+        payload,
+        private_key_file=args.private_key_file,
+        password_file=args.private_key_password_file,
+        confirmation=args.confirmation,
+    )
+    raw = canonical_json(trust_approval)
+    write_atomic(args.output, raw, private=True)
+    return {
+        "status": "signed_independent_trust_approval",
+        "trust_approval_id": trust_approval["trust_approval_id"],
+        "trust_approval_sha256": sha256_bytes(raw),
+        "activation_authorized": False,
+    }
+
+
+def _issue_trusted_keyring_command(args: argparse.Namespace) -> dict[str, Any]:
+    trust_approval = _load_json_file(args.trust_approval_file, label="policy key trust approval")
+    trust_request = _load_json_file(args.trust_request_file, label="policy trust request")
+    approval = _load_json_file(args.approval_file, label="policy approval")
+    keyring = _load_json_file(args.candidate_keyring_file, label="candidate policy keyring")
+    trusted = issue_trusted_keyring(
+        trust_approval,
+        trust_request=trust_request,
+        approval=approval,
+        candidate_keyring=keyring,
+        policy_dir=args.policy_dir,
+        trust_approver_public_key_file=args.trust_approver_public_key_file,
+        checked_at=args.checked_at or datetime.now(timezone.utc).astimezone().isoformat(),
+    )
+    raw = canonical_json(trusted)
+    write_atomic(args.output, raw)
+    return {
+        "status": "trusted_keyring_issued_policy_not_activated",
+        "trusted_keyring_sha256": sha256_bytes(raw),
+        "trust_approval_id": trusted["trust_basis"]["trust_approval_id"],
+        "activation_authorized": False,
+    }
+
+
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description=__doc__)
     commands = parser.add_subparsers(dest="command", required=True)
@@ -576,6 +1107,60 @@ def parse_args() -> argparse.Namespace:
     verify.add_argument("--policy-dir", type=Path, required=True)
     verify.add_argument("--checked-at")
     verify.add_argument("--allow-candidate-keyring", action="store_true")
+
+    trust = commands.add_parser(
+        "prepare-trust-request",
+        help="bind a verified signed approval to a candidate keyring without granting trust",
+    )
+    trust.add_argument("--approval-file", type=Path, required=True)
+    trust.add_argument("--candidate-keyring-file", type=Path, required=True)
+    trust.add_argument("--policy-dir", type=Path, required=True)
+    trust.add_argument("--requested-at")
+    trust.add_argument("--output", type=Path, required=True)
+
+    trust_prepare = commands.add_parser(
+        "prepare-trust-approval",
+        help="create the exact independent security or operations approval payload",
+    )
+    trust_prepare.add_argument("--trust-request-file", type=Path, required=True)
+    trust_prepare.add_argument("--approval-file", type=Path, required=True)
+    trust_prepare.add_argument("--candidate-keyring-file", type=Path, required=True)
+    trust_prepare.add_argument("--policy-dir", type=Path, required=True)
+    trust_prepare.add_argument("--trust-approver-public-key-file", type=Path, required=True)
+    trust_prepare.add_argument("--key-id", required=True)
+    trust_prepare.add_argument("--approver-id", required=True)
+    trust_prepare.add_argument("--reviewer", required=True)
+    trust_prepare.add_argument("--role", required=True)
+    trust_prepare.add_argument("--organization", required=True)
+    trust_prepare.add_argument("--responsibility", choices=("security", "operations"), required=True)
+    trust_prepare.add_argument("--approved-at", required=True)
+    trust_prepare.add_argument("--expires-at", required=True)
+    trust_prepare.add_argument("--key-not-before", required=True)
+    trust_prepare.add_argument("--key-not-after", required=True)
+    trust_prepare.add_argument("--output", type=Path, required=True)
+
+    trust_sign = commands.add_parser(
+        "sign-trust-approval",
+        help="sign an exact trust approval payload with the independent approver key",
+    )
+    trust_sign.add_argument("--payload-file", type=Path, required=True)
+    trust_sign.add_argument("--private-key-file", type=Path, required=True)
+    trust_sign.add_argument("--private-key-password-file", type=Path, required=True)
+    trust_sign.add_argument("--confirmation", required=True)
+    trust_sign.add_argument("--output", type=Path, required=True)
+
+    trust_issue = commands.add_parser(
+        "issue-trusted-keyring",
+        help="verify both approvals and issue a trusted keyring without activating policies",
+    )
+    trust_issue.add_argument("--trust-approval-file", type=Path, required=True)
+    trust_issue.add_argument("--trust-request-file", type=Path, required=True)
+    trust_issue.add_argument("--approval-file", type=Path, required=True)
+    trust_issue.add_argument("--candidate-keyring-file", type=Path, required=True)
+    trust_issue.add_argument("--policy-dir", type=Path, required=True)
+    trust_issue.add_argument("--trust-approver-public-key-file", type=Path, required=True)
+    trust_issue.add_argument("--checked-at")
+    trust_issue.add_argument("--output", type=Path, required=True)
     return parser.parse_args()
 
 
@@ -588,6 +1173,14 @@ def main() -> int:
             result = _finalize_command(args)
         elif args.command == "sign":
             result = _sign_command(args)
+        elif args.command == "prepare-trust-request":
+            result = _prepare_trust_request_command(args)
+        elif args.command == "prepare-trust-approval":
+            result = _prepare_trust_approval_command(args)
+        elif args.command == "sign-trust-approval":
+            result = _sign_trust_approval_command(args)
+        elif args.command == "issue-trusted-keyring":
+            result = _issue_trusted_keyring_command(args)
         else:
             result = _verify_command(args)
     except (OSError, PolicyApprovalError) as exc:

yuqei-infra/scripts/verify-provider-promotion-nodes.py

@@ -10,6 +10,7 @@ import re
 import sys
 import time
 from dataclasses import asdict, dataclass
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Callable, Literal
 from urllib.error import HTTPError, URLError
@@ -40,6 +41,10 @@ class NodeObservation:
     target_image_digest: str | None
     provider_problems: str
     kill_switch_active: bool
+    environment: str | None = None
+    generated_at: str | None = None
+    verification_mode: str | None = None
+    promotion_diagnostics_consistent: bool = False
     error: str | None = None
 
 
@@ -58,6 +63,8 @@ def fetch_node_readiness(
         method="GET",
         headers={
             "Accept": "application/json",
+            "Cache-Control": "no-cache, no-store",
+            "Pragma": "no-cache",
             "User-Agent": "yuqei-provider-promotion-canary/1",
             **({"X-AI-Platform-Service-Key": service_key} if service_key else {}),
         },
@@ -89,17 +96,44 @@ def parse_readiness_observation(node: str, payload: dict[str, Any]) -> NodeObser
     kill_switch = _check_by_code(checks, "provider_promotion_kill_switch")
     provider_details = provider.get("details") if isinstance(provider.get("details"), dict) else {}
     kill_details = kill_switch.get("details") if isinstance(kill_switch.get("details"), dict) else {}
+    diagnostics = payload.get("promotion_diagnostics")
+    if not isinstance(diagnostics, dict):
+        raise ValueError("promotion diagnostics are missing")
+    environment = _optional_string(payload.get("environment"))
+    generated_at = _parse_aware_datetime(payload.get("generated_at"), label="generated_at").isoformat()
+    verification_mode = _optional_string(diagnostics.get("verification_mode"))
+    provider_verified = payload.get("provider_acceptance_verified") is True
+    kill_switch_active = kill_details.get("active") is True
+    diagnostic_bindings = (
+        ("release_id", "target_release_id"),
+        ("image_digest", "target_image_digest"),
+        ("receipt_id", "receipt_id"),
+        ("key_id", "key_id"),
+    )
+    diagnostics_consistent = (
+        diagnostics.get("verified") is provider_verified
+        and diagnostics.get("kill_switch_active") is kill_switch_active
+        and all(
+            provider_details.get(detail_key) is None
+            or diagnostics.get(diagnostic_key) == provider_details.get(detail_key)
+            for diagnostic_key, detail_key in diagnostic_bindings
+        )
+    )
     return NodeObservation(
         node=node,
         readiness_status=str(payload.get("status")) if payload.get("status") is not None else None,
         release_allowed=payload.get("release_allowed") is True,
-        provider_acceptance_verified=payload.get("provider_acceptance_verified") is True,
+        provider_acceptance_verified=provider_verified,
         receipt_id=_optional_string(provider_details.get("receipt_id")),
         key_id=_optional_string(provider_details.get("key_id")),
         target_release_id=_optional_string(provider_details.get("target_release_id")),
         target_image_digest=_optional_string(provider_details.get("target_image_digest")),
         provider_problems=_optional_string(provider_details.get("problems")) or "",
-        kill_switch_active=kill_details.get("active") is True,
+        kill_switch_active=kill_switch_active,
+        environment=environment,
+        generated_at=generated_at,
+        verification_mode=verification_mode,
+        promotion_diagnostics_consistent=diagnostics_consistent,
     )
 
 
@@ -110,8 +144,14 @@ def verify_canary(
     expected_receipt_id: str,
     expected_key_id: str,
     expected_image_digest: str,
+    max_observation_age_seconds: float = 120.0,
+    now: datetime | None = None,
 ) -> list[str]:
-    problems: list[str] = []
+    problems = verify_observation_provenance(
+        observations,
+        max_age_seconds=max_observation_age_seconds,
+        now=now,
+    )
     for item in observations:
         if item.error:
             problems.append(f"{item.node}: {item.error}")
@@ -139,6 +179,38 @@ def verify_canary(
     return problems
 
 
+def verify_observation_provenance(
+    observations: list[NodeObservation],
+    *,
+    max_age_seconds: float,
+    now: datetime | None = None,
+) -> list[str]:
+    observed_at = now or datetime.now(UTC)
+    if observed_at.tzinfo is None:
+        raise ValueError("verification time must include a timezone")
+    problems: list[str] = []
+    for item in observations:
+        if item.error:
+            continue
+        if item.environment != "production":
+            problems.append(f"{item.node}: readiness is not from production")
+        if item.verification_mode != "offline_read_only":
+            problems.append(f"{item.node}: readiness is not offline read-only verification")
+        if not item.promotion_diagnostics_consistent:
+            problems.append(f"{item.node}: promotion diagnostics are inconsistent")
+        try:
+            generated_at = _parse_aware_datetime(item.generated_at, label="generated_at")
+        except ValueError:
+            problems.append(f"{item.node}: readiness generation time is invalid")
+            continue
+        age_seconds = (observed_at.astimezone(UTC) - generated_at.astimezone(UTC)).total_seconds()
+        if age_seconds < -30:
+            problems.append(f"{item.node}: readiness generation time is in the future")
+        elif age_seconds > max_age_seconds:
+            problems.append(f"{item.node}: readiness response is stale")
+    return problems
+
+
 def wait_for_control(
     nodes: list[str],
     *,
@@ -197,8 +269,32 @@ def _optional_string(value: Any) -> str | None:
     return value if isinstance(value, str) and value else None
 
 
+def _parse_aware_datetime(value: Any, *, label: str) -> datetime:
+    if not isinstance(value, str) or not value:
+        raise ValueError(f"{label} is missing")
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError:
+        raise ValueError(f"{label} is invalid") from None
+    if parsed.tzinfo is None:
+        raise ValueError(f"{label} must include a timezone")
+    return parsed
+
+
 def _failed_observation(node: str, error: str) -> NodeObservation:
-    return NodeObservation(node, None, False, False, None, None, None, None, "", False, error)
+    return NodeObservation(
+        node=node,
+        readiness_status=None,
+        release_allowed=False,
+        provider_acceptance_verified=False,
+        receipt_id=None,
+        key_id=None,
+        target_release_id=None,
+        target_image_digest=None,
+        provider_problems="",
+        kill_switch_active=False,
+        error=error,
+    )
 
 
 def _read_service_key(path: Path | None) -> str | None:
@@ -232,6 +328,7 @@ def _parse_args() -> argparse.Namespace:
     parser.add_argument("--timeout-seconds", type=float, default=5.0)
     parser.add_argument("--deadline-seconds", type=float, default=300.0)
     parser.add_argument("--poll-interval-seconds", type=float, default=2.0)
+    parser.add_argument("--max-observation-age-seconds", type=float, default=120.0)
     return parser.parse_args()
 
 
@@ -243,6 +340,8 @@ def main() -> int:
             raise ValueError("multi-node verification requires at least two distinct nodes")
         if not 0.1 <= args.timeout_seconds <= 60 or not math.isfinite(args.timeout_seconds):
             raise ValueError("timeout must be between 0.1 and 60 seconds")
+        if not 1 <= args.max_observation_age_seconds <= 600 or not math.isfinite(args.max_observation_age_seconds):
+            raise ValueError("maximum observation age must be between 1 and 600 seconds")
         service_key = _read_service_key(args.service_key_file)
         fetcher = lambda node: fetch_node_readiness(
             node,
@@ -262,6 +361,7 @@ def main() -> int:
                 expected_receipt_id=args.expected_receipt_id,
                 expected_key_id=args.expected_key_id,
                 expected_image_digest=args.expected_image_digest,
+                max_observation_age_seconds=args.max_observation_age_seconds,
             )
         else:
             if not 1 <= args.deadline_seconds <= 900 or not math.isfinite(args.deadline_seconds):
@@ -275,6 +375,12 @@ def main() -> int:
                 deadline_seconds=args.deadline_seconds,
                 poll_interval_seconds=args.poll_interval_seconds,
             )
+            problems.extend(
+                verify_observation_provenance(
+                    observations,
+                    max_age_seconds=args.max_observation_age_seconds,
+                )
+            )
         summary = {
             "mode": args.mode,
             "status": "passed" if not problems else "failed",

yuqei-infra/tests/test_alert_receiver.py

@@ -0,0 +1,122 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+import threading
+from pathlib import Path
+from types import ModuleType
+from urllib.error import HTTPError
+from urllib.request import Request, urlopen
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def load_script() -> ModuleType:
+    path = REPO_ROOT / "yuqei-infra/observability/receive-production-alert.py"
+    spec = importlib.util.spec_from_file_location("yuqei_alert_receiver", path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+receiver = load_script()
+
+
+def _start(tmp_path: Path) -> tuple[object, str, threading.Thread]:
+    token = "a" * 64
+    token_file = tmp_path / "token"
+    token_file.write_text(token + "\n", encoding="utf-8")
+    server = receiver.build_server(
+        "127.0.0.1",
+        0,
+        token_file=token_file,
+        event_dir=tmp_path / "events",
+        max_body_bytes=4096,
+    )
+    thread = threading.Thread(target=server.serve_forever, daemon=True)
+    thread.start()
+    url = f"http://127.0.0.1:{server.server_address[1]}/ops/alerts/{token}"
+    return server, url, thread
+
+
+def _payload() -> bytes:
+    return json.dumps(
+        {
+            "status": "warning",
+            "checked_at": "2026-08-12T10:00:00+08:00",
+            "warning_checks": ["disk_root"],
+            "failed_checks": [],
+            "alert_delivery": {"attempted": True, "delivered": False},
+        }
+    ).encode("utf-8")
+
+
+def test_receiver_persists_event_and_returns_event_id(tmp_path: Path) -> None:
+    server, url, thread = _start(tmp_path)
+    try:
+        request = Request(url, data=_payload(), method="POST", headers={"Content-Type": "application/json"})
+        with urlopen(request, timeout=5) as response:
+            result = json.loads(response.read())
+            assert response.status == 201
+        event_path = tmp_path / "events" / f"{result['event_id']}.json"
+        receipt = json.loads(event_path.read_text(encoding="utf-8"))
+        assert receipt["event_id"] == result["event_id"]
+        assert receipt["payload"]["warning_checks"] == ["disk_root"]
+        assert len(receipt["body_sha256"]) == 64
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+@pytest.mark.parametrize(
+    ("path_suffix", "content_type", "body", "expected_status"),
+    [
+        ("wrong-token", "application/json", _payload(), 404),
+        ("a" * 64, "text/plain", _payload(), 415),
+        ("a" * 64, "application/json", b'{"status":"pass","checked_at":"x"}', 400),
+    ],
+)
+def test_receiver_rejects_unauthenticated_or_invalid_alerts(
+    tmp_path: Path,
+    path_suffix: str,
+    content_type: str,
+    body: bytes,
+    expected_status: int,
+) -> None:
+    server, url, thread = _start(tmp_path)
+    try:
+        base = url.rsplit("/", 1)[0]
+        request = Request(
+            f"{base}/{path_suffix}",
+            data=body,
+            method="POST",
+            headers={"Content-Type": content_type},
+        )
+        with pytest.raises(HTTPError) as error:
+            urlopen(request, timeout=5)
+        assert error.value.code == expected_status
+        event_dir = tmp_path / "events"
+        assert not event_dir.exists() or not list(event_dir.glob("*.json"))
+    finally:
+        server.shutdown()
+        server.server_close()
+        thread.join(timeout=5)
+
+
+def test_nginx_location_is_post_only_and_loopback_proxy() -> None:
+    snippet = (
+        REPO_ROOT / "yuqei-infra/observability/nginx-production-alert-location.conf"
+    ).read_text(encoding="utf-8")
+
+    assert "location ^~ /ops/alerts/" in snippet
+    assert "proxy_pass http://127.0.0.1:19091;" in snippet
+    assert "client_max_body_size 1m;" in snippet
+    assert "limit_except POST" in snippet
+    assert "deny all;" in snippet

yuqei-infra/tests/test_observability_capacity.py

@@ -169,6 +169,8 @@ def _main_args(tmp_path: Path) -> SimpleNamespace:
         backup_require_separate_mount=True,
         timeout_seconds=1,
         webhook_url_file=tmp_path / "webhook-url",
+        alert_state_file=None,
+        alert_resend_seconds=3600,
         artifact_dir=tmp_path / "artifact",
     )
 
@@ -234,9 +236,38 @@ def test_warning_dispatches_webhook_and_records_acceptance_artifact(
         "attempted": True,
         "delivered": True,
         "error": None,
+        "receiver_event_id": None,
+        "receiver_received_at": None,
+        "receiver_body_sha256": None,
+        "suppressed": False,
+        "decision": "deduplication_not_configured",
     }
 
 
+def test_webhook_receipt_is_recorded_in_local_acceptance_artifact(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    args = _main_args(tmp_path)
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+    monkeypatch.setattr(
+        health,
+        "send_webhook",
+        lambda *_args, **_kwargs: {
+            "event_id": "yuqei-alert-0123456789abcdef",
+            "received_at": "2026-08-12T02:00:00+00:00",
+            "body_sha256": "a" * 64,
+        },
+    )
+
+    assert health.main() == 0
+    summary = json.loads((args.artifact_dir / "production-health-summary.json").read_text(encoding="utf-8"))
+    assert summary["alert_delivery"]["receiver_event_id"] == "yuqei-alert-0123456789abcdef"
+    assert summary["alert_delivery"]["receiver_received_at"] == "2026-08-12T02:00:00+00:00"
+    assert summary["alert_delivery"]["receiver_body_sha256"] == "a" * 64
+
+
 def test_webhook_delivery_failure_is_visible_to_systemd(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -259,6 +290,102 @@ def test_webhook_delivery_failure_is_visible_to_systemd(
     assert summary["alert_delivery"]["error"] == "TimeoutError"
 
 
+def test_duplicate_alert_is_suppressed_until_resend_window(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    args = _main_args(tmp_path)
+    args.alert_state_file = tmp_path / "alert-state.json"
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+    delivered: list[str] = []
+    monkeypatch.setattr(
+        health,
+        "send_webhook",
+        lambda _path, summary, **_kwargs: delivered.append(summary["status"]) or {},
+    )
+
+    assert health.main() == 0
+    assert health.main() == 0
+    summary = json.loads((args.artifact_dir / "production-health-summary.json").read_text(encoding="utf-8"))
+
+    assert delivered == ["warning"]
+    assert summary["alert_delivery"]["attempted"] is False
+    assert summary["alert_delivery"]["suppressed"] is True
+    assert summary["alert_delivery"]["decision"] == "duplicate_within_resend_window"
+
+
+def test_alert_state_change_bypasses_deduplication(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    args = _main_args(tmp_path)
+    args.alert_state_file = tmp_path / "alert-state.json"
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+    delivered: list[str] = []
+    monkeypatch.setattr(
+        health,
+        "send_webhook",
+        lambda _path, summary, **_kwargs: delivered.append(summary["status"]) or {},
+    )
+
+    assert health.main() == 0
+    monkeypatch.setattr(
+        health,
+        "check_disk",
+        lambda **policy: health.HealthCheck(
+            policy["key"],
+            "fail" if policy["key"] == "disk_root" else "pass",
+            "changed",
+        ),
+    )
+    assert health.main() == 1
+
+    assert delivered == ["warning", "fail"]
+
+
+def test_healthy_run_resets_alert_state(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    args = _main_args(tmp_path)
+    args.alert_state_file = tmp_path / "alert-state.json"
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+    monkeypatch.setattr(health, "send_webhook", lambda *_args, **_kwargs: {})
+    assert health.main() == 0
+
+    monkeypatch.setattr(
+        health,
+        "check_disk",
+        lambda **policy: health.HealthCheck(policy["key"], "pass", "healthy"),
+    )
+    assert health.main() == 0
+
+    state = json.loads(args.alert_state_file.read_text(encoding="utf-8"))
+    assert state["fingerprint"] is None
+    assert state["delivered_at"] is None
+
+
+def test_alert_state_write_failure_is_recorded_and_exits_two(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    args = _main_args(tmp_path)
+    args.alert_state_file = tmp_path / "alert-state.json"
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+    monkeypatch.setattr(health, "send_webhook", lambda *_args, **_kwargs: {})
+    monkeypatch.setattr(
+        health,
+        "write_alert_state",
+        lambda *_args, **_kwargs: (_ for _ in ()).throw(OSError("read-only filesystem")),
+    )
+
+    assert health.main() == 2
+    summary = json.loads((args.artifact_dir / "production-health-summary.json").read_text(encoding="utf-8"))
+    assert summary["alert_delivery"]["delivered"] is True
+    assert summary["alert_delivery"]["error"] == "StateWriteError"
+
+
 def test_failed_capacity_check_dispatches_webhook_and_exits_one(
     monkeypatch: pytest.MonkeyPatch,
     tmp_path: Path,
@@ -348,4 +475,8 @@ def test_systemd_timer_and_service_cover_all_capacity_paths() -> None:
     assert "--home-disk-path /home" in unit
     assert "--home-require-separate-mount" in unit
     assert "--backup-disk-path ${BACKUP_HOST_DIR}" in unit
+    assert "--backup-disk-critical-free-gib ${BACKUP_DISK_CRITICAL_FREE_GIB}" in unit
     assert "--backup-require-separate-mount" in unit
+    assert "--webhook-url-file ${PRODUCTION_ALERT_WEBHOOK_URL_FILE}" in unit
+    assert "--alert-state-file /opt/apps/ai-contract-platform-v2-production/project-docs/qa-artifacts/v2-production-health/alert-state.json" in unit
+    assert "--alert-resend-seconds 3600" in unit

yuqei-infra/tests/test_policy_legal_signing_helper.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import base64
 import importlib.util
 import json
 import shutil
@@ -193,3 +194,333 @@ def test_sign_rejects_a_different_private_key(tmp_path: Path) -> None:
             password_file=other_password,
             confirmation=policy.SIGN_CONFIRMATION,
         )
+
+
+def test_trust_request_binds_signed_approval_without_granting_trust(tmp_path: Path) -> None:
+    _package, pending, keyring, private, _public, password = prepare(tmp_path)
+    final_payload = policy.finalize_payload(
+        pending,
+        approved_at="2026-08-12T13:00:00+08:00",
+        expires_at="2026-08-19T13:00:00+08:00",
+    )
+    approval = policy.sign_payload(
+        final_payload,
+        private_key_file=private,
+        password_file=password,
+        confirmation=policy.SIGN_CONFIRMATION,
+    )
+
+    trust_request = policy.build_trust_request(
+        approval,
+        candidate_keyring=keyring,
+        policy_dir=POLICY_SOURCE,
+        requested_at="2026-08-12T14:00:00+08:00",
+    )
+
+    assert trust_request["schema_version"] == policy.TRUST_REQUEST_SCHEMA
+    assert trust_request["status"] == "pending_independent_trust_approval"
+    assert trust_request["activation_authorized"] is False
+    assert trust_request["approval_id"] == approval["approval_id"]
+    assert trust_request["approval_sha256"] == policy.sha256_bytes(policy.canonical_json(approval))
+    assert trust_request["candidate_keyring_sha256"] == policy.sha256_bytes(
+        policy.canonical_json(keyring)
+    )
+    assert "mark_candidate_keyring_trusted" in trust_request["prohibited_automatic_actions"]
+    assert "publish_policy_documents" in trust_request["prohibited_automatic_actions"]
+
+
+def test_trust_request_rejects_expired_approval_or_non_candidate_keyring(tmp_path: Path) -> None:
+    _package, pending, keyring, private, _public, password = prepare(tmp_path)
+    final_payload = policy.finalize_payload(
+        pending,
+        approved_at="2026-08-12T13:00:00+08:00",
+        expires_at="2026-08-19T13:00:00+08:00",
+    )
+    approval = policy.sign_payload(
+        final_payload,
+        private_key_file=private,
+        password_file=password,
+        confirmation=policy.SIGN_CONFIRMATION,
+    )
+
+    with pytest.raises(policy.PolicyApprovalError, match="not valid at checked_at"):
+        policy.build_trust_request(
+            approval,
+            candidate_keyring=keyring,
+            policy_dir=POLICY_SOURCE,
+            requested_at="2026-08-19T13:00:01+08:00",
+        )
+
+    trusted = {**keyring, "status": "trusted"}
+    with pytest.raises(policy.PolicyApprovalError, match="signed trust basis"):
+        policy.verify_approval(
+            approval,
+            keyring=trusted,
+            policy_dir=POLICY_SOURCE,
+            checked_at="2026-08-12T14:00:00+08:00",
+            allow_candidate_keyring=False,
+        )
+    with pytest.raises(policy.PolicyApprovalError, match="candidate_not_trusted"):
+        policy.build_trust_request(
+            approval,
+            candidate_keyring=trusted,
+            policy_dir=POLICY_SOURCE,
+            requested_at="2026-08-12T14:00:00+08:00",
+        )
+
+
+def make_signed_trust_chain(tmp_path: Path) -> tuple[dict, dict, dict, dict, Path]:
+    _package, pending, keyring, legal_private, _legal_public, legal_password = prepare(tmp_path)
+    legal_payload = policy.finalize_payload(
+        pending,
+        approved_at="2026-08-12T13:00:00+08:00",
+        expires_at="2026-08-19T13:00:00+08:00",
+    )
+    legal_approval = policy.sign_payload(
+        legal_payload,
+        private_key_file=legal_private,
+        password_file=legal_password,
+        confirmation=policy.SIGN_CONFIRMATION,
+    )
+    trust_request = policy.build_trust_request(
+        legal_approval,
+        candidate_keyring=keyring,
+        policy_dir=POLICY_SOURCE,
+        requested_at="2026-08-12T14:00:00+08:00",
+    )
+    trust_private, trust_public, trust_password = make_key_files(tmp_path, stem="trust-approver")
+    trust_payload = policy.build_trust_approval_payload(
+        trust_request,
+        approval=legal_approval,
+        candidate_keyring=keyring,
+        policy_dir=POLICY_SOURCE,
+        trust_approver_public_key_file=trust_public,
+        key_id="policy-trust-ops-2026q3",
+        approver_id="security@example.test",
+        reviewer="Named Security Reviewer",
+        role="Security Officer",
+        organization="Independent Security Operations",
+        responsibility="security",
+        approved_at="2026-08-12T15:00:00+08:00",
+        expires_at="2026-08-19T12:00:00+08:00",
+        key_not_before="2026-08-01T00:00:00+08:00",
+        key_not_after="2026-09-30T23:59:59+08:00",
+    )
+    trust_approval = policy.sign_trust_approval_payload(
+        trust_payload,
+        private_key_file=trust_private,
+        password_file=trust_password,
+        confirmation=policy.TRUST_SIGN_CONFIRMATION,
+    )
+    return legal_approval, keyring, trust_request, trust_approval, trust_public
+
+
+def test_independent_signed_trust_approval_is_required_to_issue_trusted_keyring(
+    tmp_path: Path,
+) -> None:
+    legal_approval, keyring, trust_request, trust_approval, trust_public = make_signed_trust_chain(
+        tmp_path
+    )
+
+    trusted = policy.issue_trusted_keyring(
+        trust_approval,
+        trust_request=trust_request,
+        approval=legal_approval,
+        candidate_keyring=keyring,
+        policy_dir=POLICY_SOURCE,
+        trust_approver_public_key_file=trust_public,
+        checked_at="2026-08-12T16:00:00+08:00",
+    )
+
+    assert trusted["status"] == "trusted"
+    assert trusted["keys"] == keyring["keys"]
+    assert trusted["trust_basis"]["policy_activation_authorized"] is False
+    assert trusted["trust_basis"]["trust_request_sha256"] == policy.sha256_bytes(
+        policy.canonical_json(trust_request)
+    )
+    verified = policy.verify_approval(
+        legal_approval,
+        keyring=trusted,
+        policy_dir=POLICY_SOURCE,
+        checked_at="2026-08-12T16:00:00+08:00",
+        allow_candidate_keyring=False,
+    )
+    assert verified["status"] == "verified"
+
+    tampered = json.loads(json.dumps(trusted))
+    signature = tampered["trust_basis"]["trust_approval"]["signature"]
+    tampered["trust_basis"]["trust_approval"]["signature"] = "A" + signature[1:]
+    with pytest.raises(policy.PolicyApprovalError, match="trust approval hash"):
+        policy.verify_approval(
+            legal_approval,
+            keyring=tampered,
+            policy_dir=POLICY_SOURCE,
+            checked_at="2026-08-12T16:00:00+08:00",
+            allow_candidate_keyring=False,
+        )
+
+    incomplete = json.loads(json.dumps(trusted))
+    incomplete["trust_basis"]["trust_approval"]["payload"]["decision"]["attestations"] = []
+    incomplete["trust_basis"]["trust_approval_sha256"] = policy.sha256_bytes(
+        policy.canonical_json(incomplete["trust_basis"]["trust_approval"])
+    )
+    with pytest.raises(policy.PolicyApprovalError, match="attestations are incomplete"):
+        policy.verify_approval(
+            legal_approval,
+            keyring=incomplete,
+            policy_dir=POLICY_SOURCE,
+            checked_at="2026-08-12T16:00:00+08:00",
+            allow_candidate_keyring=False,
+        )
+
+    different_approval = json.loads(json.dumps(legal_approval))
+    different_approval["approval_id"] = "policy-approval-different-material"
+    with pytest.raises(policy.PolicyApprovalError, match="exact policy approval"):
+        policy.verify_trusted_keyring_basis(trusted, approval=different_approval)
+
+
+def test_trust_approval_rejects_same_approver_or_same_key(tmp_path: Path) -> None:
+    _package, pending, keyring, legal_private, legal_public, legal_password = prepare(tmp_path)
+    legal_payload = policy.finalize_payload(
+        pending,
+        approved_at="2026-08-12T13:00:00+08:00",
+        expires_at="2026-08-19T13:00:00+08:00",
+    )
+    legal_approval = policy.sign_payload(
+        legal_payload,
+        private_key_file=legal_private,
+        password_file=legal_password,
+        confirmation=policy.SIGN_CONFIRMATION,
+    )
+    trust_request = policy.build_trust_request(
+        legal_approval,
+        candidate_keyring=keyring,
+        policy_dir=POLICY_SOURCE,
+        requested_at="2026-08-12T14:00:00+08:00",
+    )
+
+    common = dict(
+        approval=legal_approval,
+        candidate_keyring=keyring,
+        policy_dir=POLICY_SOURCE,
+        key_id="policy-trust-ops-2026q3",
+        reviewer="Named Reviewer",
+        role="Security Officer",
+        organization="Independent Security Operations",
+        responsibility="security",
+        approved_at="2026-08-12T15:00:00+08:00",
+        expires_at="2026-08-19T12:00:00+08:00",
+        key_not_before="2026-08-01T00:00:00+08:00",
+        key_not_after="2026-09-30T23:59:59+08:00",
+    )
+    _other_private, other_public, _other_password = make_key_files(tmp_path, stem="independent")
+    with pytest.raises(policy.PolicyApprovalError, match="approver must differ"):
+        policy.build_trust_approval_payload(
+            trust_request,
+            trust_approver_public_key_file=other_public,
+            approver_id="cyuqei@gmail.com",
+            **common,
+        )
+    with pytest.raises(policy.PolicyApprovalError, match="key must differ"):
+        policy.build_trust_approval_payload(
+            trust_request,
+            trust_approver_public_key_file=legal_public,
+            approver_id="security@example.test",
+            **common,
+        )
+
+
+def test_issue_trusted_keyring_rejects_tampered_request_and_signature(tmp_path: Path) -> None:
+    legal_approval, keyring, trust_request, trust_approval, trust_public = make_signed_trust_chain(
+        tmp_path
+    )
+    tampered_request = json.loads(json.dumps(trust_request))
+    tampered_request["policy_version"] = "tampered"
+    with pytest.raises(policy.PolicyApprovalError, match="source material"):
+        policy.issue_trusted_keyring(
+            trust_approval,
+            trust_request=tampered_request,
+            approval=legal_approval,
+            candidate_keyring=keyring,
+            policy_dir=POLICY_SOURCE,
+            trust_approver_public_key_file=trust_public,
+            checked_at="2026-08-12T16:00:00+08:00",
+        )
+
+
+def test_trust_approval_requires_signed_independent_review_attestations(tmp_path: Path) -> None:
+    legal_approval, keyring, trust_request, trust_approval, trust_public = make_signed_trust_chain(
+        tmp_path
+    )
+    payload = trust_approval["payload"]
+    assert payload["decision"]["responsibility"] == "security"
+    assert payload["decision"]["attestations"] == policy.TRUST_REVIEW_ATTESTATIONS
+
+    tampered = json.loads(json.dumps(trust_approval))
+    tampered["payload"]["decision"]["attestations"] = []
+    with pytest.raises(policy.PolicyApprovalError, match="attestations are incomplete"):
+        policy.issue_trusted_keyring(
+            tampered,
+            trust_request=trust_request,
+            approval=legal_approval,
+            candidate_keyring=keyring,
+            policy_dir=POLICY_SOURCE,
+            trust_approver_public_key_file=trust_public,
+            checked_at="2026-08-12T16:00:00+08:00",
+        )
+
+    tampered_approval = json.loads(json.dumps(trust_approval))
+    tampered_approval["signature"] = "A" + tampered_approval["signature"][1:]
+    with pytest.raises(policy.PolicyApprovalError, match="signature verification failed"):
+        policy.issue_trusted_keyring(
+            tampered_approval,
+            trust_request=trust_request,
+            approval=legal_approval,
+            candidate_keyring=keyring,
+            policy_dir=POLICY_SOURCE,
+            trust_approver_public_key_file=trust_public,
+            checked_at="2026-08-12T16:00:00+08:00",
+        )
+
+
+def test_issue_reconstructs_full_trust_payload_before_verifying_signature(tmp_path: Path) -> None:
+    legal_approval, keyring, trust_request, trust_approval, trust_public = make_signed_trust_chain(
+        tmp_path
+    )
+    payload = json.loads(json.dumps(trust_approval["payload"]))
+    payload["legal_approver"]["approver_id"] = "different-unbound-id@example.test"
+    trust_private = Ed25519PrivateKey.generate()
+    trust_public.write_bytes(
+        trust_private.public_key().public_bytes(
+            serialization.Encoding.PEM,
+            serialization.PublicFormat.SubjectPublicKeyInfo,
+        )
+    )
+    payload["trust_approver"]["public_key_file_sha256"] = policy.sha256_bytes(
+        trust_public.read_bytes()
+    )
+    payload["trust_approver"]["public_key_raw_sha256"] = policy.sha256_bytes(
+        trust_private.public_key().public_bytes(
+            serialization.Encoding.Raw,
+            serialization.PublicFormat.Raw,
+        )
+    )
+    payload_raw = policy.canonical_json(payload)
+    signature = trust_private.sign(policy.TRUST_SIGNATURE_DOMAIN + payload_raw)
+    forged = {
+        "schema_version": policy.TRUST_APPROVAL_SCHEMA,
+        "trust_approval_id": f"policy-key-trust-approval-{policy.sha256_bytes(payload_raw + signature)[:24]}",
+        "payload": payload,
+        "signature": base64.b64encode(signature).decode("ascii"),
+    }
+
+    with pytest.raises(policy.PolicyApprovalError, match="source material"):
+        policy.issue_trusted_keyring(
+            forged,
+            trust_request=trust_request,
+            approval=legal_approval,
+            candidate_keyring=keyring,
+            policy_dir=POLICY_SOURCE,
+            trust_approver_public_key_file=trust_public,
+            checked_at="2026-08-12T16:00:00+08:00",
+        )

yuqei-infra/tests/test_production_release_tools.py

@@ -82,6 +82,7 @@ def production_values() -> dict[str, str]:
         "AI_PLATFORM_PROVIDER_ACCEPTANCE_EXPECTED_RECEIPT_SHA256": hashlib.sha256(
             b"pinned-production-provider-acceptance-receipt"
         ).hexdigest(),
+        "PRODUCTION_ALERT_WEBHOOK_URL_FILE": "/etc/yuqei/monitoring/alert-webhook-url",
         "AI_PLATFORM_ANTI_ABUSE_HASH_SECRET": "ai-anti-abuse-secret-that-is-long-and-unique-006a",  # pragma: allowlist secret
         "CONTRACT_API_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_contract",
         "AI_PLATFORM_DATABASE_URL": "postgresql+psycopg://yuqei_prod:encoded-password@postgres:5432/yuqei_ai",
@@ -316,6 +317,32 @@ def test_provider_acceptance_promotion_files_are_present_and_hash_pinned(tmp_pat
     assert tampered["provider_acceptance_receipt_hash"].status == "fail"
 
 
+def test_production_preflight_requires_owner_only_https_alert_webhook(tmp_path: Path) -> None:
+    values = production_values()
+    webhook = tmp_path / "alert-webhook-url"
+    webhook.write_text("https://alerts.acme.cn/hooks/yuqei\n", encoding="utf-8")
+    webhook.chmod(0o600)
+    values["PRODUCTION_ALERT_WEBHOOK_URL_FILE"] = str(webhook.resolve())
+
+    checks = {check.key: check for check in preflight.validate_files(values)}
+
+    assert checks["alert_webhook_file"].status == "pass"
+    if "alert_webhook_permissions" in checks:
+        assert checks["alert_webhook_permissions"].status == "pass"
+
+    webhook.write_text("http://alerts.acme.cn/hooks/yuqei\n", encoding="utf-8")
+    assert {check.key: check for check in preflight.validate_files(values)}["alert_webhook_file"].status == "fail"
+
+
+def test_production_preflight_rejects_missing_alert_webhook_file(tmp_path: Path) -> None:
+    values = production_values()
+    values["PRODUCTION_ALERT_WEBHOOK_URL_FILE"] = str((tmp_path / "missing").resolve())
+
+    checks = {check.key: check for check in preflight.validate_files(values)}
+
+    assert checks["alert_webhook_file"].status == "fail"
+
+
 def test_production_compose_mounts_only_promotion_verification_material_read_only() -> None:
     override = (
         REPO_ROOT / "yuqei-infra/compose/docker-compose.production.yml"

yuqei-infra/tests/test_provider_promotion_controls.py

@@ -2,6 +2,7 @@ from __future__ import annotations
 
 import importlib.util
 import sys
+from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
 
@@ -30,6 +31,10 @@ def _observation(
     release_id: str = "release-1",
     problems: str = "",
     kill_switch: bool = False,
+    environment: str = "production",
+    generated_at: str | None = None,
+    verification_mode: str = "offline_read_only",
+    diagnostics_consistent: bool = True,
 ):
     return controls.NodeObservation(
         node=node,
@@ -42,6 +47,10 @@ def _observation(
         target_image_digest="sha256:" + "2" * 64,
         provider_problems=problems,
         kill_switch_active=kill_switch,
+        environment=environment,
+        generated_at=generated_at or datetime.now(UTC).isoformat(),
+        verification_mode=verification_mode,
+        promotion_diagnostics_consistent=diagnostics_consistent,
     )
 
 
@@ -68,6 +77,76 @@ def test_canary_requires_one_fingerprint_across_all_nodes() -> None:
     assert "nodes do not report one immutable promotion fingerprint" in problems
 
 
+def test_canary_requires_fresh_production_offline_read_only_diagnostics() -> None:
+    now = datetime(2026, 8, 12, 1, 0, tzinfo=UTC)
+    observations = [
+        _observation(
+            "https://node-a.example",
+            environment="staging",
+            generated_at=now.isoformat(),
+            verification_mode="repository",
+        ),
+        _observation(
+            "https://node-b.example",
+            generated_at=(now - timedelta(minutes=5)).isoformat(),
+            diagnostics_consistent=False,
+        ),
+    ]
+
+    problems = controls.verify_canary(
+        observations,
+        expected_release_id="release-1",
+        expected_receipt_id="provider-receipt-1234567890abcdef12345678",
+        expected_key_id="receipt-key-1",
+        expected_image_digest="sha256:" + "2" * 64,
+        max_observation_age_seconds=120,
+        now=now,
+    )
+
+    assert any("not from production" in item for item in problems)
+    assert any("not offline read-only" in item for item in problems)
+    assert any("diagnostics are inconsistent" in item for item in problems)
+    assert any("response is stale" in item for item in problems)
+
+
+def test_parse_readiness_observation_cross_checks_diagnostics() -> None:
+    payload = {
+        "environment": "production",
+        "status": "ready",
+        "release_allowed": True,
+        "provider_acceptance_verified": True,
+        "generated_at": datetime.now(UTC).isoformat(),
+        "checks": [
+            {
+                "code": "provider_acceptance",
+                "details": {
+                    "receipt_id": "provider-receipt-1234567890abcdef12345678",
+                    "key_id": "receipt-key-1",
+                    "target_release_id": "release-1",
+                    "target_image_digest": "sha256:" + "2" * 64,
+                    "problems": "",
+                },
+            },
+            {"code": "provider_promotion_kill_switch", "details": {"active": False}},
+        ],
+        "promotion_diagnostics": {
+            "verification_mode": "offline_read_only",
+            "verified": True,
+            "kill_switch_active": False,
+            "release_id": "release-1",
+            "image_digest": "sha256:" + "2" * 64,
+            "receipt_id": "provider-receipt-1234567890abcdef12345678",
+            "key_id": "receipt-key-1",
+        },
+    }
+
+    observation = controls.parse_readiness_observation("https://node-a.example", payload)
+    assert observation.promotion_diagnostics_consistent is True
+    payload["promotion_diagnostics"]["receipt_id"] = "provider-receipt-abcdefabcdefabcdefabcdef"
+    observation = controls.parse_readiness_observation("https://node-a.example", payload)
+    assert observation.promotion_diagnostics_consistent is False
+
+
 def test_revocation_waits_for_every_node_and_honors_sla() -> None:
     now = [0.0]
     calls = {"https://node-a.example": 0, "https://node-b.example": 0}
@@ -152,6 +231,7 @@ def test_control_workflow_is_protected_read_only_and_cleans_credentials() -> Non
     assert "environment: production-provider-promotion-control" in workflow
     assert "permissions:\n  contents: read" in workflow
     assert "verify-provider-promotion-nodes.py" in workflow
+    assert "--max-observation-age-seconds" in workflow
     assert "- canary" in workflow and "- revocation" in workflow and "- kill-switch" in workflow
     assert "${{ inputs.node_a }}" not in workflow.split("run: |", 1)[1]
     assert "${{ inputs.node_b }}" not in workflow.split("run: |", 1)[1]