diff --git a/docs/account-archive-history.md b/docs/account-archive-history.md index 80ea0c6..5e8d1c2 100644 --- a/docs/account-archive-history.md +++ b/docs/account-archive-history.md @@ -95,6 +95,21 @@ cache read does not prove that Firstrade still accepts the session: an upstream endpoint returns the sanitized `readonly_client_unavailable` response and logs the fixed diagnostic, without retrying or starting a login. +For a missing/expired cache or an upstream 401, the same log also carries +`recovery_action=manual_session_renewal_required`. Storage read failures, +upstream 403 and upstream 5xx do not imply that renewing authentication will fix +the failure. The HTTP status and response body remain unchanged. + +Use the existing manual `/session-renew` flow only with explicit operational +authorization, existing credentials and the configured non-trading client. +Confirm its outcome before separately invoking the existing cached-only +account-facts refresh. Stop on MFA, rejected authentication or an uncertain +cache write; do not create credentials or automatically retry. The account-facts +route never calls `/session-renew` itself. The generic Invoke Cloud Run workflow +can create/update Scheduler bridges, so it is unsuitable for a no-configuration- +change operation unless those effects have been excluded. Snapshot-only +Execution Report Heartbeat uses the existing bridge without configuration writes. + `Inspect cached account history schema` reads one first page for the current serving runtime's exact account and at most 31 closed UTC dates within the past year. It uses the same configured deployment identity, the existing username diff --git a/docs/unified-daily-notification.md b/docs/unified-daily-notification.md index ca7af7c..012cc24 100644 --- a/docs/unified-daily-notification.md +++ b/docs/unified-daily-notification.md @@ -6,6 +6,23 @@ Chinese or English language, in both notification modes. Actual unknown outcomes, failures, risk, persistence and authentication incidents retain immediate alerts. A candidate or archive write failure produces a separate localized coverage alert without a trade message. +The archive publisher and healthy-heartbeat coverage alert report fixed stage +and reason categories. Stages distinguish binding, storage initialization, +archive listing/read, projection, candidate history read/merge, serialization +and candidate write. Reasons distinguish authentication, permission denial, +missing objects, timeout, transport, invalid data, concurrency and unknown +failures. Exception messages, object paths, provider bodies and identities are +never copied into these diagnostics. The existing alert acknowledgement marker +still suppresses duplicate sends. + +A missing candidate on the initial metadata read is the normal create path; +creation retains `if_generation_match=0`. Only explicit generation conflicts +use the existing bounded merge attempts. An uncertain write is never retried, +and `uploaded=false` after failure does not prove that storage was unchanged +or that earlier targets in the same invocation were not published. Read back +the affected object before any separately authorized retry. A stage/category +alone does not justify changing IAM or replaying old candidate windows. + An accepted order with complete matching evidence is not a confirmed fill; missing, contradictory or failed reconciliation evidence stays attention. diff --git a/main.py b/main.py index e0b5104..d70097c 100644 --- a/main.py +++ b/main.py @@ -135,6 +135,10 @@ def _log_account_facts_failure(reason: object, stage: object = "unclassified", h status = http_status if type(http_status) is int and 100 <= http_status <= 599 else None entry = {"severity": "WARNING", "message": "account_facts_sync failed", "reason": reason, "stage": stage, "http_status": status} + if (reason in {"cache_missing", "cache_stale", "cache_missing_or_stale"} + or (reason == "read_http_error" and status == 401 + and stage in {"userinfo", "account_list", "account_read"})): + entry["recovery_action"] = "manual_session_renewal_required" project = get_project_id() header = request.headers.get("X-Cloud-Trace-Context", "") trace = re.fullmatch(r"([0-9a-fA-F]{32})(?:/[0-9]+)?(?:;o=[01])?", header) diff --git a/notifications/candidate_diagnostics.py b/notifications/candidate_diagnostics.py new file mode 100644 index 0000000..c34b1b3 --- /dev/null +++ b/notifications/candidate_diagnostics.py @@ -0,0 +1,79 @@ +"""Closed diagnostic vocabulary: never include exception text or storage paths.""" +from contextlib import contextmanager + +from google.api_core import exceptions as gcs_errors +from google.auth.exceptions import GoogleAuthError +from requests.exceptions import RequestException, Timeout + + +STAGES = { + "binding": ("绑定校验", "binding validation"), + "storage_init": ("存储初始化", "storage initialization"), + "archive_list": ("归档列表读取", "archive listing"), + "archive_read": ("归档内容读取", "archive read"), + "projection": ("候选投影", "candidate projection"), + "candidate_read": ("候选历史读取", "candidate history read"), + "candidate_merge": ("候选合并", "candidate merge"), + "storage_serialize": ("存储内容序列化", "storage serialization"), + "candidate_write": ("候选写入", "candidate write"), + "unclassified": ("未知阶段", "unknown stage"), +} +REASONS = { + "permission_denied": ("权限被拒绝", "permission denied"), + "authentication": ("认证失败", "authentication failed"), + "not_found": ("对象不存在", "object not found"), + "timeout": ("请求超时，结果未确认", "request timed out; outcome unconfirmed"), + "transport": ("传输失败，结果未确认", "transport failed; outcome unconfirmed"), + "invalid_data": ("数据或配置无效", "invalid data or configuration"), + "conflict": ("并发写入冲突", "concurrent write conflict"), + "unknown": ("原因未确认", "cause unconfirmed"), +} + + +class CandidateOperationError(RuntimeError): + def __init__(self, stage, reason): + self.stage = stage if isinstance(stage, str) and stage in STAGES else "unclassified" + self.reason = reason if isinstance(reason, str) and reason in REASONS else "unknown" + super().__init__(f"{self.stage}:{self.reason}") + + +def classify(exc, stage): + if isinstance(exc, CandidateOperationError): + return CandidateOperationError(exc.stage, exc.reason) + if isinstance(exc, gcs_errors.Forbidden): + reason = "permission_denied" + elif isinstance(exc, (gcs_errors.Unauthorized, GoogleAuthError)): + reason = "authentication" + elif isinstance(exc, gcs_errors.NotFound): + reason = "not_found" + elif isinstance(exc, (TimeoutError, Timeout, gcs_errors.DeadlineExceeded)): + reason = "timeout" + elif isinstance(exc, (RequestException, gcs_errors.ServiceUnavailable, ConnectionError)): + reason = "transport" + elif isinstance(exc, (gcs_errors.PreconditionFailed, gcs_errors.Conflict)): + reason = "conflict" + elif isinstance(exc, (ValueError, TypeError, KeyError)): + reason = "invalid_data" + else: + reason = "unknown" + return CandidateOperationError(stage, reason) + + +@contextmanager +def candidate_stage(stage): + try: + yield + except CandidateOperationError: + raise + except Exception as exc: + raise classify(exc, stage) from None + + +def failure_text(error, locale): + diagnostic = classify(error, "unclassified") + zh = str(locale or "").strip().lower().startswith("zh") + stage = STAGES[diagnostic.stage][0 if zh else 1] + reason = REASONS[diagnostic.reason][0 if zh else 1] + if zh: + return f"通知候选发布失败。\n阶段：{stage}。\n原因：{reason}。\n请查看本次运行记录。" + return f"Notification candidate publication failed.\nStage: {stage}.\nReason: {reason}.\nReview the current run record." diff --git a/notifications/unified.py b/notifications/unified.py index af52982..6fa889b 100644 --- a/notifications/unified.py +++ b/notifications/unified.py @@ -12,6 +12,7 @@ import json import os +from notifications.candidate_diagnostics import CandidateOperationError, candidate_stage, classify from notifications.digest_snapshot import digest_archive from notifications.digest_snapshot.digest_candidates import project_reports from notifications.digest_snapshot.digest_routing import defer_routine_cycle_to_digest @@ -71,8 +72,10 @@ def read(self, uri): def compare_and_set(self, uri, version, value): from google.api_core.exceptions import PreconditionFailed + with candidate_stage("storage_serialize"): + body = json.dumps(value, allow_nan=False) try: - self._blob(uri).upload_from_string(json.dumps(value, allow_nan=False), + self._blob(uri).upload_from_string(body, content_type="application/json", if_generation_match=0 if version is None else version, retry=None, timeout=30) return True @@ -189,9 +192,11 @@ def _aggregate(rows): return result +@candidate_stage("candidate_merge") def _merge(store, uri, rows, events, *, idle=None, archive=False): for _ in range(8): - version, previous = store.read(uri) + with candidate_stage("candidate_read"): + version, previous = store.read(uri) # Read the former local ledger once for migration; new payloads store # all merge history in their per-run rows, as required by QRS. sources = dict((previous or {}).get("notification_source_runs") or {}) @@ -244,9 +249,10 @@ def _merge(store, uri, rows, events, *, idle=None, archive=False): "notification_events": by_run[run_id]} for run_id in sorted(sources)] payload = dict(schema_version="qsl.digest_candidates.v1", producer_status="projected", runs=run_rows if sources else [idle] if idle else []) - if store.compare_and_set(uri, version, payload): - return True - raise RuntimeError("candidate_concurrent_update") + with candidate_stage("candidate_write"): + if store.compare_and_set(uri, version, payload): + return True + raise CandidateOperationError("candidate_merge", "conflict") def persist_report_candidate(report, fact, *, env, now, store): @@ -261,14 +267,19 @@ def persist_report_candidate(report, fact, *, env, now, store): return _merge(store, uri, {report["run_id"]: row}, {fact["dedupe_key"]: fact}) +@candidate_stage("projection") def emit_candidates(env, *, now=None, read=digest_archive._read, store=None): current = now or datetime.now(timezone.utc) - store = store or GcsJsonStore() + with candidate_stage("storage_init"): + store = store or GcsJsonStore() + with candidate_stage("binding"): + bindings = _bindings(env) results = [] - for binding in _bindings(env): + for binding in bindings: reports = [] def capture(args): - value = read(args) + with candidate_stage("archive_list" if args[0] == "ls" else "archive_read"): + value = read(args) if args[0] == "cat": reports.append(value) return value @@ -488,21 +499,33 @@ def send_coverage(): def notify_heartbeat(*, kind, send, now, env=None, store=None, recovery_reports=(), read=digest_archive._read, - incident_reason="execution_receipt", affected_keys=None, service_name=None): + incident_reason="execution_receipt", affected_keys=None, service_name=None, + send_candidate_failure=None): env = os.environ if env is None else env was_healthy = kind == "healthy_heartbeat" attempted = False + diagnostic = None + original_send = send + def send(): + if was_healthy and diagnostic is not None and send_candidate_failure is not None: + return send_candidate_failure(diagnostic) + return original_send() if not enabled(env) and kind != "healthy_heartbeat": return {"candidate_persisted": False, "fallback_acknowledged": send() is True} try: - store = store or GcsJsonStore() + with candidate_stage("storage_init"): + store = store or GcsJsonStore() persisted = False try: persisted = emit_candidates(env, now=now, read=read, store=store)["uploaded"] is True - except Exception: + except Exception as exc: + diagnostic = classify(exc, "projection") + print(json.dumps({"status": "candidate_projection_failed", + "stage": diagnostic.stage, "reason": diagnostic.reason}), flush=True) if kind == "healthy_heartbeat": kind, incident_reason = "real_error", "notification_candidate_write_failure" if kind == "healthy_heartbeat" and not persisted: + diagnostic = CandidateOperationError("projection", "unknown") kind, incident_reason = "real_error", "notification_candidate_write_failure" decisions = [] for binding in _bindings(env): @@ -559,7 +582,9 @@ def notify_heartbeat(*, kind, send, now, env=None, store=None, recovery_reports= for tracker, decision, event in claimed: delivery.append(_record_ack(tracker, decision, outcome, event=event, related_ids=related_ids)) return {"candidate_persisted": persisted, "deliveries": delivery} - except Exception: + except Exception as exc: + if diagnostic is None: + diagnostic = classify(exc, "unclassified") print("notification_consolidation_failed; coverage_unknown", flush=True) if was_healthy and not attempted: try: diff --git a/scripts/emit_unified_digest_candidates.py b/scripts/emit_unified_digest_candidates.py index 7fc4ed5..6b875f6 100644 --- a/scripts/emit_unified_digest_candidates.py +++ b/scripts/emit_unified_digest_candidates.py @@ -6,6 +6,7 @@ import json import os from notifications.unified import emit_candidates # noqa: E402 +from notifications.candidate_diagnostics import candidate_stage, classify, failure_text def _check_binding_modes(env): @@ -22,12 +23,15 @@ def _check_binding_modes(env): def main(): try: - _check_binding_modes(os.environ) + with candidate_stage("binding"): + _check_binding_modes(os.environ) print(json.dumps(emit_candidates(os.environ), sort_keys=True)) return 0 - except Exception: - print(json.dumps({"status": "candidate_projection_failed", "uploaded": False})) - from notifications.unified import alert_text, coverage_locale + except Exception as exc: + diagnostic = classify(exc, "projection") + print(json.dumps({"status": "candidate_projection_failed", "uploaded": False, + "stage": diagnostic.stage, "reason": diagnostic.reason})) + from notifications.unified import coverage_locale from scripts.execution_report_heartbeat import _send_telegram marker = os.environ.get("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH") try: @@ -35,7 +39,7 @@ def main(): except OSError: attempted = False if not attempted: - _send_telegram(alert_text("candidate_write_failure", coverage_locale())) + _send_telegram(failure_text(diagnostic, coverage_locale())) return 1 if __name__ == "__main__": diff --git a/scripts/execution_report_heartbeat.py b/scripts/execution_report_heartbeat.py index af515a8..bab440c 100644 --- a/scripts/execution_report_heartbeat.py +++ b/scripts/execution_report_heartbeat.py @@ -771,17 +771,19 @@ def _deliver_alert(message: str, *, kind: str, now: dt.datetime, affected_keys=N def _send_normal_heartbeat(name: str, detail: str, *, quiet_eligible: bool = True, now: dt.datetime | None = None, recovery_reports=(), affected_keys=None) -> int: from notifications.unified import alert_text, coverage_locale, enabled, notify_heartbeat + from notifications.candidate_diagnostics import failure_text if quiet_eligible or enabled(): locale = coverage_locale() if quiet_eligible else _notification_locale() message = alert_text("candidate_write_failure" if quiet_eligible else "receipt_unacceptable", locale) - def send_attention(): + def send_attention(diagnostic=None): marker = os.environ.get("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH") if quiet_eligible and marker: try: Path(marker).write_text("attempted\n") except OSError: pass - acknowledged = _send_telegram(message[:3900]) + text = failure_text(diagnostic, locale) if diagnostic is not None else message + acknowledged = _send_telegram(text[:3900]) if quiet_eligible and acknowledged is True and marker: try: Path(marker).write_text("acknowledged\n") @@ -790,7 +792,8 @@ def send_attention(): return acknowledged result = notify_heartbeat(kind="healthy_heartbeat" if quiet_eligible else "receipt_unacceptable", send=send_attention, now=now or dt.datetime.now(dt.timezone.utc), - recovery_reports=recovery_reports, affected_keys=affected_keys) + recovery_reports=recovery_reports, affected_keys=affected_keys, + send_candidate_failure=send_attention) return 1 if result.get("failed") or not result.get("candidate_persisted") else 0 if quiet_eligible and not _env_bool("RUNTIME_HEARTBEAT_NOTIFY_ON_SUCCESS", False): return 0 diff --git a/tests/test_account_facts_endpoint.py b/tests/test_account_facts_endpoint.py index 0dff80e..c6a5957 100644 --- a/tests/test_account_facts_endpoint.py +++ b/tests/test_account_facts_endpoint.py @@ -31,6 +31,7 @@ def failed_client(): assert response.json == {"ok": False, "error": "readonly_client_unavailable"} line = capsys.readouterr().out.strip() payload = json.loads(line) + assert payload["recovery_action"] == "manual_session_renewal_required" trace = payload.pop("logging.googleapis.com/trace") entry = {"resource": {"type": "cloud_run_revision", "labels": { "project_id": PROJECT, "location": "us-central1", "service_name": SERVICE}}, @@ -43,6 +44,36 @@ def failed_client(): assert "synthetic-native-id" not in line +@pytest.mark.parametrize("reason,stage,status,renewal", [ + ("cache_stale", "cache", None, True), + ("cache_missing", "cache", None, True), + ("read_http_error", "userinfo", 401, True), + ("read_http_error", "userinfo", 403, False), + ("read_http_error", "userinfo", 503, False), + ("cache_read_failed", "cache", None, False), +]) +def test_cached_failure_requires_separate_manual_auth_flow( + monkeypatch, capsys, reason, stage, status, renewal, +): + _configure(monkeypatch) + calls = [] + def fail_client(): + calls.append("cached_read") + raise FirstradeCachedSessionError(reason, stage=stage, http_status=status) + def forbidden(*args, **kwargs): + pytest.fail("failed cached read must not renew or publish") + monkeypatch.setattr(main, "READ_ONLY_ACCOUNT_FACTS_CLIENT_BUILDER", fail_client) + monkeypatch.setattr(main, "run_forced_session_renewal", forbidden) + monkeypatch.setattr(main, "publish_firstrade_account_snapshot", forbidden) + response = main.app.test_client().post("/account-facts-sync", headers={ + "Authorization": "Bearer synthetic-facts-token"}) + assert response.status_code == 503 + assert response.json == {"ok": False, "error": "readonly_client_unavailable"} + assert calls == ["cached_read"] + record = json.loads(capsys.readouterr().out) + assert (record.get("recovery_action") == "manual_session_renewal_required") is renewal + + @pytest.mark.parametrize("header,project", [(None, "example-project"), ("bad", "example-project"), ("a" * 32, None), ("a" * 32 + "/1;o=1", "bad/project")]) diff --git a/tests/test_candidate_failure_diagnostics.py b/tests/test_candidate_failure_diagnostics.py new file mode 100644 index 0000000..3f3d69b --- /dev/null +++ b/tests/test_candidate_failure_diagnostics.py @@ -0,0 +1,171 @@ +"""Offline failure injection at the candidate persistence boundary.""" +import json +from types import SimpleNamespace + +import pytest +from google.api_core.exceptions import Forbidden, NotFound, PreconditionFailed + +from notifications import unified +from scripts import emit_unified_digest_candidates as entry +from test_notification_consolidation import ENV, NOW, Store + + +def idle_env(): + return {**ENV, "DIGEST_NOTIFICATION_SCHEDULE_JSON": json.dumps({ + "scheduler": {"main_time": "0 0 1 1 *", "timezone": "UTC"}})} + + +def test_first_candidate_after_not_found_uses_create_precondition(): + writes = [] + + class Blob: + def reload(self, **kwargs): + raise NotFound("synthetic-private-location") + + def upload_from_string(self, body, **kwargs): + writes.append((json.loads(body), kwargs)) + + client = SimpleNamespace(bucket=lambda _: SimpleNamespace(blob=lambda _: Blob())) + result = unified.emit_candidates(idle_env(), now=NOW, read=lambda _: [], + store=unified.GcsJsonStore(client)) + assert result["uploaded"] is True + assert len(writes) == 1 + body, options = writes[0] + assert options["if_generation_match"] == 0 and options["retry"] is None + assert body["runs"][0]["actually_ran"] is False + assert "order_count" not in body["runs"][0] + + +@pytest.mark.parametrize("operation,error,stage,reason", [ + ("read", Forbidden("synthetic-private-location"), "candidate_read", "permission_denied"), + ("cas", Forbidden("synthetic-private-location"), "candidate_write", "permission_denied"), + ("cas", TimeoutError("synthetic-private-token"), "candidate_write", "timeout"), + ("read", ValueError("synthetic-private-body"), "candidate_read", "invalid_data"), +]) +def test_entry_reports_redacted_stage_and_never_retries_unknown_io( + monkeypatch, capsys, operation, error, stage, reason, +): + from scripts import execution_report_heartbeat as heartbeat + calls, alerts = [], [] + + class FailedStore(Store): + def read(self, uri): + calls.append("read") + if operation == "read": + raise error + return None, None + + def compare_and_set(self, *args): + calls.append("cas") + raise error + + monkeypatch.delenv("DIGEST_TARGET_BINDINGS_JSON", raising=False) + monkeypatch.delenv("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH", raising=False) + monkeypatch.setenv("QSL_NOTIFY_LANG", "zh") + monkeypatch.setattr(entry, "emit_candidates", lambda _: unified.emit_candidates( + idle_env(), now=NOW, read=lambda _: [], store=FailedStore())) + monkeypatch.setattr(heartbeat, "_send_telegram", lambda text: alerts.append(text)) + assert entry.main() == 1 + output = capsys.readouterr().out + result = json.loads(output) + assert result["stage"] == stage and result["reason"] == reason + assert result["uploaded"] is False + assert calls == (["read"] if operation == "read" else ["read", "cas"]) + assert len(alerts) == 1 + assert "阶段" in alerts[0] and "原因" in alerts[0] + assert "synthetic-private" not in output + alerts[0] + + +def test_storage_serialization_failure_is_distinct_and_does_not_upload(): + from notifications.candidate_diagnostics import CandidateOperationError + client = SimpleNamespace(bucket=lambda _: pytest.fail("serialization must precede I/O")) + with pytest.raises(CandidateOperationError) as caught: + unified.GcsJsonStore(client).compare_and_set("gs://synthetic/candidate", None, + {"value": float("nan")}) + assert caught.value.stage == "storage_serialize" + assert caught.value.reason == "invalid_data" + + +def test_cas_conflicts_remain_bounded_and_distinct(): + from notifications.candidate_diagnostics import CandidateOperationError + calls = [] + + class Blob: + def reload(self, **kwargs): + raise NotFound("synthetic-private-location") + + def upload_from_string(self, *args, **kwargs): + calls.append(1) + raise PreconditionFailed("synthetic-private-location") + + client = SimpleNamespace(bucket=lambda _: SimpleNamespace(blob=lambda _: Blob())) + with pytest.raises(CandidateOperationError) as caught: + unified.emit_candidates(idle_env(), now=NOW, read=lambda _: [], + store=unified.GcsJsonStore(client)) + assert caught.value.stage == "candidate_merge" + assert caught.value.reason == "conflict" + assert len(calls) == 8 + + +def test_existing_alert_ack_still_suppresses_duplicate(monkeypatch, tmp_path, capsys): + from notifications.candidate_diagnostics import CandidateOperationError + from scripts import execution_report_heartbeat as heartbeat + marker = tmp_path / "ack" + marker.write_text("acknowledged") + monkeypatch.setenv("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH", str(marker)) + monkeypatch.delenv("DIGEST_TARGET_BINDINGS_JSON", raising=False) + monkeypatch.setattr(entry, "emit_candidates", lambda _: (_ for _ in ()).throw( + CandidateOperationError("candidate_write", "unknown"))) + monkeypatch.setattr(heartbeat, "_send_telegram", lambda _: pytest.fail("duplicate alert")) + assert entry.main() == 1 + assert json.loads(capsys.readouterr().out)["stage"] == "candidate_write" + + +@pytest.mark.parametrize("locale", ["zh", "en"]) +def test_heartbeat_first_alert_has_stage_and_retains_ack_dedup(monkeypatch, tmp_path, capsys, locale): + from notifications.candidate_diagnostics import CandidateOperationError + from scripts import execution_report_heartbeat as heartbeat + for key, value in ENV.items(): + monkeypatch.setenv(key, value) + monkeypatch.delenv("DIGEST_TARGET_BINDINGS_JSON", raising=False) + monkeypatch.setenv("QSL_NOTIFY_LANG", locale) + marker = tmp_path / "coverage-ack" + monkeypatch.setenv("RUNTIME_HEARTBEAT_COVERAGE_ACK_PATH", str(marker)) + monkeypatch.setattr(unified, "GcsJsonStore", lambda: Store()) + def fail(*args, **kwargs): + raise CandidateOperationError("candidate_write", "permission_denied") + monkeypatch.setattr(unified, "emit_candidates", fail) + alerts = [] + monkeypatch.setattr(heartbeat, "_send_telegram", lambda text: alerts.append(text) or True) + assert heartbeat._send_normal_heartbeat("synthetic", "private-detail", now=NOW) == 1 + assert marker.read_text().strip() == "acknowledged" + monkeypatch.setattr(entry, "emit_candidates", fail) + assert entry.main() == 1 + assert len(alerts) == 1 + assert ("候选写入" if locale == "zh" else "candidate write") in alerts[0] + assert ("权限被拒绝" if locale == "zh" else "permission denied") in alerts[0] + assert "private-detail" not in alerts[0] + capsys.readouterr().out + + +def test_archive_read_failure_never_reaches_candidate_store(): + from notifications.candidate_diagnostics import CandidateOperationError + store = Store() + def fail(_): + raise ValueError("synthetic-private-archive-output") + with pytest.raises(CandidateOperationError) as caught: + unified.emit_candidates(idle_env(), now=NOW, read=fail, store=store) + assert caught.value.stage == "archive_list" + assert caught.value.reason == "invalid_data" + assert store.calls == [] + + +def test_unknown_exceptions_and_untrusted_diagnostics_are_not_echoed(): + from notifications.candidate_diagnostics import CandidateOperationError, classify, failure_text + error = RuntimeError("synthetic-private-body\nAuthorization: Bearer synthetic-secret") + diagnostic = classify(error, "projection") + assert diagnostic.reason == "unknown" + forged = CandidateOperationError("private-stage", "private-reason") + forged.stage, forged.reason = "private-stage", "private-reason" + for locale in ("zh", "en"): + text = failure_text(diagnostic, locale) + failure_text(forged, locale) + assert "private" not in text and "synthetic-secret" not in text diff --git a/tests/test_daily_notification_policy.py b/tests/test_daily_notification_policy.py index 324bcac..d690161 100644 --- a/tests/test_daily_notification_policy.py +++ b/tests/test_daily_notification_policy.py @@ -344,7 +344,10 @@ def test_scheduled_exporter_failure_sends_only_localized_coverage(monkeypatch): sent = [] monkeypatch.setattr(heartbeat, "_send_telegram", lambda text: sent.append(text) or True) assert emitter.main() == 1 - assert sent == [unified.alert_text("candidate_write_failure", "en")] + assert sent == ["Notification candidate publication failed.\n" + "Stage: candidate projection.\n" + "Reason: invalid data or configuration.\n" + "Review the current run record."] assert "private URI" not in sent[0]