fix: classify cached auth and candidate persistence failures

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

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.
 

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)

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."

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:

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,20 +23,23 @@ 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:
             attempted = bool(marker and Path(marker).read_text().strip() in {"attempted", "acknowledged"})
         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__":

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

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")])