fix(ops): enforce Prometheus runtime identity

services/metrics_edge_controlled_apply_service.py

@@ -244,6 +244,8 @@ def _prometheus_identity() -> dict[str, str]:
         "{{.Id}}|{{.Image}}|{{range .Mounts}}{{if eq .Destination \"/prometheus\"}}{{.Name}}{{end}}{{end}}",
     ])
     container_id, image_id, data_volume = (completed.stdout.strip().split("|", 2) + ["", ""])[:3]
+    if not container_id or not image_id or not data_volume:
+        raise RuntimeError("Prometheus container identity is incomplete")
     return {
         "container_id": container_id,
         "image_id": image_id,
@@ -251,6 +253,16 @@ def _prometheus_identity() -> dict[str, str]:
     }
 
 
+def _prometheus_identity_preserved(
+    before: dict[str, str],
+    after: dict[str, str],
+) -> bool:
+    return bool(
+        before.get("image_id") == after.get("image_id")
+        and before.get("data_volume") == after.get("data_volume")
+    )
+
+
 def _prometheus_config_sha(path: Path) -> tuple[str, str]:
     host_sha = hashlib.sha256(path.read_bytes()).hexdigest()
     completed = _run([
@@ -369,6 +381,10 @@ def apply_metrics_edge_plan(
         if host_sha != container_sha:
             raise RuntimeError("Prometheus config SHA mismatch after controlled rebind")
         prometheus_after = _prometheus_identity()
+        identity_preserved = _prometheus_identity_preserved(
+            prometheus_before,
+            prometheus_after,
+        )
         receipt["prometheus_runtime"] = {
             "before": prometheus_before,
             "after": prometheus_after,
@@ -378,6 +394,7 @@ def apply_metrics_edge_plan(
                 prometheus_before["data_volume"] == prometheus_after["data_volume"]
             ),
             "image_unchanged": prometheus_before["image_id"] == prometheus_after["image_id"],
+            "identity_preserved": identity_preserved,
         }
         canary_command = ["curl", "--noproxy", "*", "-sS", "--max-time", "20"]
         if metrics_canary_resolve:
@@ -395,10 +412,13 @@ def apply_metrics_edge_plan(
             "metrics_internal_canary": internal_code == "200",
             "metrics_target_health": target_health,
             "prometheus_config_inode_reconciled": True,
+            "prometheus_identity_preserved": identity_preserved,
         }
-        if internal_code != "200" or target_health != "up":
+        if internal_code != "200" or target_health != "up" or not identity_preserved:
             raise RuntimeError(
-                f"post-verifier failed: internal={internal_code}, target={target_health}"
+                "post-verifier failed: "
+                f"internal={internal_code}, target={target_health}, "
+                f"identity_preserved={identity_preserved}"
             )
         receipt["status"] = "pass"
         receipt["completed_at"] = datetime.now(timezone.utc).isoformat()

tests/test_metrics_edge_controlled_apply_service.py

@@ -144,6 +144,24 @@ def test_prometheus_config_sha_detects_stale_container_mount(tmp_path: Path, mon
     assert container_sha == stale_sha
 
 
+def test_prometheus_identity_verifier_requires_same_image_and_data_volume():
+    from services.metrics_edge_controlled_apply_service import (
+        _prometheus_identity_preserved,
+    )
+
+    before = {"image_id": "image-a", "data_volume": "volume-a"}
+
+    assert _prometheus_identity_preserved(before, dict(before)) is True
+    assert _prometheus_identity_preserved(
+        before,
+        {"image_id": "image-b", "data_volume": "volume-a"},
+    ) is False
+    assert _prometheus_identity_preserved(
+        before,
+        {"image_id": "image-a", "data_volume": "volume-b"},
+    ) is False
+
+
 def test_render_rejects_incomplete_templates():
     from services.metrics_edge_controlled_apply_service import (
         render_nginx_config,