feat(observability): verify SigNoz three-signal route parity

scripts/reboot-recovery/tests/test_signoz_canonical_route_migration.py

@@ -15,6 +15,10 @@ PREFLIGHT = ROOT / "scripts/reboot-recovery/signoz-canonical-route-preflight.py"
 BRIDGE = ROOT / "scripts/reboot-recovery/signoz-188-otlp-bridge.sh"
 STATIC = ROOT / "scripts/reboot-recovery/signoz-110-static-ingest.sh"
 UNIT = ROOT / "ops/systemd/user/awoooi-signoz-otlp-bridge.service"
+SIGNAL_VERIFIER = ROOT / "scripts/reboot-recovery/verify-signoz-otel-signals.sh"
+PARITY_SNAPSHOT = ROOT / "ops/signoz/p0-obs-002-three-signal-route-parity.yaml"
+OTEL_DAEMONSET = ROOT / "k8s/observability/otel-collector-daemonset.yaml"
+FILELOG_CANARY = ROOT / "ops/signoz/filelog-normalization-canary.yaml"
 
 
 def test_wave_a_contract_keeps_active_route_and_all_producers_on_incumbent() -> None:
@@ -82,6 +86,95 @@ def test_bridge_unit_is_user_scoped_restarting_and_hardened() -> None:
     assert "sudo" not in text
 
 
+def test_three_signal_verifier_is_route_scoped_and_exact() -> None:
+    text = SIGNAL_VERIFIER.read_text(encoding="utf-8")
+    for expected in (
+        "OTLP_BASE_ENDPOINT",
+        "ROUTE_ID",
+        "/v1/$signal",
+        "resourceLogs",
+        "resourceMetrics",
+        "resourceSpans",
+        "signoz_logs.logs_v2",
+        "signoz_metrics.time_series_v4",
+        "signoz_metrics.samples_v4",
+        "signoz_traces.signoz_index_v3",
+        "exact_log_1_metric_series_1_metric_sample_1_trace_1",
+    ):
+        assert expected in text
+    assert "github.com" not in text
+    assert "SIGNOZ-API-KEY" not in text
+    assert "Authorization:" not in text
+
+
+def test_three_signal_snapshot_keeps_synthetic_parity_separate_from_promotion() -> None:
+    contract = yaml.safe_load(CONTRACT.read_text(encoding="utf-8"))
+    parity = contract["runtime_observations"]["synthetic_three_signal_route_parity"]
+    assert parity["incumbent_bridge"] == "pass"
+    assert parity["direct_110_from_120"] == "pass"
+    assert parity["direct_110_from_121"] == "pass"
+    assert parity["exact_count_per_signal_and_route"] == 1
+    assert parity["real_application_trace_freshness_inferred"] is False
+    assert parity["producer_queue_failure_gate_inferred"] is False
+    assert parity["promotion_authorized"] is False
+
+    snapshot = yaml.safe_load(PARITY_SNAPSHOT.read_text(encoding="utf-8"))
+    assert snapshot["trace_id"] == parity["trace_id"]
+    assert snapshot["route_parity_terminal"] == "pass"
+    assert snapshot["program_terminal"] == "partial_degraded"
+    assert snapshot["controlled_apply_contract"]["learning_writeback"]["status"] == "partial"
+    assert snapshot["promotion"]["authorized"] is False
+    assert snapshot["production_mutations"]["active_route_changed"] is False
+    assert snapshot["production_mutations"]["secret_read"] is False
+
+
+def test_filelog_fallback_preserves_body_and_exposes_normalization_coverage() -> None:
+    text = OTEL_DAEMONSET.read_text(encoding="utf-8")
+    assert "id: cri_parser" in text
+    assert "on_error: send_quiet" not in text
+    assert 'if: \'body matches "^[^ ]+ (stdout|stderr) [^ ]* .*$"\'' in text
+    assert "id: mark_cri_parsed" in text
+    assert 'value: parsed_cri' in text
+    assert "id: move_cri_body" in text
+    assert "if: attributes.log != nil" in text
+    assert "id: mark_cri_fallback" in text
+    assert 'value: fallback_unparsed' in text
+    assert 'if: attributes["awoooi.log.parse_status"] == nil' in text
+
+    contract = yaml.safe_load(CONTRACT.read_text(encoding="utf-8"))
+    observation = contract["runtime_observations"][
+        "filelog_normalization_error_readback"
+    ]
+    assert observation["transport_queue_size"] == 0
+    assert observation["transport_send_failed_log_records"] == 0
+    assert observation["source_fix"]["preserves_unparsed_body"] is True
+    assert observation["source_fix"]["runtime_canary_passed"] is False
+
+
+def test_filelog_canary_is_synthetic_bounded_and_has_no_network_exporter() -> None:
+    text = FILELOG_CANARY.read_text(encoding="utf-8")
+    documents = list(yaml.safe_load_all(text))
+    config_map, pod = documents
+    assert config_map["kind"] == "ConfigMap"
+    assert "P0OBS_FILELOG_PARSED_CRI" in config_map["data"]["input.log"]
+    assert "P0OBS_FILELOG_FALLBACK_UNPARSED" in config_map["data"]["input.log"]
+    config = config_map["data"]["config.yaml"]
+    assert "on_error: send_quiet" not in config
+    assert 'body matches "^[^ ]+ (stdout|stderr) [^ ]* .*$"' in config
+    assert "fallback_unparsed" in config
+    assert "debug:" in config
+    assert "otlp" not in config.lower()
+
+    spec = pod["spec"]
+    assert spec["automountServiceAccountToken"] is False
+    assert spec["restartPolicy"] == "Never"
+    assert spec["activeDeadlineSeconds"] == 90
+    container = spec["containers"][0]
+    assert "@sha256:" in container["image"]
+    assert container["securityContext"]["allowPrivilegeEscalation"] is False
+    assert container["securityContext"]["readOnlyRootFilesystem"] is True
+
+
 def test_managed_promotion_and_bridge_retirement_fail_closed_without_receipts() -> None:
     env = os.environ.copy()
     env.pop("MANAGED_PROMOTION_RECEIPT", None)