Merge deployment recovery hardening (#110)

Author: alexeygrigorev

_docs/compatibility/development-terminology-allowlist.json

@@ -180,7 +180,7 @@
     },
     {
       "path": "core/tests/test_deployment_release.py",
-      "sha256": "c905b5bd64f5defdf7ca86be5548fbbdddbad47723e715140d7efee5cfbef975",
+      "sha256": "9d8881f2fbb73182bdafd5dcb06629e93ad08dada39a0c86e90626d8038f062a",
       "class": "legacy_contract_test",
       "reason": "Release tests exercise exact physical task, secret, repository, and tag values.",
       "follow_up": "#94",
@@ -196,7 +196,7 @@
     },
     {
       "path": "core/tests/test_deployment_workflow.py",
-      "sha256": "b5dca94b59c246569909ea1d5bc6ce0da5d9aa5d86e729f7b0e97dc98f284cf8",
+      "sha256": "e98b201c57383085dabcaf8a38cc36d167cab0b62016e2d4a1ca3c1f3354f09f",
       "class": "legacy_contract_test",
       "reason": "Workflow tests exercise removed inputs, frozen Gate-B records, and exact physical values.",
       "follow_up": "#94",

_docs/runbooks/development-release.md

@@ -46,7 +46,15 @@ Local runs use `local-development-build-version-not-configured` with null source
 deployed smoke reject it. A strict schema-1 reader exists only for an already-active or recorded
 prior/rollback target and represents its VERSION as its full source SHA; it may not invent a
 timestamp, publish a new schema-1 image, register a schema-1 task, or write a schema-1 success
-record.
+record. Recovery of that prior target first proves the exact receipt-bound ECS task-definition
+pair, task identity, image digest, source SHA, terminal counts, and singleton worker. Its final
+public proof then requires the exact schema-1 health contract: liveness contains only
+`status=ok` plus `version=<full source SHA>`, and readiness contains only `status=ready` plus the
+successful configuration, database, and migrations checks. Schema-1 verification never accepts a
+schema-2 or mixed health shape. Schema-2 recovery continues to require the exact recorded
+VERSION/source/digest triplet on both health endpoints. A retained receipt/observation error or a
+failed terminal-pair proof prevents either schema's public-health request; evidence records
+`not_attempted` and never claims exact prior-SHA readiness in that state.
 
 ## One-time bootstrap
 
@@ -812,8 +820,11 @@ failure after web mutation restores the prior exact web task definition, count,
 deployment ID. If worker `UpdateService` was actually invoked, its restoration is receipt-bound as
 well. If worker was untouched, compensation issues no worker mutation and instead read-only proves
 its captured task definition, count, PRIMARY deployment ID, terminal state, and singleton bound as
-part of the exact pair. It then validates the prior digest/SHA and public health. The database
-remains migrated forward.
+part of the exact pair. It then validates the prior digest/SHA and uses the identity schema from
+that proved pair to select the final public-health contract: exact legacy full-SHA health for a
+schema-1 prior, or the exact VERSION/source/digest triplet for schema 2. A receipt, observation, or
+terminal-pair failure blocks that request and records public health as not attempted and false.
+The database remains migrated forward.
 
 The workflow concurrency group is `website-development-release` with cancellation disabled. Never
 cancel an in-progress release to start another one.

core/tests/test_deployment_release.py

@@ -677,6 +677,140 @@ def test_independent_inclusive_240_420_deadlines_have_one_final_read(self) -> No
         self.assertFalse(any(value > 420 for value in gateway.observation_times["worker"]))
         self.assertEqual(gateway.current, 420)
 
+    def test_schema1_recovery_proves_terminal_receipts_before_legacy_public_health(self) -> None:
+        gateway = CooperativeRecoveryGateway(complete_at={"web": 0, "worker": 0})
+        legacy_identity = ReleaseIdentity.legacy(SHA_A, DIGEST_A, REPOSITORY)
+        original_update = gateway.update_service
+
+        def update_legacy(*args, **kwargs):  # type: ignore[no-untyped-def]
+            receipt = original_update(*args, **kwargs)
+            workload = receipt.workload
+            snapshot = gateway.snapshots[workload]
+            gateway.snapshots[workload] = ServiceSnapshot(
+                service_name=snapshot.service_name,
+                task_definition_arn=snapshot.task_definition_arn,
+                desired_count=snapshot.desired_count,
+                running_count=snapshot.running_count,
+                pending_count=snapshot.pending_count,
+                source_sha=legacy_identity.source_sha,
+                image_digest=legacy_identity.image_digest,
+                primary_deployment_id=snapshot.primary_deployment_id,
+                version=legacy_identity.version,
+                identity_schema=legacy_identity.identity_schema,
+            )
+            return receipt
+
+        gateway.update_service = update_legacy  # type: ignore[method-assign]
+        targets, terminal, attempted = self.restore_phase()
+        attempted_states: dict[str, ServiceTarget | ServicePredecessor] = dict(attempted)
+
+        _compensate(
+            gateway,
+            targets,
+            terminal,
+            legacy_identity,
+            attempted_states,
+        )
+
+        terminal_index = gateway.operations.index("terminal-at:0")
+        health_index = gateway.operations.index("health-at:0")
+        self.assertLess(terminal_index, health_index)
+        self.assertIn(f"health:{SHA_A}:{SHA_A}", gateway.operations)
+
+    def test_schema1_mixed_worker_never_attempts_or_passes_public_health(self) -> None:
+        gateway = CooperativeRecoveryGateway(complete_at={"web": 0, "worker": 0})
+        legacy_identity = ReleaseIdentity.legacy(SHA_A, DIGEST_A, REPOSITORY)
+        original_update = gateway.update_service
+
+        def update_mixed_worker(*args, **kwargs):  # type: ignore[no-untyped-def]
+            receipt = original_update(*args, **kwargs)
+            workload = receipt.workload
+            snapshot = gateway.snapshots[workload]
+            source_sha = "f" * 40 if workload == "worker" else legacy_identity.source_sha
+            gateway.snapshots[workload] = ServiceSnapshot(
+                service_name=snapshot.service_name,
+                task_definition_arn=snapshot.task_definition_arn,
+                desired_count=snapshot.desired_count,
+                running_count=snapshot.running_count,
+                pending_count=snapshot.pending_count,
+                source_sha=source_sha,
+                image_digest=legacy_identity.image_digest,
+                primary_deployment_id=snapshot.primary_deployment_id,
+                version=source_sha,
+                identity_schema=legacy_identity.identity_schema,
+            )
+            return receipt
+
+        gateway.update_service = update_mixed_worker  # type: ignore[method-assign]
+        targets, terminal, attempted = self.restore_phase()
+        attempted_states: dict[str, ServiceTarget | ServicePredecessor] = dict(attempted)
+        Path(".tmp").mkdir(exist_ok=True)
+        with tempfile.TemporaryDirectory(dir=".tmp") as directory:
+            evidence_path = Path(directory) / "recovery-evidence.json"
+            with self.assertRaises(CompensationError):
+                _compensate(
+                    gateway,
+                    targets,
+                    terminal,
+                    legacy_identity,
+                    attempted_states,
+                    evidence_path=evidence_path,
+                )
+            stages = json.loads(evidence_path.read_text())["stages"]
+
+        self.assertFalse(any(operation.startswith("health") for operation in gateway.operations))
+        terminal_evidence = next(
+            item for item in stages if item["stage"] == "recovery_terminal_pair"
+        )
+        public_evidence = next(item for item in stages if item["stage"] == "recovery_public_health")
+        total_evidence = next(item for item in stages if item["stage"] == "recovery_total")
+        self.assertEqual(terminal_evidence["result"], "contract_contradiction")
+        self.assertEqual(public_evidence["result"], "not_attempted")
+        self.assertEqual(public_evidence["proof"]["attempted"], False)
+        self.assertEqual(public_evidence["proof"]["exact_prior_sha_ready"], False)
+        self.assertEqual(total_evidence["result"], "contract_contradiction")
+        self.assertEqual(total_evidence["proof"]["terminal_pair"], False)
+        self.assertEqual(total_evidence["proof"]["public_health"], False)
+
+    def test_retained_receipt_error_blocks_schema2_public_health_after_terminal_read(self) -> None:
+        gateway = CooperativeRecoveryGateway(complete_at={"web": 0, "worker": 0})
+        original_observe = gateway.observe_recovery_receipt
+
+        def observe_with_worker_error(
+            receipt: ServiceUpdateReceipt,
+            *,
+            workload_deadline: float,
+            phase_deadline: float,
+        ) -> bool:
+            if receipt.workload == "worker":
+                raise ReleaseContractError("injected worker receipt contradiction")
+            return original_observe(
+                receipt,
+                workload_deadline=workload_deadline,
+                phase_deadline=phase_deadline,
+            )
+
+        gateway.observe_recovery_receipt = observe_with_worker_error  # type: ignore[method-assign]
+        Path(".tmp").mkdir(exist_ok=True)
+        with tempfile.TemporaryDirectory(dir=".tmp") as directory:
+            evidence_path = Path(directory) / "recovery-evidence.json"
+            with self.assertRaises(CompensationError):
+                self.compensate(gateway, evidence_path=evidence_path)
+            stages = json.loads(evidence_path.read_text())["stages"]
+
+        self.assertFalse(any(operation.startswith("health") for operation in gateway.operations))
+        terminal_evidence = next(
+            item for item in stages if item["stage"] == "recovery_terminal_pair"
+        )
+        public_evidence = next(item for item in stages if item["stage"] == "recovery_public_health")
+        total_evidence = next(item for item in stages if item["stage"] == "recovery_total")
+        self.assertEqual(terminal_evidence["result"], "passed")
+        self.assertEqual(public_evidence["result"], "not_attempted")
+        self.assertEqual(public_evidence["proof"]["attempted"], False)
+        self.assertEqual(public_evidence["proof"]["exact_prior_sha_ready"], False)
+        self.assertEqual(total_evidence["result"], "contract_contradiction")
+        self.assertEqual(total_evidence["proof"]["public_health"], False)
+
     def test_worker_deadline_cannot_be_rescued_by_a_later_terminal_fixture(self) -> None:
         gateway = CooperativeRecoveryGateway(complete_at={"web": 10, "worker": 430})
 

core/tests/test_deployment_workflow.py

@@ -39,6 +39,7 @@
     ServiceTarget,
     ServiceUpdateReceipt,
 )
+from deploy.legacy_development_compatibility import ECR_REPOSITORY_URI
 from deploy.task_definitions import (
     FIXED_NONSECRET_ENVIRONMENT,
     TaskDefinitionConfig,
@@ -5808,3 +5809,27 @@ def test_public_health_polls_until_exact_readiness_or_timeout(self) -> None:
             self.assertRaisesMessage(ReleaseContractError, "ALB target readiness"),
         ):
             gateway.verify_public_web(identity)
+
+    def test_schema1_public_recovery_uses_only_the_legacy_health_contract(self) -> None:
+        gateway = self.gateway(FakeMigrationEcs({}, {}))
+        source_sha = "a" * 40
+        identity = ReleaseIdentity.legacy(
+            source_sha,
+            f"sha256:{'a' * 64}",
+            ECR_REPOSITORY_URI,
+        )
+        gateway._service = Mock(return_value={"desiredCount": 1})  # type: ignore[method-assign]
+        gateway.elbv2 = Mock()
+        gateway.elbv2.describe_target_health.return_value = {
+            "TargetHealthDescriptions": [{"TargetHealth": {"State": "healthy"}}]
+        }
+
+        with (
+            patch("deploy.aws_gateway.time.monotonic", side_effect=[0] * 8),
+            patch("deploy.aws_gateway.verify_legacy_health") as legacy_health,
+            patch("deploy.aws_gateway.verify_health") as schema2_health,
+        ):
+            gateway.verify_public_web(identity)
+
+        legacy_health.assert_called_once_with(gateway.config.base_url, source_sha)
+        schema2_health.assert_not_called()

core/tests/test_release_identity.py

@@ -24,7 +24,7 @@
     SourceIdentityConstructor,
     validate_schema2_version,
 )
-from deploy.smoke import verify_health
+from deploy.smoke import Response, verify_health, verify_legacy_health
 
 SHA = "3f6c227" + "a" * 33
 DIGEST = f"sha256:{'b' * 64}"
@@ -331,6 +331,127 @@ def test_task_capture_and_smoke_reject_calendar_invalid_version_before_use() ->
     request.assert_not_called()
 
 
+def test_schema1_health_accepts_the_exact_legacy_contract() -> None:
+    responses = [
+        Response(
+            status=200,
+            headers={"x-robots-tag": "noindex, nofollow"},
+            body=json.dumps({"status": "ok", "version": SHA}).encode(),
+        ),
+        Response(
+            status=200,
+            headers={"x-robots-tag": "noindex, nofollow"},
+            body=json.dumps(
+                {
+                    "status": "ready",
+                    "checks": {
+                        "configuration": {"status": "ok"},
+                        "database": {"status": "ok"},
+                        "migrations": {"status": "ok"},
+                    },
+                }
+            ).encode(),
+        ),
+    ]
+
+    with patch("deploy.smoke._request", side_effect=responses) as request:
+        verify_legacy_health("https://web.dtcdev.click", SHA)
+
+    assert [call.args[1] for call in request.call_args_list] == [
+        "/health/live",
+        "/health/ready",
+    ]
+
+
+@pytest.mark.parametrize(
+    ("live_payload", "ready_payload", "message"),
+    [
+        (
+            {"status": "ok", "version": "4" * 40},
+            {
+                "status": "ready",
+                "checks": {
+                    "configuration": {"status": "ok"},
+                    "database": {"status": "ok"},
+                    "migrations": {"status": "ok"},
+                },
+            },
+            "exact legacy release identity",
+        ),
+        (
+            {"status": "ok", "version": SHA, "source_sha": SHA},
+            {
+                "status": "ready",
+                "checks": {
+                    "configuration": {"status": "ok"},
+                    "database": {"status": "ok"},
+                    "migrations": {"status": "ok"},
+                },
+            },
+            "exact legacy release identity",
+        ),
+        (
+            {"status": "ok", "version": SHA},
+            {
+                "status": "ready",
+                "source_sha": "4" * 40,
+                "checks": {
+                    "configuration": {"status": "ok"},
+                    "database": {"status": "ok"},
+                    "migrations": {"status": "ok"},
+                },
+            },
+            "exact legacy readiness contract",
+        ),
+    ],
+)
+def test_schema1_health_rejects_wrong_or_mixed_identity_contracts(
+    live_payload: dict[str, object],
+    ready_payload: dict[str, object],
+    message: str,
+) -> None:
+    responses = [
+        Response(
+            status=200,
+            headers={"x-robots-tag": "noindex, nofollow"},
+            body=json.dumps(live_payload).encode(),
+        ),
+        Response(
+            status=200,
+            headers={"x-robots-tag": "noindex, nofollow"},
+            body=json.dumps(ready_payload).encode(),
+        ),
+    ]
+
+    with (
+        patch("deploy.smoke._request", side_effect=responses),
+        pytest.raises(ReleaseContractError, match=message),
+    ):
+        verify_legacy_health("https://web.dtcdev.click", SHA)
+
+
+def test_schema2_health_still_rejects_a_mismatched_exact_triplet() -> None:
+    version = f"20260809-143205-{SHA[:7]}"
+    live = Response(
+        status=200,
+        headers={"x-robots-tag": "noindex, nofollow"},
+        body=json.dumps(
+            {
+                "status": "ok",
+                "version": version,
+                "source_sha": "4" * 40,
+                "image_digest": DIGEST,
+            }
+        ).encode(),
+    )
+
+    with (
+        patch("deploy.smoke._request", return_value=live),
+        pytest.raises(ReleaseContractError, match="exact release identity"),
+    ):
+        verify_health("https://web.dtcdev.click", version, SHA, DIGEST)
+
+
 def test_schema2_version_format_has_one_parser_implementation() -> None:
     definitions = []
     for source_root in (ROOT / "core", ROOT / "deploy"):

deploy/aws_gateway.py

@@ -39,7 +39,7 @@
     ECR_REPOSITORY_URI,
     RESOURCE_ENVIRONMENT_TAG,
 )
-from deploy.smoke import run_http_smoke, verify_health
+from deploy.smoke import run_http_smoke, verify_health, verify_legacy_health
 from deploy.task_definitions import (
     TaskDefinitionConfig,
     assert_normalized_service_pair,
@@ -2495,12 +2495,15 @@ def verify_public_web(
 
         while time.monotonic() < deadline:
             try:
-                verify_health(
-                    self.config.base_url,
-                    identity.version,
-                    identity.source_sha,
-                    identity.image_digest,
-                )
+                if identity.identity_schema == 1:
+                    verify_legacy_health(self.config.base_url, identity.source_sha)
+                else:
+                    verify_health(
+                        self.config.base_url,
+                        identity.version,
+                        identity.source_sha,
+                        identity.image_digest,
+                    )
                 self._require_not_after_deadline(deadline, context="public web health")
                 return
             except Exception as error: