chore: close support scope and capacity launch gates

project-docs/V2-production-launch-scope-2026-08-12.json

@@ -0,0 +1,39 @@
+{
+  "schema_version": 1,
+  "decision_id": "support-access-first-release-exclusion-2026-08-12",
+  "release_scope": "v2-first-production-release",
+  "decision": "excluded",
+  "support_access": {
+    "included": false,
+    "enabled": false,
+    "kill_switch_engaged": true,
+    "executor_enabled": false
+  },
+  "identity_executor": {
+    "enabled": false
+  },
+  "destructive_identity_write_api": {
+    "included": false,
+    "forbidden_operations": [
+      "execute",
+      "grant",
+      "revoke",
+      "impersonate",
+      "impersonation"
+    ]
+  },
+  "future_enablement_gates": [
+    "external_idp",
+    "mfa_step_up",
+    "session_device_binding",
+    "real_resource_toctou",
+    "cross_node_revocation_sla",
+    "kill_switch_drill",
+    "immutable_per_request_audit",
+    "privacy_legal_approval"
+  ],
+  "notes": [
+    "This is a first-release scope exclusion, not approval of support access.",
+    "No shared password, manual bypass, identity write API, or execute API may restore this capability."
+  ]
+}

yuqei-contract-platform/services/contract-api/tests/test_privileged_identity_operations.py

@@ -1496,9 +1496,22 @@ def test_no_execution_or_destructive_identity_route_is_registered(tmp_path: Path
         "/grant",
         "/revoke",
         "/support-access",
+        "/support_access",
+        "/impersonate",
+        "/impersonation",
     )
     assert not any(path.endswith(forbidden_suffixes) for path in paths)
     assert not any(path.endswith("/retry") for path in paths)
+    forbidden_fragments = (
+        "/execute",
+        "/grant",
+        "/revoke",
+        "/support-access",
+        "/support_access",
+        "/impersonate",
+        "/impersonation",
+    )
+    assert not any(fragment in path.lower() for path in paths for fragment in forbidden_fragments)
     destructive_write_routes = [
         (method.upper(), path)
         for path, operations in openapi_paths.items()

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

@@ -37,6 +37,22 @@ BACKUP_HOST_DIR=/mnt/yuqei-v2-backups
 BACKUP_REQUIRE_SEPARATE_MOUNT=true
 BACKUP_MAX_AGE_HOURS=30
 
+# Capacity guardrails are evaluated every minute. A disk enters warning or
+# critical state when either its used-percent ceiling or free-space floor is
+# crossed. /home and BACKUP_HOST_DIR are expected to be separate mounts.
+ROOT_DISK_WARNING_PERCENT=85
+ROOT_DISK_CRITICAL_PERCENT=90
+ROOT_DISK_WARNING_FREE_GIB=20
+ROOT_DISK_CRITICAL_FREE_GIB=10
+HOME_DISK_WARNING_PERCENT=85
+HOME_DISK_CRITICAL_PERCENT=90
+HOME_DISK_WARNING_FREE_GIB=20
+HOME_DISK_CRITICAL_FREE_GIB=10
+BACKUP_DISK_WARNING_PERCENT=85
+BACKUP_DISK_CRITICAL_PERCENT=90
+BACKUP_DISK_WARNING_FREE_GIB=20
+BACKUP_DISK_CRITICAL_FREE_GIB=10
+
 # 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.
 AI_PROVIDER_API_KEY_FILE=/etc/yuqei/secrets/ai-provider-api-key

yuqei-infra/deploy/preflight-production.py

@@ -29,6 +29,59 @@ LOOPBACK_VALUES = {"127.0.0.1", "::1", "localhost"}
 PROVIDER_ACCEPTANCE_RECEIPT_MAX_BYTES = 1024 * 1024
 PROVIDER_ACCEPTANCE_KEYRING_MAX_BYTES = 256 * 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"
+)
+FORBIDDEN_SUPPORT_ACCESS_ACTIVATION_KEYS = frozenset(
+    {
+        "CONTRACT_API_IDENTITY_EXECUTOR_ENABLED",
+        "CONTRACT_API_IDENTITY_WRITE_API_ENABLED",
+        "CONTRACT_API_PRIVILEGED_IDENTITY_EXECUTOR_ENABLED",
+        "CONTRACT_API_PRIVILEGED_IDENTITY_WRITE_API_ENABLED",
+        "CONTRACT_API_SUPPORT_ACCESS_ENABLED",
+        "CONTRACT_API_SUPPORT_ACCESS_EXECUTOR_ENABLED",
+        "CONTRACT_API_SUPPORT_ACCESS_GRANT_ENABLED",
+        "CONTRACT_API_SUPPORT_ACCESS_IMPERSONATION_ENABLED",
+        "CONTRACT_API_SUPPORT_ACCESS_REVOKE_ENABLED",
+    }
+)
+EXPECTED_LAUNCH_SCOPE = {
+    "schema_version": 1,
+    "decision_id": "support-access-first-release-exclusion-2026-08-12",
+    "release_scope": "v2-first-production-release",
+    "decision": "excluded",
+    "support_access": {
+        "included": False,
+        "enabled": False,
+        "kill_switch_engaged": True,
+        "executor_enabled": False,
+    },
+    "identity_executor": {"enabled": False},
+    "destructive_identity_write_api": {
+        "included": False,
+        "forbidden_operations": [
+            "execute",
+            "grant",
+            "revoke",
+            "impersonate",
+            "impersonation",
+        ],
+    },
+    "future_enablement_gates": [
+        "external_idp",
+        "mfa_step_up",
+        "session_device_binding",
+        "real_resource_toctou",
+        "cross_node_revocation_sla",
+        "kill_switch_drill",
+        "immutable_per_request_audit",
+        "privacy_legal_approval",
+    ],
+    "notes": [
+        "This is a first-release scope exclusion, not approval of support access.",
+        "No shared password, manual bypass, identity write API, or execute API may restore this capability.",
+    ],
+}
 
 
 @dataclass(frozen=True)
@@ -120,6 +173,8 @@ def validate_environment(values: dict[str, str], *, template_mode: bool = False)
     checks.append(_result("production_modes", not invalid_exact, "Production modes are strict." if not invalid_exact else f"Invalid production values: {', '.join(invalid_exact)}."))
     checks.append(_validate_identity_email_delivery(values))
     checks.append(_validate_ai_timeout_budget(values))
+    checks.append(_validate_capacity_thresholds(values))
+    checks.extend(_validate_launch_scope_exclusion(values))
 
     placeholders = [
         key
@@ -446,6 +501,71 @@ def _validate_ai_timeout_budget(values: dict[str, str]) -> Check:
     )
 
 
+def _validate_capacity_thresholds(values: dict[str, str]) -> Check:
+    prefixes = ("ROOT_DISK", "HOME_DISK", "BACKUP_DISK")
+    invalid: list[str] = []
+    for prefix in prefixes:
+        warning_percent = _parse_float(values.get(f"{prefix}_WARNING_PERCENT", ""))
+        critical_percent = _parse_float(values.get(f"{prefix}_CRITICAL_PERCENT", ""))
+        warning_free_gib = _parse_float(values.get(f"{prefix}_WARNING_FREE_GIB", ""))
+        critical_free_gib = _parse_float(values.get(f"{prefix}_CRITICAL_FREE_GIB", ""))
+        finite = all(
+            value is not None and math.isfinite(value)
+            for value in (warning_percent, critical_percent, warning_free_gib, critical_free_gib)
+        )
+        valid = bool(
+            finite
+            and warning_percent is not None
+            and critical_percent is not None
+            and warning_free_gib is not None
+            and critical_free_gib is not None
+            and 0 <= warning_percent < critical_percent <= 100
+            and 0 <= critical_free_gib <= warning_free_gib
+        )
+        if not valid:
+            invalid.append(prefix)
+    return _result(
+        "capacity_thresholds",
+        not invalid,
+        "Root, /home, and backup capacity thresholds are finite and ordered."
+        if not invalid
+        else f"Invalid or missing capacity threshold groups: {', '.join(invalid)}.",
+    )
+
+
+def _validate_launch_scope_exclusion(
+    values: dict[str, str],
+    *,
+    manifest_path: Path = LAUNCH_SCOPE_MANIFEST_PATH,
+) -> list[Check]:
+    try:
+        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    except (OSError, UnicodeError, json.JSONDecodeError):
+        manifest = None
+
+    active_flags = sorted(
+        key
+        for key in FORBIDDEN_SUPPORT_ACCESS_ACTIVATION_KEYS
+        if values.get(key, "").strip().lower() not in {"", "0", "false", "no", "off"}
+    )
+    return [
+        _result(
+            "launch_scope_support_access_exclusion",
+            manifest == EXPECTED_LAUNCH_SCOPE,
+            "The versioned launch scope excludes support access, keeps its kill switch engaged, and keeps identity execution disabled."
+            if manifest == EXPECTED_LAUNCH_SCOPE
+            else "The versioned launch-scope manifest is missing, invalid, or does not preserve the support-access exclusion.",
+        ),
+        _result(
+            "launch_scope_activation_flags",
+            not active_flags,
+            "No production activation flag enables support access, identity execution, or destructive identity-write APIs."
+            if not active_flags
+            else f"Forbidden support-access or identity-write activation flags are enabled: {', '.join(active_flags)}.",
+        ),
+    ]
+
+
 def _parse_float(value: str) -> float | None:
     try:
         return float(value)

yuqei-infra/observability/README.md

@@ -7,7 +7,9 @@
 - AI Platform health, schema-check, and `provider_runtime`;
 - worker runtime health;
 - latest production backup age;
-- root disk warning and critical thresholds.
+- root, `/home`, and backup-mount warning and critical thresholds;
+- both used-percent ceilings and remaining-free-space floors;
+- separate-mount enforcement for `/home` and the off-host backup path.
 
 Example:
 
@@ -15,13 +17,45 @@ Example:
 python3 yuqei-infra/observability/check-production-health.py \
   --edge-base-url https://api.contracts.example.com \
   --contract-api-base http://127.0.0.1:8100 \
+  --contract-proxy-key-file /etc/yuqei/monitoring/contract-proxy-key \
   --ai-api-base http://127.0.0.1:8101 \
   --ai-service-key-file /etc/yuqei/secrets/ai-platform-service-key \
   --backup-root /mnt/yuqei-v2-backups \
   --backup-max-age-hours 30 \
+  --root-disk-path / \
+  --home-disk-path /home \
+  --home-require-separate-mount \
+  --backup-disk-path /mnt/yuqei-v2-backups \
+  --backup-require-separate-mount \
   --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`.
 
+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
+is a warning or failure when either boundary is crossed. Missing paths and
+required paths that are not separate mounts fail closed.
+
+Warning and failure summaries are posted to the configured webhook when one is
+provided. A warning still exits `0` after successful delivery, a failed health
+check exits `1`, and an attempted webhook delivery failure exits `2` so
+systemd records the alerting failure. Every run writes
+`production-health-summary.json`, including the exact capacity policy and
+alert-delivery result, under
+`project-docs/qa-artifacts/v2-production-health/`.
+
+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.
+
+Acceptance commands (they do not start services or change traffic):
+
+```bash
+python3 -m pytest yuqei-infra/tests/test_observability_capacity.py -q
+systemd-analyze verify \
+  yuqei-infra/observability/yuqei-production-health.service \
+  yuqei-infra/observability/yuqei-production-health.timer
+```
+
 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

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import argparse
 import json
+import os
 import shutil
 import sys
 import time
@@ -104,22 +105,68 @@ def check_backup_freshness(backup_root: Path, *, max_age_hours: float) -> Health
     return HealthCheck("backup_freshness", "pass", f"Latest backup age_hours={age_hours:.2f}.")
 
 
-def check_disk(path: Path, *, warning_percent: float, critical_percent: float) -> HealthCheck:
+def validate_capacity_thresholds(
+    *,
+    warning_percent: float,
+    critical_percent: float,
+    warning_free_gib: float,
+    critical_free_gib: float,
+) -> None:
+    if not 0 <= warning_percent < critical_percent <= 100:
+        raise ValueError("Capacity percent thresholds must satisfy 0 <= warning < critical <= 100.")
+    if not 0 <= critical_free_gib <= warning_free_gib:
+        raise ValueError("Capacity free-space thresholds must satisfy 0 <= critical <= warning.")
+
+
+def check_disk(
+    path: Path,
+    *,
+    key: str = "disk_usage",
+    warning_percent: float,
+    critical_percent: float,
+    warning_free_gib: float = 0.0,
+    critical_free_gib: float = 0.0,
+    require_mount: bool = False,
+) -> HealthCheck:
+    try:
+        validate_capacity_thresholds(
+            warning_percent=warning_percent,
+            critical_percent=critical_percent,
+            warning_free_gib=warning_free_gib,
+            critical_free_gib=critical_free_gib,
+        )
+    except ValueError as exc:
+        return HealthCheck(key, "fail", str(exc))
+    if not path.is_dir():
+        return HealthCheck(key, "fail", f"Capacity path is unavailable: path={path}.")
+    if require_mount and not os.path.ismount(path):
+        return HealthCheck(key, "fail", f"Capacity path is not a separate mount: path={path}.")
     try:
         usage = shutil.disk_usage(path)
     except OSError as exc:
-        return HealthCheck("disk_usage", "fail", f"Disk usage is unavailable: {type(exc).__name__}.")
+        return HealthCheck(key, "fail", f"Disk usage is unavailable: {type(exc).__name__}.")
-    used_percent = (usage.used / usage.total * 100) if usage.total else 100.0
+    free_gib = usage.free / 1024 ** 3
-    if used_percent >= critical_percent:
+    # `shutil.disk_usage` reports `free` as space available to the caller,
+    # while ext4 may reserve blocks that are included in `total` but not
+    # available to non-root services. Use the non-reserved capacity base so
+    # the percentage matches `df` and does not understate a production alarm.
+    available_capacity = usage.used + usage.free
+    used_percent = (usage.used / available_capacity * 100) if available_capacity else 100.0
+    if used_percent >= critical_percent or (critical_free_gib > 0 and free_gib <= critical_free_gib):
         status = "fail"
-    elif used_percent >= warning_percent:
+    elif used_percent >= warning_percent or (warning_free_gib > 0 and free_gib <= warning_free_gib):
         status = "warning"
     else:
         status = "pass"
     return HealthCheck(
-        "disk_usage",
+        key,
         status,
-        f"used_percent={used_percent:.2f}, free_gib={usage.free / 1024 ** 3:.2f}.",
+        (
+            f"path={path}, used_percent={used_percent:.2f}, free_gib={free_gib:.2f}, "
+            f"warning_percent={warning_percent:.2f}, critical_percent={critical_percent:.2f}, "
+            f"warning_free_gib={warning_free_gib:.2f}, critical_free_gib={critical_free_gib:.2f}, "
+            f"require_mount={str(require_mount).lower()}."
+        ),
     )
 
 
@@ -152,13 +199,48 @@ def _parse_args() -> argparse.Namespace:
     parser.add_argument("--edge-base-url", required=True)
     parser.add_argument("--contract-api-base", default="http://127.0.0.1:8100")
     parser.add_argument("--ai-api-base", default="http://127.0.0.1:8101")
+    parser.add_argument("--contract-proxy-key-file", type=Path)
     parser.add_argument("--ai-service-key-file", type=Path)
     parser.add_argument("--tenant-id", default="default")
     parser.add_argument("--backup-root", type=Path, required=True)
     parser.add_argument("--backup-max-age-hours", type=float, default=30.0)
-    parser.add_argument("--disk-path", type=Path, default=Path("/"))
+    parser.add_argument("--root-disk-path", "--disk-path", dest="root_disk_path", type=Path, default=Path("/"))
-    parser.add_argument("--disk-warning-percent", type=float, default=80.0)
+    parser.add_argument(
-    parser.add_argument("--disk-critical-percent", type=float, default=90.0)
+        "--root-disk-warning-percent",
+        "--disk-warning-percent",
+        dest="root_disk_warning_percent",
+        type=float,
+        default=85.0,
+    )
+    parser.add_argument(
+        "--root-disk-critical-percent",
+        "--disk-critical-percent",
+        dest="root_disk_critical_percent",
+        type=float,
+        default=90.0,
+    )
+    parser.add_argument("--root-disk-warning-free-gib", type=float, default=20.0)
+    parser.add_argument("--root-disk-critical-free-gib", type=float, default=10.0)
+    parser.add_argument("--home-disk-path", type=Path, default=Path("/home"))
+    parser.add_argument("--home-disk-warning-percent", type=float, default=85.0)
+    parser.add_argument("--home-disk-critical-percent", type=float, default=90.0)
+    parser.add_argument("--home-disk-warning-free-gib", type=float, default=20.0)
+    parser.add_argument("--home-disk-critical-free-gib", type=float, default=10.0)
+    parser.add_argument(
+        "--home-require-separate-mount",
+        action=argparse.BooleanOptionalAction,
+        default=False,
+    )
+    parser.add_argument("--backup-disk-path", type=Path)
+    parser.add_argument("--backup-disk-warning-percent", type=float, default=85.0)
+    parser.add_argument("--backup-disk-critical-percent", type=float, default=90.0)
+    parser.add_argument("--backup-disk-warning-free-gib", type=float, default=20.0)
+    parser.add_argument("--backup-disk-critical-free-gib", type=float, default=10.0)
+    parser.add_argument(
+        "--backup-require-separate-mount",
+        action=argparse.BooleanOptionalAction,
+        default=True,
+    )
     parser.add_argument("--timeout-seconds", type=float, default=10.0)
     parser.add_argument("--webhook-url-file", type=Path)
     parser.add_argument("--artifact-dir", type=Path)
@@ -167,11 +249,42 @@ def _parse_args() -> argparse.Namespace:
 
 def main() -> int:
     args = _parse_args()
+    capacity_policies = [
+        {
+            "key": "disk_root",
+            "path": args.root_disk_path.resolve(),
+            "warning_percent": args.root_disk_warning_percent,
+            "critical_percent": args.root_disk_critical_percent,
+            "warning_free_gib": args.root_disk_warning_free_gib,
+            "critical_free_gib": args.root_disk_critical_free_gib,
+            "require_mount": False,
+        },
+        {
+            "key": "disk_home",
+            "path": args.home_disk_path.resolve(),
+            "warning_percent": args.home_disk_warning_percent,
+            "critical_percent": args.home_disk_critical_percent,
+            "warning_free_gib": args.home_disk_warning_free_gib,
+            "critical_free_gib": args.home_disk_critical_free_gib,
+            "require_mount": args.home_require_separate_mount,
+        },
+        {
+            "key": "disk_backup",
+            "path": (args.backup_disk_path or args.backup_root).resolve(),
+            "warning_percent": args.backup_disk_warning_percent,
+            "critical_percent": args.backup_disk_critical_percent,
+            "warning_free_gib": args.backup_disk_warning_free_gib,
+            "critical_free_gib": args.backup_disk_critical_free_gib,
+            "require_mount": args.backup_require_separate_mount,
+        },
+    ]
     try:
+        contract_proxy_key = read_secret_file(args.contract_proxy_key_file)
         service_key = read_secret_file(args.ai_service_key_file)
     except OSError:
-        print("PRODUCTION_HEALTH_FAIL ai_service_key_file_unreadable")
+        print("PRODUCTION_HEALTH_FAIL health_secret_file_unreadable")
         return 2
+    contract_headers = {"X-Contract-Proxy-Key": contract_proxy_key} if contract_proxy_key else None
     ai_headers = {"X-Tenant-Id": args.tenant_id}
     if service_key:
         ai_headers["X-AI-Platform-Service-Key"] = service_key
@@ -189,7 +302,12 @@ def main() -> int:
         timeout=args.timeout_seconds,
     )
     checks.append(contract_health)
-    contract_schema, _ = check_schema("contract_schema", args.contract_api_base, headers=None, timeout=args.timeout_seconds)
+    contract_schema, _ = check_schema(
+        "contract_schema",
+        args.contract_api_base,
+        headers=contract_headers,
+        timeout=args.timeout_seconds,
+    )
     checks.append(contract_schema)
     ai_health, _ = check_json_endpoint(
         "ai_api_health",
@@ -207,13 +325,7 @@ def main() -> int:
     )
     checks.append(worker_health)
     checks.append(check_backup_freshness(args.backup_root.resolve(), max_age_hours=args.backup_max_age_hours))
-    checks.append(
+    checks.extend(check_disk(**policy) for policy in capacity_policies)
-        check_disk(
-            args.disk_path.resolve(),
-            warning_percent=args.disk_warning_percent,
-            critical_percent=args.disk_critical_percent,
-        )
-    )
 
     failed = [check for check in checks if check.status == "fail"]
     warnings = [check for check in checks if check.status == "warning"]
@@ -223,22 +335,39 @@ def main() -> int:
         "checked_at": checked_at.isoformat(),
         "failed_checks": [check.key for check in failed],
         "warning_checks": [check.key for check in warnings],
+        "capacity_policy": [
+            {**policy, "path": str(policy["path"])} for policy in capacity_policies
+        ],
+        "alert_delivery": {
+            "configured": bool(args.webhook_url_file),
+            "required": bool(failed or warnings),
+            "attempted": False,
+            "delivered": False,
+            "error": None,
+        },
         "checks": [asdict(check) for check in checks],
     }
     artifact_dir = (args.artifact_dir or DEFAULT_ARTIFACT_ROOT / checked_at.strftime("%Y-%m-%dT%H-%M-%S")).resolve()
     artifact_dir.mkdir(parents=True, exist_ok=True)
     summary_path = artifact_dir / "production-health-summary.json"
-    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
+    alert_delivery_failed = False
-
+    if (failed or warnings) and args.webhook_url_file:
-    if failed 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)
+        else:
+            summary["alert_delivery"]["delivered"] = True
+    summary_path.write_text(json.dumps(summary, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
     print(
         f"PRODUCTION_HEALTH_{summary['status'].upper()} "
         f"failed={len(failed)} warnings={len(warnings)} artifact={summary_path}"
     )
+    if alert_delivery_failed:
+        return 2
     return 1 if failed else 0
 
 

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

@@ -5,15 +5,15 @@ Wants=network-online.target
 
 [Service]
 Type=oneshot
-User=yuqei-ops
+User=nobody
-Group=yuqei-ops
+Group=nobody
 WorkingDirectory=/opt/apps/ai-contract-platform-v2-production
-ExecStart=/usr/bin/python3 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} --ai-api-base http://127.0.0.1:${AI_PLATFORM_API_HOST_PORT} --ai-service-key-file /etc/yuqei/secrets/ai-platform-service-key --backup-root ${BACKUP_HOST_DIR} --backup-max-age-hours ${BACKUP_MAX_AGE_HOURS} --webhook-url-file /etc/yuqei/secrets/alert-webhook-url
+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
 EnvironmentFile=/etc/yuqei/production.env
 NoNewPrivileges=true
 PrivateTmp=true
 ProtectSystem=strict
-ProtectHome=true
+ProtectHome=read-only
 ReadWritePaths=/opt/apps/ai-contract-platform-v2-production/project-docs/qa-artifacts
 
 [Install]

yuqei-infra/tests/test_observability_capacity.py

@@ -0,0 +1,351 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from pathlib import Path
+from types import ModuleType, SimpleNamespace
+
+import pytest
+
+
+REPO_ROOT = Path(__file__).resolve().parents[2]
+GIB = 1024**3
+
+
+def load_script(name: str, relative_path: str) -> ModuleType:
+    path = REPO_ROOT / relative_path
+    spec = importlib.util.spec_from_file_location(name, path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+health = load_script(
+    "yuqei_observability_capacity",
+    "yuqei-infra/observability/check-production-health.py",
+)
+preflight = load_script(
+    "yuqei_observability_preflight",
+    "yuqei-infra/deploy/preflight-production.py",
+)
+
+
+@pytest.mark.parametrize(
+    ("used_gib", "free_gib", "expected"),
+    [
+        (79, 21, "pass"),
+        (80, 20, "warning"),
+        (89, 9, "fail"),
+        (90, 10, "fail"),
+    ],
+)
+def test_capacity_check_applies_percent_ceiling_and_free_space_floor(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+    used_gib: int,
+    free_gib: int,
+    expected: str,
+) -> None:
+    total_gib = used_gib + free_gib
+    monkeypatch.setattr(
+        health.shutil,
+        "disk_usage",
+        lambda _path: SimpleNamespace(total=total_gib * GIB, used=used_gib * GIB, free=free_gib * GIB),
+    )
+
+    result = health.check_disk(
+        tmp_path,
+        key="disk_root",
+        warning_percent=80,
+        critical_percent=90,
+        warning_free_gib=20,
+        critical_free_gib=10,
+    )
+
+    assert result.status == expected
+    assert "warning_percent=80.00" in result.message
+    assert "critical_free_gib=10.00" in result.message
+
+
+def test_required_capacity_mount_fails_closed(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
+    monkeypatch.setattr(health.os.path, "ismount", lambda _path: False)
+
+    result = health.check_disk(
+        tmp_path,
+        key="disk_backup",
+        warning_percent=80,
+        critical_percent=90,
+        warning_free_gib=20,
+        critical_free_gib=10,
+        require_mount=True,
+    )
+
+    assert result.status == "fail"
+    assert "not a separate mount" in result.message
+
+
+def test_capacity_percent_excludes_filesystem_reserved_blocks(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    # total includes a reserved-block pool that is unavailable to the service.
+    monkeypatch.setattr(
+        health.shutil,
+        "disk_usage",
+        lambda _path: SimpleNamespace(total=100 * GIB, used=80 * GIB, free=10 * GIB),
+    )
+
+    result = health.check_disk(
+        tmp_path,
+        key="disk_root",
+        warning_percent=85,
+        critical_percent=90,
+        warning_free_gib=5,
+        critical_free_gib=2,
+    )
+
+    assert result.status == "warning"
+    assert "used_percent=88.89" in result.message
+
+
+@pytest.mark.parametrize(
+    ("warning_percent", "critical_percent", "warning_free_gib", "critical_free_gib"),
+    [
+        (90, 80, 20, 10),
+        (80, 101, 20, 10),
+        (80, 90, 10, 20),
+        (-1, 90, 20, 10),
+    ],
+)
+def test_invalid_capacity_threshold_order_fails_closed(
+    tmp_path: Path,
+    warning_percent: float,
+    critical_percent: float,
+    warning_free_gib: float,
+    critical_free_gib: float,
+) -> None:
+    result = health.check_disk(
+        tmp_path,
+        key="disk_root",
+        warning_percent=warning_percent,
+        critical_percent=critical_percent,
+        warning_free_gib=warning_free_gib,
+        critical_free_gib=critical_free_gib,
+    )
+
+    assert result.status == "fail"
+    assert "thresholds must satisfy" in result.message
+
+
+def _main_args(tmp_path: Path) -> SimpleNamespace:
+    return SimpleNamespace(
+        edge_base_url="https://api.contracts.example.com",
+        contract_api_base="http://127.0.0.1:8100",
+        ai_api_base="http://127.0.0.1:8101",
+        contract_proxy_key_file=None,
+        ai_service_key_file=None,
+        tenant_id="default",
+        backup_root=tmp_path,
+        backup_max_age_hours=30,
+        root_disk_path=tmp_path,
+        root_disk_warning_percent=80,
+        root_disk_critical_percent=90,
+        root_disk_warning_free_gib=20,
+        root_disk_critical_free_gib=10,
+        home_disk_path=tmp_path,
+        home_disk_warning_percent=80,
+        home_disk_critical_percent=90,
+        home_disk_warning_free_gib=20,
+        home_disk_critical_free_gib=10,
+        home_require_separate_mount=True,
+        backup_disk_path=tmp_path,
+        backup_disk_warning_percent=80,
+        backup_disk_critical_percent=90,
+        backup_disk_warning_free_gib=20,
+        backup_disk_critical_free_gib=10,
+        backup_require_separate_mount=True,
+        timeout_seconds=1,
+        webhook_url_file=tmp_path / "webhook-url",
+        artifact_dir=tmp_path / "artifact",
+    )
+
+
+def _mock_non_capacity_checks(monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(
+        health,
+        "check_json_endpoint",
+        lambda key, *_args, **_kwargs: (health.HealthCheck(key, "pass", "ok"), {"status": "ok"}),
+    )
+    monkeypatch.setattr(
+        health,
+        "check_schema",
+        lambda key, *_args, **_kwargs: (
+            health.HealthCheck(key, "pass", "ok"),
+            {"checks": [{"name": "provider_runtime", "status": "ok"}]},
+        ),
+    )
+    monkeypatch.setattr(
+        health,
+        "check_backup_freshness",
+        lambda *_args, **_kwargs: health.HealthCheck("backup_freshness", "pass", "ok"),
+    )
+    monkeypatch.setattr(
+        health,
+        "check_disk",
+        lambda **policy: health.HealthCheck(
+            policy["key"],
+            "warning" if policy["key"] == "disk_home" else "pass",
+            "simulated",
+        ),
+    )
+
+
+def test_warning_dispatches_webhook_and_records_acceptance_artifact(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    args = _main_args(tmp_path)
+    delivered: list[str] = []
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+    monkeypatch.setattr(
+        health,
+        "send_webhook",
+        lambda _path, summary, **_kwargs: delivered.append(summary["status"]),
+    )
+
+    exit_code = health.main()
+    summary = json.loads((args.artifact_dir / "production-health-summary.json").read_text(encoding="utf-8"))
+
+    assert exit_code == 0
+    assert delivered == ["warning"]
+    assert summary["warning_checks"] == ["disk_home"]
+    assert [item["key"] for item in summary["capacity_policy"]] == [
+        "disk_root",
+        "disk_home",
+        "disk_backup",
+    ]
+    assert summary["alert_delivery"] == {
+        "configured": True,
+        "required": True,
+        "attempted": True,
+        "delivered": True,
+        "error": None,
+    }
+
+
+def test_webhook_delivery_failure_is_visible_to_systemd(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    args = _main_args(tmp_path)
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+
+    def fail_delivery(*_args: object, **_kwargs: object) -> None:
+        raise TimeoutError("simulated")
+
+    monkeypatch.setattr(health, "send_webhook", fail_delivery)
+
+    exit_code = health.main()
+    summary = json.loads((args.artifact_dir / "production-health-summary.json").read_text(encoding="utf-8"))
+
+    assert exit_code == 2
+    assert summary["alert_delivery"]["attempted"] is True
+    assert summary["alert_delivery"]["delivered"] is False
+    assert summary["alert_delivery"]["error"] == "TimeoutError"
+
+
+def test_failed_capacity_check_dispatches_webhook_and_exits_one(
+    monkeypatch: pytest.MonkeyPatch,
+    tmp_path: Path,
+) -> None:
+    args = _main_args(tmp_path)
+    delivered: list[str] = []
+    monkeypatch.setattr(health, "_parse_args", lambda: args)
+    _mock_non_capacity_checks(monkeypatch)
+    monkeypatch.setattr(
+        health,
+        "check_disk",
+        lambda **policy: health.HealthCheck(
+            policy["key"],
+            "fail" if policy["key"] == "disk_backup" else "pass",
+            "simulated",
+        ),
+    )
+    monkeypatch.setattr(
+        health,
+        "send_webhook",
+        lambda _path, summary, **_kwargs: delivered.append(summary["status"]),
+    )
+
+    exit_code = health.main()
+
+    assert exit_code == 1
+    assert delivered == ["fail"]
+
+
+def test_healthy_run_does_not_dispatch_webhook(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,
+        "check_disk",
+        lambda **policy: health.HealthCheck(policy["key"], "pass", "simulated"),
+    )
+
+    def unexpected_delivery(*_args: object, **_kwargs: object) -> None:
+        pytest.fail("healthy runs must not dispatch alerts")
+
+    monkeypatch.setattr(health, "send_webhook", unexpected_delivery)
+
+    exit_code = health.main()
+    summary = json.loads((args.artifact_dir / "production-health-summary.json").read_text(encoding="utf-8"))
+
+    assert exit_code == 0
+    assert summary["alert_delivery"]["required"] is False
+    assert summary["alert_delivery"]["attempted"] is False
+
+
+def test_preflight_rejects_missing_or_inverted_capacity_thresholds() -> None:
+    values = {
+        "ROOT_DISK_WARNING_PERCENT": "80",
+        "ROOT_DISK_CRITICAL_PERCENT": "90",
+        "ROOT_DISK_WARNING_FREE_GIB": "20",
+        "ROOT_DISK_CRITICAL_FREE_GIB": "10",
+        "HOME_DISK_WARNING_PERCENT": "95",
+        "HOME_DISK_CRITICAL_PERCENT": "90",
+        "HOME_DISK_WARNING_FREE_GIB": "20",
+        "HOME_DISK_CRITICAL_FREE_GIB": "10",
+        "BACKUP_DISK_WARNING_PERCENT": "80",
+        "BACKUP_DISK_CRITICAL_PERCENT": "90",
+        "BACKUP_DISK_WARNING_FREE_GIB": "20",
+    }
+
+    result = preflight._validate_capacity_thresholds(values)
+
+    assert result.status == "fail"
+    assert "HOME_DISK" in result.message
+    assert "BACKUP_DISK" in result.message
+
+
+def test_systemd_timer_and_service_cover_all_capacity_paths() -> None:
+    unit = (REPO_ROOT / "yuqei-infra/observability/yuqei-production-health.service").read_text(
+        encoding="utf-8"
+    )
+    timer = (REPO_ROOT / "yuqei-infra/observability/yuqei-production-health.timer").read_text(
+        encoding="utf-8"
+    )
+
+    assert "OnUnitActiveSec=1min" in timer
+    assert "Persistent=true" in timer
+    assert "ProtectHome=read-only" in unit
+    assert "--root-disk-path /" in unit
+    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-require-separate-mount" in unit

yuqei-infra/tests/test_production_release_tools.py

@@ -144,6 +144,71 @@ def test_production_preflight_accepts_strict_configuration() -> None:
     assert [check.key for check in checks if check.status == "fail"] == []
 
 
+def test_production_launch_scope_excludes_support_access_and_identity_execution() -> None:
+    manifest = json.loads(preflight.LAUNCH_SCOPE_MANIFEST_PATH.read_text(encoding="utf-8"))
+
+    assert manifest == preflight.EXPECTED_LAUNCH_SCOPE
+    assert manifest["decision"] == "excluded"
+    assert manifest["support_access"] == {
+        "included": False,
+        "enabled": False,
+        "kill_switch_engaged": True,
+        "executor_enabled": False,
+    }
+    assert manifest["identity_executor"] == {"enabled": False}
+    assert manifest["destructive_identity_write_api"]["included"] is False
+    assert set(manifest["destructive_identity_write_api"]["forbidden_operations"]) == {
+        "execute",
+        "grant",
+        "revoke",
+        "impersonate",
+        "impersonation",
+    }
+
+
+@pytest.mark.parametrize(
+    "activation_key",
+    sorted(preflight.FORBIDDEN_SUPPORT_ACCESS_ACTIVATION_KEYS),
+)
+def test_production_preflight_rejects_support_access_or_identity_write_activation(
+    activation_key: str,
+) -> None:
+    values = production_values()
+    values[activation_key] = "true"
+
+    checks = {check.key: check for check in preflight.validate_environment(values)}
+
+    assert checks["launch_scope_activation_flags"].status == "fail"
+    assert activation_key in checks["launch_scope_activation_flags"].message
+
+
+@pytest.mark.parametrize("artifact_state", ["missing", "tampered", "invalid"])
+def test_production_preflight_fails_closed_on_launch_scope_artifact_drift(
+    tmp_path: Path,
+    artifact_state: str,
+) -> None:
+    manifest_path = tmp_path / "launch-scope.json"
+    if artifact_state == "tampered":
+        tampered = dict(preflight.EXPECTED_LAUNCH_SCOPE)
+        tampered["support_access"] = {
+            **tampered["support_access"],
+            "enabled": True,
+        }
+        manifest_path.write_text(json.dumps(tampered), encoding="utf-8")
+    elif artifact_state == "invalid":
+        manifest_path.write_text("not-json", encoding="utf-8")
+
+    checks = {
+        check.key: check
+        for check in preflight._validate_launch_scope_exclusion(
+            production_values(),
+            manifest_path=manifest_path,
+        )
+    }
+
+    assert checks["launch_scope_support_access_exclusion"].status == "fail"
+
+
 @pytest.mark.parametrize(
     ("contract_timeout", "provider_timeout"),
     [("30", "30"), ("0", "30"), ("601", "30"), ("NaN", "30"), ("60", "inf")],