From 4ed6710fe843c36c5f602de1194daeec6bb02c35 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sun, 1 Nov 2026 15:38:46 +0800 Subject: [PATCH] fix: expose bounded candidate publisher failure diagnostics Co-Authored-By: Codex --- docs/unified-daily-notification.md | 15 ++ scripts/emit_unified_digest_candidates.py | 76 +++++++++- tests/test_candidate_failure_diagnostics.py | 153 ++++++++++++++++++++ 3 files changed, 238 insertions(+), 6 deletions(-) create mode 100644 tests/test_candidate_failure_diagnostics.py diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index 0e48ffa..83245aa 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -129,6 +129,21 @@ resource paths and credential context stay out of command output. A permission category identifies the failing storage operation; it does not establish that an alerting Secret Manager denial caused the storage failure. +The scheduled `scripts.emit_unified_digest_candidates` command also emits a +bounded `stage` and `error_code` for binding validation, archive reads, source +selection, schedule evaluation, projection, candidate merge and output +serialization failures. Only exact allowlisted local codes are exposed; an +unclassified failure retains its enclosing stage with `unexpected_error`. +Storage failures retain `error_category` for compatibility and repeat that +category as `error_code`. `uploaded=false` is the existing failure flag, not +proof that no write took effect: an uncertain write or an earlier target may +have persisted. Inspect existing evidence before retrying. + +The coverage-alert fallback keeps its existing delivery attempt and same-run +suppression. Its raw output is suppressed; failed or uncertain delivery emits +only `coverage_alert` / `alert_delivery_unconfirmed` on stderr. This secondary +diagnostic does not replace the original projection failure on stdout. + Account identifiers displayed in logs, notification results and account-selection errors retain at most two trailing characters. Identifiers of two characters or fewer are fully masked. Runtime failure alerts omit the exception detail. diff --git a/scripts/emit_unified_digest_candidates.py b/scripts/emit_unified_digest_candidates.py index 178ed76..b1a0009 100644 --- a/scripts/emit_unified_digest_candidates.py +++ b/scripts/emit_unified_digest_candidates.py @@ -1,11 +1,64 @@ """Project existing private reports; no service or broker invocation.""" from __future__ import annotations +from contextlib import redirect_stderr, redirect_stdout import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import json import os -from notifications.unified import emit_candidates # noqa: E402 +from notifications.unified import CandidateStorageError, emit_candidates # noqa: E402 + + +# Exact local codes only. Never print provider messages, exception class names, +# binding values, or object locators, even when they resemble a known code. +_ERROR_STAGES = { + **dict.fromkeys(( + "candidate_bindings_invalid", "candidate_binding_mode_conflict", + "candidate_target_enablement_unknown", "candidate_target_invalid", + "candidate_store_identity_missing", "candidate_identity_missing", + "candidate_store_configuration_missing", + ), "binding_validation"), + **dict.fromkeys(( + "candidate_archive_read_failed", "candidate_archive_budget_exceeded", + "candidate_source_locator_mismatch", "candidate_archive_time_missing", + ), "archive_read"), + "candidate_source_absent": "source_selection", + "candidate_due_source_absent": "schedule_evaluation", + **dict.fromkeys(( + "candidate_identity_or_time_missing", "candidate_source_identity_mismatch", + "candidate_source_invalid", "candidate_source_time_invalid", + "candidate_source_incomplete", "candidate_run_id_missing", + "candidate_duplicate_run", "candidate_plugin_observations_invalid", + ), "projection"), + **dict.fromkeys(( + "candidate_legacy_aggregate_requires_archive_refresh", + "candidate_plugin_observations_conflict", "candidate_notification_event_conflict", + "candidate_identity_conflict", "candidate_event_source_run_missing", + ), "candidate_merge"), + "candidate_concurrent_update": "candidate_write", +} +_STORAGE_CATEGORIES = frozenset({ + "permission_denied", "generation_conflict", "timeout_unknown", + "serialization_invalid", "storage_unknown", +}) + + +def _failure_diagnostic(error, stage): + if isinstance(error, CandidateStorageError): + stage = error.stage if error.stage in ("candidate_read", "candidate_write") else stage + category = error.category + code = category if type(category) is str and category in _STORAGE_CATEGORIES else "storage_unknown" + return {"stage": stage, "error_code": code, "error_category": code} + code = error.args[0] if len(error.args) == 1 and type(error.args[0]) is str else None + if isinstance(error, (ValueError, RuntimeError)) and code in _ERROR_STAGES: + return {"stage": _ERROR_STAGES[code], "error_code": code} + if isinstance(error, json.JSONDecodeError): + code = "json_invalid" + elif stage == "output_serialization" and isinstance(error, (TypeError, ValueError)): + code = "serialization_invalid" + else: + code = "unexpected_error" + return {"stage": stage, "error_code": code} def _check_binding_modes(env): @@ -21,14 +74,16 @@ def _check_binding_modes(env): def main(): + stage = "binding_validation" try: _check_binding_modes(os.environ) - print(json.dumps(emit_candidates(os.environ), sort_keys=True)) + stage = "projection" + result = emit_candidates(os.environ) + stage = "output_serialization" + print(json.dumps(result, sort_keys=True)) return 0 except Exception as error: - from notifications.unified import CandidateStorageError - diagnostic = ({"stage": error.stage, "error_category": error.category} - if isinstance(error, CandidateStorageError) else {}) + diagnostic = _failure_diagnostic(error, stage) print(json.dumps({"status": "candidate_projection_failed", "uploaded": False, **diagnostic})) from notifications.unified import alert_text, coverage_locale from scripts.execution_report_heartbeat import _send_telegram @@ -38,7 +93,16 @@ def main(): except OSError: attempted = False if not attempted: - _send_telegram(alert_text("candidate_write_failure", coverage_locale())) + # The legacy sender may print raw Secret Manager errors. Preserve its + # delivery attempt without exposing that context or masking the cause. + with open(os.devnull, "w") as sink, redirect_stdout(sink), redirect_stderr(sink): + try: + delivered = _send_telegram(alert_text("candidate_write_failure", coverage_locale())) + except Exception: + delivered = False + if not delivered: + print(json.dumps({"stage": "coverage_alert", "error_code": "alert_delivery_unconfirmed"}), + file=sys.stderr) return 1 if __name__ == "__main__": diff --git a/tests/test_candidate_failure_diagnostics.py b/tests/test_candidate_failure_diagnostics.py new file mode 100644 index 0000000..74b823f --- /dev/null +++ b/tests/test_candidate_failure_diagnostics.py @@ -0,0 +1,153 @@ +"""Publisher diagnostics use fixed vocabulary and synthetic failures only.""" +import json + +import pytest +from google.api_core.exceptions import Forbidden + +from notifications import unified +from scripts import emit_unified_digest_candidates as publisher +from scripts import execution_report_heartbeat as heartbeat +from test_notification_consolidation import ENV, NOW, Store + + +@pytest.fixture(autouse=True) +def offline(monkeypatch): + monkeypatch.delenv("DIGEST_TARGET_BINDINGS_JSON", raising=False) + monkeypatch.delenv("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH", raising=False) + monkeypatch.setattr(heartbeat, "_send_telegram", lambda text: True) + + +def result(capsys): + output = capsys.readouterr() + assert "synthetic-private" not in output.out + output.err + assert "gs://" not in output.out + output.err + return json.loads(output.out) + + +@pytest.mark.parametrize("raw,code", [ + ('{"synthetic-private":', "json_invalid"), + ('{"synthetic-private": true}', "candidate_bindings_invalid"), + ('[{"QSL_ROUTINE_NOTIFICATION_MODE":"immediate"}]', "candidate_binding_mode_conflict"), +]) +def test_binding_failure_is_reported_before_any_projection(raw, code, monkeypatch, capsys): + monkeypatch.setenv("QSL_ROUTINE_NOTIFICATION_MODE", "daily_digest") + monkeypatch.setenv("DIGEST_TARGET_BINDINGS_JSON", raw) + monkeypatch.setattr(publisher, "emit_candidates", lambda env: pytest.fail("unexpected projection")) + assert publisher.main() == 1 + assert result(capsys) == {"status": "candidate_projection_failed", "uploaded": False, + "stage": "binding_validation", "error_code": code} + + +@pytest.mark.parametrize("code,stage", [ + ("candidate_store_identity_missing", "binding_validation"), + ("candidate_archive_read_failed", "archive_read"), + ("candidate_source_absent", "source_selection"), + ("candidate_due_source_absent", "schedule_evaluation"), + ("candidate_identity_conflict", "candidate_merge"), + ("candidate_concurrent_update", "candidate_write"), + ("candidate_source_invalid", "projection"), +]) +def test_known_failure_codes_are_bounded(code, stage, monkeypatch, capsys): + def fail(env): + raise ValueError(code) + monkeypatch.setattr(publisher, "emit_candidates", fail) + assert publisher.main() == 1 + output = result(capsys) + assert output["stage"] == stage and output["error_code"] == code + + +@pytest.mark.parametrize("error", [ + ValueError("candidate_source_absent gs://synthetic-private/object"), + KeyError("synthetic-private-account"), + RuntimeError({"synthetic-private": "credential"}), +]) +def test_unknown_errors_do_not_echo_values_or_guess_a_cause(error, monkeypatch, capsys): + def fail(env): + raise error + monkeypatch.setattr(publisher, "emit_candidates", fail) + assert publisher.main() == 1 + output = result(capsys) + assert output["stage"] == "projection" and output["error_code"] == "unexpected_error" + + +@pytest.mark.parametrize("stage,error,code", [ + ("candidate_read", Forbidden("gs://synthetic-private/object"), "permission_denied"), + ("candidate_write", Forbidden("synthetic-private-account"), "permission_denied"), + ("candidate_write", TimeoutError("synthetic-private"), "timeout_unknown"), + ("candidate_write", ValueError("synthetic-private"), "serialization_invalid"), +]) +def test_real_merge_failure_keeps_stage_and_never_retries_unknown_write(stage, error, code, monkeypatch, capsys): + calls = [] + class FailedStore: + def read(self, uri): + calls.append("read") + if stage == "candidate_read": + raise error + return None, None + def compare_and_set(self, *args): + calls.append("write") + raise error + def emit(env): + return unified._merge(FailedStore(), "gs://synthetic-private/object", {}, {}, idle={"actually_ran": False}) + monkeypatch.setattr(publisher, "emit_candidates", emit) + assert publisher.main() == 1 + output = result(capsys) + assert output["stage"] == stage and output["error_code"] == code + assert output["error_category"] == code # Preserve the existing storage diagnostic. + assert calls == (["read"] if stage == "candidate_read" else ["read", "write"]) + + +def test_tampered_storage_diagnostics_cannot_escape_allowlist(monkeypatch, capsys): + error = unified.CandidateStorageError("gs://synthetic-private/object", RuntimeError()) + error.category = "synthetic-private-account" + monkeypatch.setattr(publisher, "emit_candidates", lambda env: (_ for _ in ()).throw(error)) + assert publisher.main() == 1 + output = result(capsys) + assert output["stage"] == "projection" and output["error_code"] == "storage_unknown" + + +def test_successful_idle_projection_uses_real_producer_without_diagnostics(monkeypatch, capsys): + env = {**ENV, "DIGEST_NOTIFICATION_SCHEDULE_JSON": json.dumps({ + "scheduler": {"main_time": "0 20 1 * *", "timezone": "UTC"}})} + store = Store() + monkeypatch.setattr(publisher, "emit_candidates", lambda _: unified.emit_candidates( + env, now=NOW, read=lambda args: [], store=store)) + assert publisher.main() == 0 + assert result(capsys) == {"status": "projected", "uploaded": True, "targets": 1} + assert len(store.data) == 1 + + +@pytest.mark.parametrize("raises", [False, True]) +def test_alert_fallback_does_not_leak_or_replace_original_diagnostic(raises, monkeypatch, capsys): + def fail(env): + raise ValueError("candidate_source_absent") + def send(text): + import sys + print("gs://synthetic-private/object") + print("synthetic-private-credential", file=sys.stderr) + if raises: + raise RuntimeError("synthetic-private-account") + return False + monkeypatch.setattr(publisher, "emit_candidates", fail) + monkeypatch.setattr(heartbeat, "_send_telegram", send) + assert publisher.main() == 1 + assert result(capsys)["error_code"] == "candidate_source_absent" + + +def test_missing_source_uses_real_projection_and_preserves_empty_store(monkeypatch, capsys): + store = Store() + monkeypatch.setattr(publisher, "emit_candidates", lambda _: unified.emit_candidates( + ENV, now=NOW, read=lambda args: [], store=store)) + assert publisher.main() == 1 + output = result(capsys) + assert output["stage"] == "source_selection" + assert output["error_code"] == "candidate_source_absent" + assert store.data == {} + + +def test_unserializable_output_has_its_own_stage(monkeypatch, capsys): + monkeypatch.setattr(publisher, "emit_candidates", lambda _: {"synthetic-private": object()}) + assert publisher.main() == 1 + output = result(capsys) + assert output["stage"] == "output_serialization" + assert output["error_code"] == "serialization_invalid"