test(deploy): pin replay recovery predecessor

infra/deploy-runner/test_device_control_core_migration_replay_audit_artifact.py

@@ -94,7 +94,7 @@ class DeviceControlCoreMigrationReplayAuditArtifactTest(unittest.TestCase):
     def test_database_audit_returns_all_mismatches_without_mutation(self):
         database_result = mock.Mock(
             returncode=0,
-            stdout="2\t0\tfalse\tfalse\tfalse\n",
+            stdout="2\t0\tfalse\tfalse\tfalse\ttrue\n",
             stderr="",
         )
         with mock.patch.object(
@@ -117,7 +117,9 @@ class DeviceControlCoreMigrationReplayAuditArtifactTest(unittest.TestCase):
         self.assertFalse(evidence["constraintValidated"])
         self.assertFalse(evidence["constraintCoversFinalKinds"])
         self.assertFalse(evidence["hostTelemetryTableAbsent"])
+        self.assertTrue(evidence["constraintMatchesReplay011Kinds"])
         self.assertFalse(evidence["recovery044Ready"])
+        self.assertFalse(evidence["finalStateReady"])
         command = run.call_args.args[0]
         query = command[command.index("-c") + 1]
         self.assertTrue(query.lstrip().lower().startswith("with constraint_state"))
@@ -129,10 +131,10 @@ class DeviceControlCoreMigrationReplayAuditArtifactTest(unittest.TestCase):
             )
         )
 
-    def test_recovery_rejection_prints_five_bounded_values(self):
+    def test_recovery_rejection_prints_bounded_database_evidence(self):
         database_result = mock.Mock(
             returncode=0,
-            stdout="0\t0\ttrue\ttrue\ttrue\n",
+            stdout="0\t0\ttrue\ttrue\ttrue\tfalse\n",
             stderr="",
         )
         output = io.StringIO()
@@ -156,9 +158,129 @@ class DeviceControlCoreMigrationReplayAuditArtifactTest(unittest.TestCase):
             "constraint_validated=true",
             "constraint_covers_final_kinds=true",
             "host_telemetry_table_absent=true",
+            "constraint_matches_replay_011_kinds=false",
+            "recovery_final_state_ready=false",
         ):
             self.assertIn(marker, rendered)
 
+    def test_audit_preflight_resolves_recovery_from_canonical_inbox(self):
+        with tempfile.TemporaryDirectory(
+            prefix="nodedc-control-core-migration-replay-preflight-",
+        ) as directory:
+            root = Path(directory)
+            inbox = root / "inbox"
+            runner_tmp = root / "runner-tmp"
+            source = root / "device-plane"
+            inbox.mkdir()
+            runner_tmp.mkdir()
+            live_migration = (
+                source / RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_014_REL
+            )
+            live_migration.parent.mkdir(parents=True)
+            live_migration.write_bytes(b"migration-014-predecessor")
+            recovery_artifact = (
+                inbox
+                / RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_ARTIFACT
+            )
+            recovery_artifact.write_bytes(b"reviewed-recovery-044")
+            core = {
+                "containerId": "c" * 64,
+                "imageId": RUNNER.DEVICE_PLANE_CONTROL_CORE_V3_PREAPPLY_IMAGE_ID,
+                "status": "running",
+                "health": "unhealthy",
+                "restartCount": 3,
+            }
+            database = {
+                "invalidCommandKindCount": 0,
+                "triggeringReceiptCount": 0,
+                "constraintValidated": True,
+                "constraintCoversFinalKinds": True,
+                "hostTelemetryTableAbsent": True,
+                "constraintMatchesReplay011Kinds": True,
+                "recovery044Ready": False,
+                "finalStateReady": False,
+            }
+            failure_evidence = {
+                "firstBackup": root / "first-backup",
+                "secondBackup": root / "second-backup",
+            }
+            recovery_manifest = {
+                "id": RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_PATCH_ID,
+                "component": "device-plane",
+                "type": "app-overlay",
+            }
+
+            with (
+                mock.patch.object(RUNNER, "INBOX", inbox),
+                mock.patch.object(RUNNER, "TMP_DIR", runner_tmp),
+                mock.patch.object(
+                    RUNNER,
+                    "DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_ARTIFACT_SHA256",
+                    hashlib.sha256(recovery_artifact.read_bytes()).hexdigest(),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "DEVICE_PLANE_CONTROL_CORE_MIGRATION_014_PREDECESSOR_SHA256",
+                    hashlib.sha256(live_migration.read_bytes()).hexdigest(),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_migration_replay_audit_payload",
+                    return_value=(
+                        RUNNER.expected_device_plane_control_core_migration_replay_audit_descriptor()
+                    ),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_double_failure_evidence",
+                    return_value=failure_evidence,
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "load_artifact",
+                    return_value=(
+                        recovery_manifest,
+                        RUNNER.DEVICE_PLANE_CONTROL_CORE_MIGRATION_REPLAY_RECOVERY_ENTRIES,
+                        root / "recovery-payload",
+                    ),
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_migration_replay_recovery_payload",
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "load_state",
+                    return_value=[],
+                ) as load_state,
+                mock.patch.object(RUNNER, "component_root", return_value=source),
+                mock.patch.object(
+                    RUNNER,
+                    "validate_device_plane_control_core_migration_replay_preserved_runtime",
+                    return_value={"current": {"services": []}, "core": core},
+                ),
+                mock.patch.object(
+                    RUNNER,
+                    "collect_device_plane_control_core_migration_replay_database_evidence",
+                    return_value=database,
+                ) as collect_database,
+            ):
+                evidence = (
+                    RUNNER.validate_device_plane_control_core_migration_replay_audit_evidence(
+                        root / "audit-payload"
+                    )
+                )
+
+        self.assertEqual(evidence["recoveryArtifact"], recovery_artifact)
+        self.assertEqual(evidence["database"], database)
+        self.assertEqual(
+            [call.args for call in load_state.call_args_list],
+            [(RUNNER.STATE_FILE,), (RUNNER.FAILED_STATE_FILE,)],
+        )
+        collect_database.assert_called_once_with(
+            enforce_recovery_invariants=False,
+        )
+
     def test_plan_and_apply_paths_preserve_plan_only_boundary(self):
         plan_source = inspect.getsource(RUNNER.plan_artifact)
         apply_source = inspect.getsource(RUNNER.apply_artifact)

infra/deploy-runner/test_device_control_core_migration_replay_recovery_artifact.py

@@ -25,6 +25,10 @@ BUILDER = (
     SCRIPT_DIR
     / "build-device-control-core-migration-replay-recovery-artifact.mjs"
 )
+MIGRATION_011 = (
+    SCRIPT_DIR.parents[1]
+    / "services/device-control-core/migrations/011_device_control_resource_commands.sql"
+)
 
 
 def load_runner():
@@ -135,7 +139,7 @@ class DeviceControlCoreMigrationReplayRecoveryArtifactTest(unittest.TestCase):
     def test_database_preflight_is_read_only_and_covers_live_receipts(self):
         database_result = mock.Mock(
             returncode=0,
-            stdout="0\t3\ttrue\ttrue\ttrue\n",
+            stdout="0\t8\tfalse\tfalse\ttrue\ttrue\n",
             stderr="",
         )
         with mock.patch.object(
@@ -148,11 +152,18 @@ class DeviceControlCoreMigrationReplayRecoveryArtifactTest(unittest.TestCase):
             return_value=database_result,
         ) as run:
             evidence = (
-                RUNNER.collect_device_plane_control_core_migration_replay_database_evidence()
+                RUNNER.collect_device_plane_control_core_migration_replay_database_evidence(
+                    expected_state="replay-011-predecessor",
+                )
             )
 
         self.assertEqual(evidence["invalidCommandKindCount"], 0)
-        self.assertEqual(evidence["triggeringReceiptCount"], 3)
+        self.assertEqual(evidence["triggeringReceiptCount"], 8)
+        self.assertFalse(evidence["constraintValidated"])
+        self.assertFalse(evidence["constraintCoversFinalKinds"])
+        self.assertTrue(evidence["constraintMatchesReplay011Kinds"])
+        self.assertTrue(evidence["recovery044Ready"])
+        self.assertFalse(evidence["finalStateReady"])
         command = run.call_args.args[0]
         query = command[command.index("-c") + 1]
         self.assertTrue(query.lstrip().lower().startswith("with constraint_state"))
@@ -163,12 +174,76 @@ class DeviceControlCoreMigrationReplayRecoveryArtifactTest(unittest.TestCase):
                 query,
             )
         )
+        self.assertIn("regexp_matches", query)
+        for command_kind in RUNNER.DEVICE_PLANE_CONTROL_CORE_REPLAY_011_COMMAND_KINDS:
+            self.assertIn(f"'{command_kind}'", query)
+
+    def test_runner_predecessor_kind_set_matches_migration_011(self):
+        migration = MIGRATION_011.read_text(encoding="utf-8")
+        command_kinds = tuple(re.findall(r"'([^']+)'", migration))
+
+        self.assertEqual(
+            command_kinds,
+            RUNNER.DEVICE_PLANE_CONTROL_CORE_REPLAY_011_COMMAND_KINDS,
+        )
+
+    def test_database_acceptance_requires_final_validated_constraint(self):
+        database_result = mock.Mock(
+            returncode=0,
+            stdout="0\t8\ttrue\ttrue\ttrue\tfalse\n",
+            stderr="",
+        )
+        with mock.patch.object(
+            RUNNER,
+            "device_plane_service_container_ids",
+            return_value=("p" * 64,),
+        ), mock.patch.object(
+            RUNNER.subprocess,
+            "run",
+            return_value=database_result,
+        ):
+            evidence = (
+                RUNNER.collect_device_plane_control_core_migration_replay_database_evidence(
+                    expected_state="final",
+                )
+            )
+
+        self.assertTrue(evidence["constraintValidated"])
+        self.assertTrue(evidence["constraintCoversFinalKinds"])
+        self.assertFalse(evidence["constraintMatchesReplay011Kinds"])
+        self.assertFalse(evidence["recovery044Ready"])
+        self.assertTrue(evidence["finalStateReady"])
+
+    def test_database_preflight_rejects_ambiguous_constraint_shape(self):
+        database_result = mock.Mock(
+            returncode=0,
+            stdout="0\t8\tfalse\tfalse\ttrue\tfalse\n",
+            stderr="",
+        )
+        with mock.patch.object(
+            RUNNER,
+            "device_plane_service_container_ids",
+            return_value=("p" * 64,),
+        ), mock.patch.object(
+            RUNNER.subprocess,
+            "run",
+            return_value=database_result,
+        ), self.assertRaises(RUNNER.DeployError):
+            RUNNER.collect_device_plane_control_core_migration_replay_database_evidence(
+                expected_state="replay-011-predecessor",
+            )
 
     def test_rollback_accepts_exact_degraded_predecessor_boundary(self):
         rollback_source = inspect.getsource(RUNNER.rollback_device_plane_apply)
         acceptance_source = inspect.getsource(
             RUNNER.accept_device_plane_control_core_rollback_runtime
         )
+        preflight_source = inspect.getsource(
+            RUNNER.validate_device_plane_control_core_migration_replay_recovery_evidence
+        )
+        recovery_acceptance_source = inspect.getsource(
+            RUNNER.accept_device_plane_control_core_migration_replay_recovery
+        )
         self.assertIn(
             "is_device_plane_control_core_migration_replay_recovery_slice",
             rollback_source,
@@ -176,6 +251,14 @@ class DeviceControlCoreMigrationReplayRecoveryArtifactTest(unittest.TestCase):
         self.assertIn("retag_device_plane_control_core_image", rollback_source)
         self.assertIn("not predecessor_was_healthy", acceptance_source)
         self.assertIn("changed preserved service", acceptance_source)
+        self.assertIn(
+            'expected_state="replay-011-predecessor"',
+            preflight_source,
+        )
+        self.assertIn(
+            'expected_state="final"',
+            recovery_acceptance_source,
+        )
 
 
 if __name__ == "__main__":