fix(obs): keep bridge apply partial until verifier

scripts/reboot-recovery/signoz-188-otlp-grpc-bridge.sh

@@ -229,7 +229,7 @@ systemctl --user is-enabled --quiet '$REMOTE_SERVICE'"
     receipt execute pass "grpc_bridge_unit_enabled_active_allowlist_120_121"
     receipt verify pass "nodes_120_121_listener_and_upstream_4317_tcp_ready"
     receipt learning partial "application_trace_metric_10m_post_verifier_pending"
-    receipt terminal pass "grpc_transport_restored_post_verifier_pending"
+    receipt terminal partial "grpc_transport_restored_independent_600s_post_verifier_pending"
     ;;
   --status)
     remote "set -eu

scripts/reboot-recovery/signoz-canonical-route-preflight.py

@@ -136,6 +136,10 @@ def evaluate(root: Path, contract_path: Path) -> dict[str, Any]:
     grpc_verifier = _read(
         root, "scripts/reboot-recovery/verify-signoz-otel-grpc-runtime.sh"
     )
+    _, grpc_apply_marker, grpc_after_apply = grpc_bridge.partition("  --apply)")
+    grpc_apply_block, grpc_status_marker, _ = grpc_after_apply.partition(
+        "  --status)"
+    )
     checks["grpc_transport_recovery_source_ready"] = all(
         value in grpc_bridge
         for value in (
@@ -164,6 +168,45 @@ def evaluate(root: Path, contract_path: Path) -> dict[str, Any]:
             "restart_delta_0",
         )
     )
+    checks["grpc_apply_waits_for_independent_post_verifier"] = (
+        bool(grpc_apply_marker)
+        and bool(grpc_status_marker)
+        and (
+            'receipt learning partial '
+            '"application_trace_metric_10m_post_verifier_pending"'
+            in grpc_apply_block
+        )
+        and (
+            'receipt terminal partial '
+            '"grpc_transport_restored_independent_600s_post_verifier_pending"'
+            in grpc_apply_block
+        )
+        and "receipt terminal pass" not in grpc_apply_block
+        and grpc_apply_block.index("receipt learning partial")
+        < grpc_apply_block.index("receipt terminal partial")
+        and (
+            'receipt terminal pass '
+            '"real_application_otlp_grpc_trace_metric_freshness_closed"'
+            in grpc_verifier
+        )
+    )
+
+    incumbent = contract.get("routes", {}).get("incumbent", {})
+    grpc_runtime = contract.get("runtime_observations", {}).get(
+        "application_otlp_grpc", {}
+    )
+    grpc_source_fix = grpc_runtime.get("source_fix", {})
+    checks["grpc_contract_runtime_state_consistent"] = (
+        incumbent.get("grpc_bridge_status")
+        == "user_systemd_enabled_active_runtime_verified"
+        and contract.get("runtime_observations", {}).get("grpc_bridge_supervisor")
+        == "user_systemd_enabled_active"
+        and grpc_runtime.get("runtime_status") == "restored_verified"
+        and grpc_source_fix.get("runtime_apply_status")
+        == "applied_post_verifier_pass"
+        and grpc_source_fix.get("runtime_verifier_terminal") == "pass"
+        and int(grpc_source_fix.get("runtime_verifier_window_seconds") or 0) >= 600
+    )
     static_recovery = _read(
         root, "scripts/reboot-recovery/signoz-110-static-ingest.sh"
     )

scripts/reboot-recovery/tests/test_signoz_canonical_route_migration.py

@@ -67,6 +67,8 @@ def test_preflight_is_no_write_and_reports_explicit_partial_runtime() -> None:
     assert payload["runtime_status"] == "partial_degraded"
     assert payload["active_route"] == "incumbent"
     assert payload["checks"] and all(payload["checks"].values())
+    assert payload["checks"]["grpc_apply_waits_for_independent_post_verifier"] is True
+    assert payload["checks"]["grpc_contract_runtime_state_consistent"] is True
     assert "organization_not_initialized" in payload["blockers"]
     assert "signoz_control_plane_health_route_not_deployed" in payload["blockers"]
     assert "wave_b_not_authorized" in payload["blockers"]
@@ -223,6 +225,47 @@ def test_grpc_bridge_is_independently_supervised_allowlisted_and_reversible() ->
     assert "User=" not in unit
 
 
+def test_grpc_apply_terminal_stays_partial_until_independent_runtime_verifier() -> None:
+    bridge = GRPC_BRIDGE.read_text(encoding="utf-8")
+    verifier = GRPC_VERIFIER.read_text(encoding="utf-8")
+    apply_block = bridge.split("  --apply)", 1)[1].split("  --status)", 1)[0]
+
+    learning = (
+        'receipt learning partial '
+        '"application_trace_metric_10m_post_verifier_pending"'
+    )
+    terminal = (
+        'receipt terminal partial '
+        '"grpc_transport_restored_independent_600s_post_verifier_pending"'
+    )
+    assert learning in apply_block
+    assert terminal in apply_block
+    assert apply_block.index(learning) < apply_block.index(terminal)
+    assert "receipt terminal pass" not in apply_block
+    assert (
+        'receipt terminal pass '
+        '"real_application_otlp_grpc_trace_metric_freshness_closed"'
+        in verifier
+    )
+
+
+def test_contract_grpc_route_state_matches_verified_runtime_evidence() -> None:
+    contract = yaml.safe_load(CONTRACT.read_text(encoding="utf-8"))
+    incumbent = contract["routes"]["incumbent"]
+    observations = contract["runtime_observations"]
+    grpc = observations["application_otlp_grpc"]
+
+    assert (
+        incumbent["grpc_bridge_status"]
+        == "user_systemd_enabled_active_runtime_verified"
+    )
+    assert observations["grpc_bridge_supervisor"] == "user_systemd_enabled_active"
+    assert grpc["runtime_status"] == "restored_verified"
+    assert grpc["source_fix"]["runtime_apply_status"] == "applied_post_verifier_pass"
+    assert grpc["source_fix"]["runtime_verifier_terminal"] == "pass"
+    assert grpc["source_fix"]["runtime_verifier_window_seconds"] >= 600
+
+
 def test_grpc_runtime_verifier_uses_real_service_aggregates_and_ten_minute_gate() -> None:
     text = GRPC_VERIFIER.read_text(encoding="utf-8")
     for expected in (