fix: expose bounded candidate publisher failure diagnostics

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.

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

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"