Implement offline recovery handoff within existing scheduling scope

Apply the user-directed Emergent Security guidance to close the redundant role gate. Add closed request/receipt validation bound to run, cluster, implementation, mode and authority reference; preserve platform enforcement and operation admission gates.

Decision-Class: task-scoped-code-and-tests

Delegation: explicit-user-task; Emergent-Security-A6

Evidence: 653-tests-passed; 1-skipped; focused-43-passed; ruff-clean

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a100de-2e1c-7e23-9ae9-2da0e0499411

tests/test_recovery_contract.py

@@ -0,0 +1,129 @@
+"""Offline owner-handoff contract checks; no platform calls or credentials."""
+from datetime import UTC, datetime, timedelta
+from uuid import uuid4
+
+import pytest
+from activity_core.recovery_contract import parse_recovery_request, validate_recovery_receipt
+
+NOW = datetime(2026, 10, 3, 10, tzinfo=UTC)
+
+
+def request(mode="apply"):
+    return parse_recovery_request({
+        "run_id": str(uuid4()), "mode": mode,
+        "cluster_uid": str(uuid4()), "implementation_sha256": "b" * 64,
+        "authority_ref": str(uuid4()) if mode == "apply" else None,
+    })
+
+
+def receipt(req):
+    return {
+        "schema_id": "activity.platform-recovery-receipt.v1",
+        "run_id": str(req.run_id), "operation": req.operation, "target": req.target,
+        "cluster_uid": str(req.cluster_uid),
+        "mode": req.mode, "authority_ref": str(req.authority_ref) if req.authority_ref else None,
+        "outcome": "succeeded" if req.mode == "apply" else "planned",
+        "started_at": (NOW - timedelta(minutes=5)).isoformat(),
+        "finished_at": (NOW - timedelta(minutes=1)).isoformat(),
+        "validation_succeeded": req.mode == "apply",
+        "cleanup": "completed" if req.mode == "apply" else "not_created",
+        "evidence_sha256": "a" * 64, "implementation_sha256": "b" * 64,
+    }
+
+
+def check(req, value):
+    return validate_recovery_receipt(req, value, observed_at=NOW, max_age=timedelta(hours=1))
+
+
+@pytest.mark.parametrize("mode", ["apply", "dry_run"])
+def test_exact_owner_receipt_preserves_outcome(mode):
+    req = request(mode)
+    result = check(req, receipt(req))
+    assert result.outcome == ("succeeded" if mode == "apply" else "planned")
+
+
+@pytest.mark.parametrize("change", [
+    {"operation": "platform_root"}, {"target": "forgejo-db"},
+    {"command": "arbitrary"}, {"credential": "SECRET_CANARY"},
+    {"authority_ref": None}, {"mode": "unknown"},
+])
+def test_request_rejects_widening_and_missing_apply_reference_without_echo(change):
+    value = request().model_dump(mode="json")
+    value.update(change)
+    with pytest.raises(ValueError) as exc:
+        parse_recovery_request(value)
+    assert str(exc.value) == "invalid recovery request envelope"
+    assert exc.value.__suppress_context__
+
+
+@pytest.mark.parametrize("change", [
+    {"run_id": str(uuid4())}, {"authority_ref": str(uuid4())},
+    {"cluster_uid": str(uuid4())}, {"implementation_sha256": "c" * 64},
+    {"target": "production"}, {"mode": "dry_run"},
+    {"validation_succeeded": False}, {"validation_succeeded": "true"},
+    {"cleanup": "failed"}, {"cleanup": "unknown"},
+    {"finished_at": (NOW + timedelta(seconds=1)).isoformat()},
+    {"started_at": NOW.isoformat()},
+    {"finished_at": "2026-10-03T09:59:00"},
+    {"evidence_sha256": "invalid"},
+    {"stderr": "SECRET_CANARY"}, {"evidence_sha256": "SECRET_CANARY"},
+])
+def test_receipt_rejects_mismatch_invalid_proof_and_sensitive_fields(change):
+    req = request()
+    value = receipt(req)
+    value.update(change)
+    with pytest.raises(ValueError) as exc:
+        check(req, value)
+    assert "SECRET_CANARY" not in str(exc.value)
+
+
+def test_stale_receipt_is_not_freshened_by_scheduler():
+    req = request()
+    value = receipt(req)
+    value.update(started_at=(NOW - timedelta(hours=3)).isoformat(),
+                 finished_at=(NOW - timedelta(hours=2)).isoformat())
+    with pytest.raises(ValueError, match="stale"):
+        check(req, value)
+
+
+@pytest.mark.parametrize("outcome", ["failed", "rejected", "unknown"])
+def test_failure_and_ambiguity_remain_visible(outcome):
+    req = request()
+    value = receipt(req)
+    value.update(outcome=outcome, validation_succeeded=False, cleanup="unknown")
+    assert check(req, value).outcome == outcome
+
+
+def test_dry_run_cannot_claim_a_restore():
+    req = request("dry_run")
+    value = receipt(req)
+    value.update(outcome="succeeded", validation_succeeded=True, cleanup="completed")
+    with pytest.raises(ValueError):
+        check(req, value)
+
+
+@pytest.mark.parametrize("observed_at,max_age", [
+    (NOW.replace(tzinfo=None), timedelta(hours=1)),
+    (NOW, timedelta(0)), (NOW, timedelta(seconds=-1)),
+])
+def test_freshness_policy_must_be_explicit_and_valid(observed_at, max_age):
+    req = request()
+    with pytest.raises(ValueError):
+        validate_recovery_receipt(req, receipt(req), observed_at=observed_at, max_age=max_age)
+
+
+def test_receipt_requires_an_explicit_supported_schema():
+    req = request()
+    for schema in (None, "activity.platform-recovery-receipt.v2"):
+        value = receipt(req)
+        if schema is None:
+            del value["schema_id"]
+        else:
+            value["schema_id"] = schema
+        with pytest.raises(ValueError, match="invalid recovery receipt envelope"):
+            check(req, value)
+
+
+def test_offline_contract_does_not_register_an_operation():
+    from activity_core.bounded_operations import BOUNDED_OPERATION_REGISTRY
+    assert "apps_pg_restore_evidence" not in BOUNDED_OPERATION_REGISTRY