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,6 +158,8 @@ 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)
 
@@ -192,7 +196,9 @@ class DeviceControlCoreMigrationReplayAuditArtifactTest(unittest.TestCase):
                 "constraintValidated": True,
                 "constraintCoversFinalKinds": True,
                 "hostTelemetryTableAbsent": True,
+                "constraintMatchesReplay011Kinds": True,
                 "recovery044Ready": False,
+                "finalStateReady": False,
             }
             failure_evidence = {
                 "firstBackup": root / "first-backup",

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__":