milestone 19 &20

MILESTONES.md

@@ -42,7 +42,7 @@ This is the working delivery tracker for GuestOps Web. Update a milestone when i
 | 15 | Knowledge, AI, and FAQ activation | B | In progress | Owners can run a bounded no-send batch evaluation against current FAQ rules and approved knowledge, with false-positive/negative results and documented zero-error activation thresholds and stop conditions. Curate hotel-specific cases, evaluate AI suggestions separately, train staff, name monitoring/rollback owners, and retain staged-activation evidence. |
 | 16 | Identity, preferences, and privacy | B/C | In progress | Login throttling now uses the client address only after one-hop processing from the explicitly trusted reverse proxy. Finish privacy/retention decisions, preference coverage, identity acceptance and audit review. |
 | 17 | Inbox usability and desktop parity | B | In progress | The inbox now uses tenant-scoped stable cursor pagination in pages of 50 and protects unsaved drafts during navigation, conversation selection, filtering and search changes. Inbox, activity, mailbox-health and FAQ-history timestamps use the saved hotel timezone; finish the remaining secondary screens and agreed desktop-parity acceptance. |
-| 18 | Pilot, capacity, and release approval | B/C | In progress | A bounded read-only sandbox capacity probe and release-bound pilot exit checklist are implemented. Agree targets, run the probe with host monitoring, complete the supervised pilot and incident exercises, resolve or explicitly contain findings, and retain separate hotel-owner and technical go/no-go decisions. |
+| 18 | Pilot, capacity, and release approval | B/C | In progress | A bounded read-only sandbox capacity probe, machine-validated pilot decision and Gate B/C incident-exercise record are implemented. Agree targets, run the probe with host monitoring, complete the supervised pilot and incident exercises, resolve or explicitly contain findings, and retain separate hotel-owner and technical go/no-go decisions. |
 
 ## Delivery sequence
 

RELEASE_NOTES.md

@@ -11,7 +11,7 @@ The reviewed candidate is now promoted into the local `main` history. It is not
 - FAQ automation controls, team invitations, password recovery, and stronger Google connection recovery.
 - No-send FAQ batch evaluation with false-positive and false-negative reporting before activation.
 - Stable tenant-scoped inbox pagination beyond the former 500-message view, stronger unsaved-draft navigation guards, and hotel-timezone inbox timestamps.
-- A bounded read-only sandbox capacity probe and release-bound supervised-pilot approval checklist.
+- A bounded read-only sandbox capacity probe plus machine-validated, release-bound supervised-pilot and incident-exercise records.
 - Backup, restore, opt-in systemd scheduling, deployment, persistence-drill, diagnostic, release-evidence, Google acceptance-record validation and rollback tooling.
 
 These capabilities still require their separately documented provider, host and operational acceptance. Google, PMS and payment-provider acceptance is not established by local automated tests.

deploy/incident-exercise.example.json

@@ -0,0 +1,38 @@
+{
+  "schemaVersion": 1,
+  "system": "guestops-incident-exercise",
+  "targetGate": "B",
+  "dataClassification": "synthetic-only",
+  "releaseCommit": "0000000000000000000000000000000000000000",
+  "releaseRecordSha256": "0000000000000000000000000000000000000000000000000000000000000000",
+  "environment": "https://sandbox-guestops.example.invalid",
+  "operator": "Exercise operator",
+  "incidentCommander": "Incident commander",
+  "reviewedBy": "Independent reviewer",
+  "startedAt": "2026-09-29T09:00:00Z",
+  "endedAt": "2026-09-29T10:00:00Z",
+  "reviewedAt": "2026-09-29T11:00:00Z",
+  "targetsMinutes": {
+    "detection": 10,
+    "containment": 20,
+    "recovery": 60
+  },
+  "observedMinutes": {
+    "detection": 0,
+    "containment": 0,
+    "recovery": 0
+  },
+  "scenarios": [
+    {"id": "alert-and-escalate", "status": "not-run", "evidence": []},
+    {"id": "controlled-recovery", "status": "not-run", "evidence": []},
+    {"id": "disable-worker-writes", "status": "not-run", "evidence": []},
+    {"id": "google-uncertain-send", "status": "not-run", "evidence": []},
+    {"id": "image-rollback", "status": "not-run", "evidence": []},
+    {"id": "restore-readiness", "status": "not-run", "evidence": []}
+  ],
+  "postExerciseState": {
+    "externalWrites": "disabled",
+    "faqMode": "off",
+    "unresolvedOperations": 0
+  }
+}

deploy/incident_exercise.py

@@ -0,0 +1,132 @@
+#!/usr/bin/env python3
+"""Validate a restricted GuestOps incident and rollback exercise record."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+from pathlib import Path
+import re
+from urllib.parse import urlparse
+
+
+GATE_B_SCENARIOS = {
+    "alert-and-escalate",
+    "disable-worker-writes",
+    "google-uncertain-send",
+    "restore-readiness",
+    "image-rollback",
+    "controlled-recovery",
+}
+GATE_C_SCENARIOS = GATE_B_SCENARIOS | {
+    "pms-ambiguous-write",
+    "payment-ambiguous-create",
+}
+
+
+def require(condition: bool, message: str) -> None:
+    if not condition:
+        raise ValueError(message)
+
+
+def utc_timestamp(value: object, field: str) -> dt.datetime:
+    require(isinstance(value, str) and value.endswith("Z"), f"{field} must be a UTC timestamp ending in Z.")
+    try:
+        parsed = dt.datetime.fromisoformat(value.removesuffix("Z") + "+00:00")
+    except ValueError as error:
+        raise ValueError(f"{field} is not a valid timestamp.") from error
+    require(parsed.tzinfo == dt.timezone.utc, f"{field} must be UTC.")
+    return parsed
+
+
+def safe_name(value: object, field: str) -> str:
+    name = str(value or "").strip()
+    require(2 <= len(name) <= 120 and "@" not in name, f"{field} requires a name without an email address.")
+    return name
+
+
+def validate(record: object) -> None:
+    require(isinstance(record, dict), "Exercise record must be a JSON object.")
+    require(record.get("schemaVersion") == 1, "Unsupported exercise record schema.")
+    require(record.get("system") == "guestops-incident-exercise",
+            "Exercise record system must be guestops-incident-exercise.")
+    gate = record.get("targetGate")
+    require(gate in ("B", "C"), "targetGate must be B or C.")
+    require(record.get("dataClassification") == "synthetic-only",
+            "Incident exercises must use synthetic data only.")
+    require(re.fullmatch(r"[0-9a-f]{40}", str(record.get("releaseCommit", ""))) is not None,
+            "releaseCommit must be a full lowercase Git SHA.")
+    require(re.fullmatch(r"[0-9a-f]{64}", str(record.get("releaseRecordSha256", ""))) is not None,
+            "releaseRecordSha256 must be a SHA-256 digest.")
+
+    environment = str(record.get("environment", ""))
+    parsed_url = urlparse(environment)
+    require(parsed_url.scheme == "https" and parsed_url.hostname and parsed_url.path in ("", "/")
+            and not parsed_url.query and not parsed_url.fragment and parsed_url.username is None
+            and parsed_url.password is None,
+            "environment must be an HTTPS origin without credentials, path, query or fragment.")
+
+    operator = safe_name(record.get("operator"), "operator")
+    commander = safe_name(record.get("incidentCommander"), "incidentCommander")
+    reviewer = safe_name(record.get("reviewedBy"), "reviewedBy")
+    require(len({operator.casefold(), commander.casefold(), reviewer.casefold()}) == 3,
+            "operator, incidentCommander and reviewedBy must be different people.")
+
+    started = utc_timestamp(record.get("startedAt"), "startedAt")
+    ended = utc_timestamp(record.get("endedAt"), "endedAt")
+    reviewed = utc_timestamp(record.get("reviewedAt"), "reviewedAt")
+    require(started <= ended <= reviewed, "Exercise timestamps are out of order.")
+
+    targets = record.get("targetsMinutes")
+    observed = record.get("observedMinutes")
+    require(isinstance(targets, dict) and isinstance(observed, dict),
+            "targetsMinutes and observedMinutes are required.")
+    metric_keys = {"detection", "containment", "recovery"}
+    require(set(targets) == metric_keys and set(observed) == metric_keys,
+            "Timing records require exactly detection, containment and recovery.")
+    for metric in sorted(metric_keys):
+        target = targets[metric]
+        actual = observed[metric]
+        require(isinstance(target, (int, float)) and not isinstance(target, bool) and 0 < target <= 1440,
+                f"{metric} target must be greater than zero and no more than 1440 minutes.")
+        require(isinstance(actual, (int, float)) and not isinstance(actual, bool) and 0 <= actual <= target,
+                f"Observed {metric} time must meet its pre-agreed target.")
+
+    scenarios = record.get("scenarios")
+    require(isinstance(scenarios, list), "scenarios must be a list.")
+    ids = [item.get("id") for item in scenarios if isinstance(item, dict)]
+    required = GATE_C_SCENARIOS if gate == "C" else GATE_B_SCENARIOS
+    require(len(ids) == len(scenarios) and len(ids) == len(set(ids)) and set(ids) == required,
+            f"Gate {gate} requires the exact incident scenario set.")
+    for item in scenarios:
+        scenario_id = item["id"]
+        require(item.get("status") == "pass", f"Scenario {scenario_id} has not passed.")
+        evidence = item.get("evidence")
+        require(isinstance(evidence, list) and 1 <= len(evidence) <= 10 and all(
+            isinstance(value, str) and 3 <= len(value) <= 200 and "@" not in value
+            for value in evidence
+        ), f"Scenario {scenario_id} requires safe opaque evidence references without email addresses.")
+
+    final_state = record.get("postExerciseState")
+    require(isinstance(final_state, dict) and final_state == {
+        "externalWrites": "disabled",
+        "faqMode": "off",
+        "unresolvedOperations": 0,
+    }, "Exercise must end with writes disabled, FAQ mode off and no unresolved operations.")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("record", type=Path)
+    args = parser.parse_args()
+    validate(json.loads(args.record.read_text(encoding="utf-8")))
+    print("Incident exercise record is structurally complete and passed. This validates the record, not its restricted evidence.")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Incident exercise record rejected: {error}", file=__import__("sys").stderr)
+        raise SystemExit(1)

deploy/pilot-approval.example.json

@@ -0,0 +1,22 @@
+{
+  "approvals": {
+    "hotelOwner": {"approvedAt": "2026-01-01T00:00:00Z", "name": "REPLACE"},
+    "technicalOwner": {"approvedAt": "2026-01-01T00:00:00Z", "name": "REPLACE"}
+  },
+  "capacity": {
+    "observedConcurrency": 0,
+    "observedErrorRate": 1,
+    "observedP95Ms": 0,
+    "reportSha256": "0000000000000000000000000000000000000000000000000000000000000000",
+    "targetConcurrency": 1,
+    "targetErrorRate": 0,
+    "targetP95Ms": 0
+  },
+  "decidedAt": "2026-01-01T00:00:00Z",
+  "decision": "pending",
+  "evidence": [],
+  "releaseCommit": "0000000000000000000000000000000000000000",
+  "releaseRecordSha256": "0000000000000000000000000000000000000000000000000000000000000000",
+  "schemaVersion": 1,
+  "targetGate": "B"
+}

deploy/pilot_approval.py

@@ -0,0 +1,112 @@
+#!/usr/bin/env python3
+"""Validate a GuestOps supervised-pilot go/no-go record."""
+
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import json
+from pathlib import Path
+import re
+
+
+GATE_B = {
+    "release-ci", "debian-host", "persistence", "backup-restore", "google-mailbox",
+    "automation", "identity-privacy", "inbox-usability", "capacity",
+    "incident-support", "pilot-findings",
+}
+GATE_C = GATE_B | {"pms-provider", "payment-provider"}
+
+
+def require(condition: bool, message: str) -> None:
+    if not condition:
+        raise ValueError(message)
+
+
+def timestamp(value: object, field: str) -> dt.datetime:
+    require(isinstance(value, str) and value.endswith("Z"), f"{field} must be a UTC timestamp ending in Z.")
+    try:
+        parsed = dt.datetime.fromisoformat(value[:-1] + "+00:00")
+    except ValueError as error:
+        raise ValueError(f"{field} is not a valid timestamp.") from error
+    return parsed
+
+
+def validate(record: object) -> None:
+    require(isinstance(record, dict), "Approval record must be a JSON object.")
+    require(record.get("schemaVersion") == 1, "Unsupported approval schema.")
+    gate = record.get("targetGate")
+    require(gate in ("B", "C"), "targetGate must be B or C.")
+    require(record.get("decision") == "approved", "Only an explicit approved decision passes validation.")
+    require(re.fullmatch(r"[0-9a-f]{40}", str(record.get("releaseCommit", ""))) is not None,
+            "releaseCommit must be a full lowercase Git SHA.")
+    require(re.fullmatch(r"[0-9a-f]{64}", str(record.get("releaseRecordSha256", ""))) is not None,
+            "releaseRecordSha256 must be a SHA-256 digest.")
+    decided = timestamp(record.get("decidedAt"), "decidedAt")
+
+    approvals = record.get("approvals")
+    require(isinstance(approvals, dict) and set(approvals) == {"hotelOwner", "technicalOwner"},
+            "Separate hotelOwner and technicalOwner approvals are required.")
+    approver_names = []
+    for role, approval in approvals.items():
+        name = str(approval.get("name", "")).strip() if isinstance(approval, dict) else ""
+        require(2 <= len(name) <= 120 and "@" not in name,
+                f"{role} approval requires a named owner without an email address.")
+        approver_names.append(name.casefold())
+        require(timestamp(approval.get("approvedAt"), f"{role}.approvedAt") <= decided,
+                f"{role} approval cannot occur after the decision.")
+    require(len(set(approver_names)) == 2, "hotelOwner and technicalOwner must be different people.")
+
+    evidence = record.get("evidence")
+    require(isinstance(evidence, list), "evidence must be a list.")
+    ids = [item.get("id") for item in evidence if isinstance(item, dict)]
+    required = GATE_C if gate == "C" else GATE_B
+    require(len(ids) == len(evidence) and len(ids) == len(set(ids)) and set(ids) == required,
+            f"Gate {gate} requires the exact evidence set.")
+    for item in evidence:
+        item_id = item["id"]
+        status = item.get("status")
+        require(status in ("pass", "contained"), f"Evidence {item_id} must pass or have approved containment.")
+        references = item.get("references")
+        require(isinstance(references, list) and 1 <= len(references) <= 10 and all(isinstance(value, str) and 3 <= len(value) <= 200 and "@" not in value for value in references),
+                f"Evidence {item_id} requires safe opaque references without email addresses.")
+        if status == "contained":
+            containment = item.get("containment")
+            require(isinstance(containment, dict), f"Evidence {item_id} requires containment details.")
+            require(2 <= len(str(containment.get("owner", ""))) <= 120, f"Evidence {item_id} containment requires an owner.")
+            require(timestamp(containment.get("expiresAt"), f"{item_id}.containment.expiresAt") > decided,
+                    f"Evidence {item_id} containment must expire after the decision.")
+            require(5 <= len(str(containment.get("rollbackTrigger", ""))) <= 300,
+                    f"Evidence {item_id} containment requires a rollback trigger.")
+
+    capacity = record.get("capacity")
+    require(isinstance(capacity, dict), "Capacity thresholds and observations are required.")
+    require(re.fullmatch(r"[0-9a-f]{64}", str(capacity.get("reportSha256", ""))) is not None,
+            "capacity.reportSha256 must identify the retained probe report.")
+    for observed, target in (("observedP95Ms", "targetP95Ms"), ("observedErrorRate", "targetErrorRate")):
+        values = (capacity.get(observed), capacity.get(target))
+        require(all(isinstance(value, (int, float)) and not isinstance(value, bool) for value in values)
+                and 0 <= capacity[observed] <= capacity[target],
+                f"Capacity {observed} must be within its approved {target}.")
+    require(capacity["targetP95Ms"] > 0, "Capacity targetP95Ms must be positive.")
+    require(capacity["targetErrorRate"] <= 1, "Capacity targetErrorRate must be a ratio no greater than 1.")
+    concurrency = (capacity.get("observedConcurrency"), capacity.get("targetConcurrency"))
+    require(all(isinstance(value, int) and not isinstance(value, bool) for value in concurrency)
+            and capacity["observedConcurrency"] >= capacity["targetConcurrency"] > 0,
+            "Observed concurrency must meet the approved positive target.")
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("record", type=Path)
+    args = parser.parse_args()
+    validate(json.loads(args.record.read_text(encoding="utf-8")))
+    print("Pilot approval record is structurally complete and approved. This validates the record, not its restricted evidence.")
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except (OSError, ValueError, json.JSONDecodeError) as error:
+        print(f"Pilot approval rejected: {error}", file=__import__("sys").stderr)
+        raise SystemExit(1)

docs/incident-exercise.md

@@ -0,0 +1,28 @@
+# Incident and rollback exercise
+
+Run this exercise on the approved sandbox release with synthetic records only. Agree detection, containment and recovery targets before starting. The operator, incident commander and independent reviewer must be different people. Keep actual guest addresses, credentials, tokens, provider payloads and raw incident evidence in the restricted acceptance store.
+
+## Gate B scenarios
+
+| Record ID | Exercise | Passing evidence |
+| --- | --- | --- |
+| `alert-and-escalate` | Introduce an approved synthetic failure and use normal monitoring and support routes. | The alert is detected, the incident commander is engaged and timestamps meet the agreed detection target. |
+| `disable-worker-writes` | Use the documented controls to stop worker-driven provider writes and FAQ automation. | Pending work does not reach a provider, controls remain effective across a worker restart and staff can identify the safe state. |
+| `google-uncertain-send` | Simulate an interrupted or ambiguous Gmail submission. | Staff do not resend blindly, reconcile using the stable message ID and retain the disposition. |
+| `restore-readiness` | Use the isolated restore procedure and inspect provider-facing records before enabling workers. | The restored system becomes ready, older approvals remain held and external effects are reconciled. |
+| `image-rollback` | Deploy the retained prior image IDs using the rollback procedure without replacing MongoDB or key volumes. | The recorded images run, readiness and read-only smoke checks pass and persistent state remains available. |
+| `controlled-recovery` | Recover the approved release after containment. | Monitoring is healthy, held operations are dispositioned and the exercise ends with writes disabled and FAQ mode off. |
+
+For Gate C, also exercise `pms-ambiguous-write` and `payment-ambiguous-create`. In both cases, lose or interrupt the synthetic provider response and prove the operation is reconciled read-only without creating a replacement request or replaying the approval.
+
+## Record and validation
+
+Copy `deploy/incident-exercise.example.json` into the restricted acceptance store. Bind it to the same full release commit and release-record SHA-256 used by the deployment and pilot decision. Replace the example timestamps, people, targets and evidence references. Change a scenario to `pass` only after its evidence has been independently reviewed. The example is intentionally invalid because its scenarios are `not-run`.
+
+For a Gate C exercise, set `targetGate` to `C` and add the PMS and payment scenario records. Validate the completed record with:
+
+```sh
+python3 deploy/incident_exercise.py /secure/acceptance/incident-exercise.json
+```
+
+The validator checks record completeness, release binding, scenario coverage, independent roles, timing targets and the safe final state. It does not inspect the referenced evidence, create an incident response capability or authorize a pilot. Retain its output and the record checksum, then reference them from `incident-support` in the pilot approval record.

docs/pilot-release.md

@@ -38,3 +38,13 @@ The go/no-go record must bind all evidence to the same release commit and releas
 - every pilot usability or security finding.
 
 Approval requires separate named decisions from the hotel pilot owner and technical release owner. Gate C additionally requires independently accepted PMS and payment-provider evidence. A conditional approval must identify the containment, owner, expiry and rollback trigger; an unresolved finding is not silently converted into acceptance. Retain the signed decision with the release rather than committing guest, credential or incident data to this repository.
+
+Copy `deploy/pilot-approval.example.json` into the restricted release store and complete it only after reviewing the referenced evidence. The example is intentionally invalid while its decision is `pending`. For a contained finding, record its named owner, future expiry and objective rollback trigger. Validate the completed record with:
+
+```sh
+python3 deploy/pilot_approval.py /secure/acceptance/pilot-approval.json
+```
+
+The validator requires the exact Gate B evidence set, or that set plus independently accepted PMS and payment-provider evidence for Gate C. It also verifies that observed concurrency meets the pre-agreed target and that p95 latency and error rate remain within their pre-agreed bounds. Structural validation does not inspect evidence or authorize rollout by itself.
+
+Complete the [incident and rollback exercise](incident-exercise.md) before marking `incident-support` as passed. Its record must use the same release identifiers and target gate as this decision. Reference the retained exercise record and validator output; do not substitute a local automated-test result for the supervised exercise.

tests/test_incident_exercise.py

@@ -0,0 +1,78 @@
+import importlib.util
+from pathlib import Path
+import unittest
+
+
+spec = importlib.util.spec_from_file_location("incident_exercise", Path(__file__).resolve().parents[1] / "deploy" / "incident_exercise.py")
+exercise = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(exercise)
+
+
+def valid_record(gate="B"):
+    required = exercise.GATE_C_SCENARIOS if gate == "C" else exercise.GATE_B_SCENARIOS
+    return {
+        "schemaVersion": 1,
+        "system": "guestops-incident-exercise",
+        "targetGate": gate,
+        "dataClassification": "synthetic-only",
+        "releaseCommit": "a" * 40,
+        "releaseRecordSha256": "b" * 64,
+        "environment": "https://sandbox-guestops.futuresens.co.uk",
+        "operator": "Exercise operator",
+        "incidentCommander": "Incident commander",
+        "reviewedBy": "Independent reviewer",
+        "startedAt": "2026-09-29T09:00:00Z",
+        "endedAt": "2026-09-29T10:00:00Z",
+        "reviewedAt": "2026-09-29T11:00:00Z",
+        "targetsMinutes": {"detection": 10, "containment": 20, "recovery": 60},
+        "observedMinutes": {"detection": 5, "containment": 15, "recovery": 45},
+        "scenarios": [
+            {"id": scenario, "status": "pass", "evidence": [f"restricted-ticket-{index}"]}
+            for index, scenario in enumerate(sorted(required), 1)
+        ],
+        "postExerciseState": {
+            "externalWrites": "disabled", "faqMode": "off", "unresolvedOperations": 0,
+        },
+    }
+
+
+class IncidentExerciseTests(unittest.TestCase):
+    def test_gate_b_and_c_complete_records_pass(self):
+        exercise.validate(valid_record("B"))
+        exercise.validate(valid_record("C"))
+
+    def test_gate_specific_scenarios_are_required(self):
+        record = valid_record("C")
+        record["scenarios"].pop()
+        with self.assertRaisesRegex(ValueError, "exact incident scenario"):
+            exercise.validate(record)
+
+    def test_timings_must_meet_preagreed_targets(self):
+        record = valid_record()
+        record["observedMinutes"]["containment"] = 21
+        with self.assertRaisesRegex(ValueError, "Observed containment"):
+            exercise.validate(record)
+        record = valid_record()
+        record["targetsMinutes"]["recovery"] = True
+        with self.assertRaisesRegex(ValueError, "recovery target"):
+            exercise.validate(record)
+
+    def test_independent_people_and_safe_evidence_are_required(self):
+        record = valid_record()
+        record["reviewedBy"] = record["operator"]
+        with self.assertRaisesRegex(ValueError, "different people"):
+            exercise.validate(record)
+        record = valid_record()
+        record["scenarios"][0]["evidence"] = ["guest@example.invalid"]
+        with self.assertRaisesRegex(ValueError, "safe opaque"):
+            exercise.validate(record)
+
+    def test_safe_post_exercise_state_is_required(self):
+        record = valid_record()
+        record["postExerciseState"]["externalWrites"] = "enabled"
+        with self.assertRaisesRegex(ValueError, "writes disabled"):
+            exercise.validate(record)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_pilot_approval.py

@@ -0,0 +1,70 @@
+import importlib.util
+from pathlib import Path
+import unittest
+
+
+spec = importlib.util.spec_from_file_location("pilot_approval", Path(__file__).resolve().parents[1] / "deploy" / "pilot_approval.py")
+approval = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(approval)
+
+
+def valid_record(gate="B"):
+    ids = approval.GATE_C if gate == "C" else approval.GATE_B
+    return {
+        "schemaVersion": 1, "targetGate": gate, "decision": "approved",
+        "releaseCommit": "a" * 40, "releaseRecordSha256": "b" * 64,
+        "decidedAt": "2026-09-29T12:00:00Z",
+        "approvals": {
+            "hotelOwner": {"name": "Hotel owner", "approvedAt": "2026-09-29T11:00:00Z"},
+            "technicalOwner": {"name": "Technical owner", "approvedAt": "2026-09-29T11:30:00Z"},
+        },
+        "capacity": {"reportSha256": "c" * 64, "targetConcurrency": 10, "observedConcurrency": 10, "targetP95Ms": 500, "observedP95Ms": 250, "targetErrorRate": 0.01, "observedErrorRate": 0},
+        "evidence": [{"id": item, "status": "pass", "references": ["restricted-ticket-" + item]} for item in sorted(ids)],
+    }
+
+
+class PilotApprovalTests(unittest.TestCase):
+    def test_gate_b_and_c_complete_records_pass(self):
+        approval.validate(valid_record("B"))
+        approval.validate(valid_record("C"))
+
+    def test_pending_or_missing_evidence_fails(self):
+        record = valid_record();record["decision"] = "pending"
+        with self.assertRaisesRegex(ValueError, "explicit approved"):
+            approval.validate(record)
+        record = valid_record();record["evidence"].pop()
+        with self.assertRaisesRegex(ValueError, "exact evidence"):
+            approval.validate(record)
+
+    def test_capacity_must_meet_preapproved_targets(self):
+        record = valid_record();record["capacity"]["observedP95Ms"] = 501
+        with self.assertRaisesRegex(ValueError, "observedP95Ms"):
+            approval.validate(record)
+        record = valid_record();record["capacity"]["observedConcurrency"] = 9
+        with self.assertRaisesRegex(ValueError, "Observed concurrency"):
+            approval.validate(record)
+        record = valid_record();record["capacity"]["targetConcurrency"] = True
+        with self.assertRaisesRegex(ValueError, "Observed concurrency"):
+            approval.validate(record)
+        record = valid_record();record["capacity"]["targetErrorRate"] = 2
+        with self.assertRaisesRegex(ValueError, "ratio"):
+            approval.validate(record)
+
+    def test_approvers_must_be_separate_people_without_email_addresses(self):
+        record = valid_record();record["approvals"]["technicalOwner"]["name"] = "Hotel owner"
+        with self.assertRaisesRegex(ValueError, "different people"):
+            approval.validate(record)
+        record = valid_record();record["approvals"]["hotelOwner"]["name"] = "owner@example.invalid"
+        with self.assertRaisesRegex(ValueError, "without an email"):
+            approval.validate(record)
+
+    def test_containment_requires_owner_future_expiry_and_trigger(self):
+        record = valid_record();item = record["evidence"][0];item["status"] = "contained"
+        with self.assertRaisesRegex(ValueError, "containment details"):
+            approval.validate(record)
+        item["containment"] = {"owner": "Release owner", "expiresAt": "2026-10-10T12:00:00Z", "rollbackTrigger": "Rollback if the contained condition occurs."}
+        approval.validate(record)
+
+
+if __name__ == "__main__":
+    unittest.main()