fix: redact account displays to two trailing characters

application/firstrade_client.py

@@ -144,10 +144,11 @@ def is_live_trading_enabled(env: Callable[[str, str | None], str | None] = os.ge
 
 
 def mask_account_id(account_id: str) -> str:
+    """Expose at most two trailing characters; fully mask shorter identifiers."""
     value = str(account_id or "")
-    if len(value) <= 4:
+    if len(value) <= 2:
         return "*" * len(value)
-    return f"{'*' * max(0, len(value) - 4)}{value[-4:]}"
+    return f"{'*' * (len(value) - 2)}{value[-2:]}"
 
 def normalize_account_number(value: object) -> str | None:
     """Coerce Firstrade account identifiers to comparable non-empty strings."""

application/rebalance_service.py

@@ -28,7 +28,9 @@
 )
 from application.runtime_broker_adapters import build_runtime_broker_adapters
 from application.signal_snapshot import build_signal_snapshot
-from application.state_persistence import GcsStateStore, build_gcs_state_store_from_env
+from application.state_persistence import (
+    GcsStateStore, build_gcs_state_store_from_env, legacy_account_state_id,
+)
 from application.strategy_run_persistence import (
     build_strategy_run_state,
     claim_live_strategy_run,
@@ -628,17 +630,18 @@ def log_message(message: str) -> None:
         evaluation_metadata=getattr(evaluation, "metadata", None),
     )
     masked_account = mask_account_id(account)
+    state_account = legacy_account_state_id(account)
     existing_run = None
     if persist_strategy_runs and not settings.dry_run_only:
         existing_run = read_latest_strategy_run_state(
             store=store,
-            account=masked_account,
+            account=state_account,
             strategy_profile=strategy_runtime.profile,
             run_period=run_period,
         )
         existing_submission_claim = read_live_strategy_run_claim(
             store=store,
-            account=masked_account,
+            account=state_account,
             strategy_profile=strategy_runtime.profile,
             run_period=run_period,
         )
@@ -678,6 +681,7 @@ def log_message(message: str) -> None:
                     strategy_run_persisted = persist_strategy_run_state(
                         store=store,
                         state=duplicate_state,
+                        storage_account=state_account,
                         now=now,
                     )
                 except Exception as exc:
@@ -745,6 +749,7 @@ def log_message(message: str) -> None:
             strategy_run_persisted = persist_strategy_run_state(
                 store=store,
                 state=planned_state,
+                storage_account=state_account,
                 now=now,
             )
         except Exception as exc:
@@ -763,7 +768,7 @@ def acquire_submission_claim() -> bool:
             return True
         submission_claim_acquired = claim_live_strategy_run(
             store=store,
-            account=masked_account,
+            account=state_account,
             strategy_profile=strategy_runtime.profile,
             run_period=run_period,
             now=now,
@@ -792,7 +797,7 @@ def acquire_submission_claim() -> bool:
     if execution_result.idempotency_blocked:
         existing_run = read_latest_strategy_run_state(
             store=store,
-            account=masked_account,
+            account=state_account,
             strategy_profile=strategy_runtime.profile,
             run_period=run_period,
         ) or {
@@ -965,6 +970,7 @@ def acquire_submission_claim() -> bool:
             result["strategy_run_persisted"] = persist_strategy_run_state(
                 store=store,
                 state=completed_state,
+                storage_account=state_account,
                 now=now,
             )
         except Exception as exc:

application/session_check_service.py

@@ -22,7 +22,9 @@
     is_live_trading_enabled,
     mask_account_id,
 )
-from application.state_persistence import GcsStateStore, build_gcs_state_store_from_env
+from application.state_persistence import (
+    GcsStateStore, build_gcs_state_store_from_env, legacy_account_state_id,
+)
 from strategy_registry import (
     FIRSTRADE_PLATFORM,
     resolve_strategy_definition,
@@ -162,7 +164,7 @@ def _session_check_state_key(
     env: Callable[[str, str | None], str | None] = os.getenv,
 ) -> str:
     account_selector = str(env("FIRSTRADE_ACCOUNT", "") or "").strip()
-    account_key = _safe_key(mask_account_id(account_selector)) if account_selector else "auto"
+    account_key = _safe_key(legacy_account_state_id(account_selector)) if account_selector else "auto"
     return (
         f"session-checks/{account_key}/{_safe_key(strategy_profile)}/"
         f"{_safe_key(period)}/latest.json"

application/state_persistence.py

@@ -12,6 +12,14 @@ class StatePersistenceError(RuntimeError):
     """Raised when the configured state store cannot read or write."""
 
 
+def legacy_account_state_id(account: str) -> str:
+    """Keep existing private state/claim locators stable; never use for display."""
+    value = str(account or "")
+    if len(value) <= 4:
+        return "*" * len(value)
+    return "*" * (len(value) - 4) + value[-4:]
+
+
 def _normalize_prefix(prefix: str | None) -> str:
     return str(prefix or "").strip("/")
 
@@ -51,15 +59,15 @@ def read_json(self, key: str) -> dict[str, Any] | None:
             error_msg = str(exc)
             if "404" in error_msg or "NOT_FOUND" in error_msg.upper():
                 return None
-            raise StatePersistenceError(f"GCS read failed for {key}: {exc}") from exc
+            raise StatePersistenceError("GCS state read failed") from None
         if text is None:
             return None
         try:
             payload = json.loads(text)
         except json.JSONDecodeError as exc:
-            raise StatePersistenceError(f"GCS payload for {key} is not valid JSON: {exc}") from exc
+            raise StatePersistenceError("GCS state payload is not valid JSON") from None
         if not isinstance(payload, dict):
-            raise StatePersistenceError(f"GCS payload for {key} is not a JSON object.")
+            raise StatePersistenceError("GCS state payload is not a JSON object.")
         return payload
 
     def write_json(self, key: str, payload: dict[str, Any]) -> bool:
@@ -68,7 +76,7 @@ def write_json(self, key: str, payload: dict[str, Any]) -> bool:
         try:
             _object_store().write_text(uri, data)
         except Exception as exc:
-            raise StatePersistenceError(f"GCS write failed for {key}: {exc}") from exc
+            raise StatePersistenceError("GCS state write failed") from None
         return True
 
     def create_json(self, key: str, payload: dict[str, Any]) -> bool:
@@ -77,7 +85,7 @@ def create_json(self, key: str, payload: dict[str, Any]) -> bool:
         data = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
         bucket_name, separator, object_name = uri[5:].partition("/")
         if not separator or not bucket_name or not object_name:
-            raise StatePersistenceError(f"Invalid GCS state URI for {key}")
+            raise StatePersistenceError("Invalid GCS state URI")
         try:
             from google.api_core.exceptions import Conflict, PreconditionFailed
             from google.cloud import storage
@@ -97,7 +105,7 @@ def create_json(self, key: str, payload: dict[str, Any]) -> bool:
         except (Conflict, PreconditionFailed):
             return False
         except Exception as exc:
-            raise StatePersistenceError(f"GCS atomic create failed for {key}: {exc}") from exc
+            raise StatePersistenceError("GCS state atomic create failed") from None
 
 
 def build_gcs_state_store_from_env(

application/strategy_run_persistence.py

@@ -9,6 +9,7 @@
 from typing import Any
 
 from application.state_persistence import GcsStateStore
+from application.firstrade_client import mask_account_id
 from quant_platform_kit.common.execution_outcomes import (
     DEFAULT_TERMINAL_STRATEGY_RUN_STAGES,
     STAGE_NO_ACTION,
@@ -135,7 +136,7 @@ def claim_live_strategy_run(
     payload = {
         "stage": "PENDING_SUBMISSION",
         "submission_claim_boundary": LIVE_SUBMISSION_CLAIM_BOUNDARY,
-        "account": account,
+        "account": mask_account_id(account),
         "strategy_profile": strategy_profile,
         "run_period": run_period,
         "as_of": (now or utcnow()).isoformat(),
@@ -233,7 +234,7 @@ def build_strategy_run_state(
     payload = {
         "stage": stage,
         "as_of": as_of.isoformat(),
-        "account": account,
+        "account": mask_account_id(account),
         "strategy_profile": strategy_profile,
         "strategy_display_name": strategy_display_name,
         "run_period": run_period,
@@ -260,10 +261,11 @@ def persist_strategy_run_state(
     *,
     store: GcsStateStore,
     state: Mapping[str, Any],
+    storage_account: str | None = None,
     now: datetime | None = None,
 ) -> bool:
     as_of = now or utcnow()
-    account = str(state.get("account") or "unknown")
+    account = storage_account if storage_account is not None else str(state.get("account") or "unknown")
     strategy_profile = str(state.get("strategy_profile") or "unknown")
     run_period = str(state.get("run_period") or f"{as_of.year:04d}-{as_of.month:02d}")
     stage = str(state.get("stage") or "UNKNOWN")

docs/unified-daily-notification.md

@@ -128,3 +128,12 @@ Candidate persistence failures expose only a fixed `candidate_read` or
 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.
+
+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.
+
+Private state object and submission-claim locators retain their existing identity
+format so a display change cannot bypass an existing duplicate-submission lock
+or session-maintenance marker. New state payload account fields use the two-character
+mask. State-store errors omit object locators and provider exception details.

notifications/telegram.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+from application.firstrade_client import mask_account_id
 from quant_platform_kit.common.display_names import display_name
 
 from quant_platform_kit.common.operational_notification_localization import resolve_operational_notification_locale
@@ -1033,7 +1034,7 @@ def render_cycle_summary(result: Mapping[str, Any], *, lang: str = "en") -> str:
     if strategy_name:
         lines.append(translator("strategy_label", name=strategy_name))
     if account:
-        lines.append(translator("account_label", account=account))
+        lines.append(translator("account_label", account=mask_account_id(account)))
     if not submitted:
         account_snapshot = result.get("heartbeat_account_snapshot")
         account_snapshot = account_snapshot if isinstance(account_snapshot, Mapping) else {}

tests/test_account_alert_redaction.py

@@ -0,0 +1,114 @@
+"""Exercise real HTTP error handling and Telegram formatting with offline I/O."""
+from copy import deepcopy
+import json
+from types import SimpleNamespace
+
+import pytest
+import requests
+
+import main
+from application.firstrade_client import FirstradeBrokerClient, FirstradeCredentials
+from notifications import unified
+
+
+class MemoryStore:
+    def __init__(self):
+        self.data = {}
+
+    def read(self, key):
+        return deepcopy(self.data.get(key, (None, None)))
+
+    def compare_and_set(self, key, version, value):
+        current = self.data.get(key, (None, None))[0]
+        if current != version:
+            return False
+        self.data[key] = ((current or 0) + 1, deepcopy(value))
+        return True
+
+
+@pytest.mark.parametrize("lang", ["en", "zh"])
+@pytest.mark.parametrize("unified_enabled", [False, True])
+def test_account_selection_failure_through_real_alert_sender(
+    monkeypatch, capsys, lang, unified_enabled,
+):
+    # Synthetic data only. Keep select_account, HTTP error handling, notification
+    # routing and build_sender real; replace broker/storage/network boundaries.
+    account = "87654321"
+    client = FirstradeBrokerClient(FirstradeCredentials(username="synthetic", password="synthetic"))
+    monkeypatch.setattr(client, "account_numbers", lambda: [])
+    monkeypatch.setattr(main, "run_session_check", lambda: client.select_account(account))
+    target = dict(platform_id="firstrade", strategy_profile="synthetic",
+                  service_name="synthetic-service", account_scope="synthetic-scope",
+                  account_selector=[account], deployment_selector="synthetic-slot", dry_run_only=False)
+    env = {
+        "FIRSTRADE_RUN_SESSION_CHECK_ON_HTTP": "true",
+        unified.FLAG: str(unified_enabled).lower(),
+        "QSL_ROUTINE_NOTIFICATION_MODE": "daily_digest",
+        "RUNTIME_TARGET_ENABLED": "true", "RUNTIME_TARGET_JSON": json.dumps(target),
+        "DIGEST_STORE_TARGET": "synthetic", "DIGEST_OPAQUE_ACCOUNT_UID": "opaque-synthetic",
+        "DIGEST_TARGET_ID": "target-synthetic", "DIGEST_CANDIDATES_GCS_URI": "gs://synthetic/candidates",
+        # No archive source: the immediate incident path must still work.
+        "EXECUTION_REPORT_GCS_URI": "", "DIGEST_TARGET_BINDINGS_JSON": "",
+        "QSL_NOTIFICATION_MARKER_GCS_URI": "", "DIGEST_NOTIFICATION_SCHEDULE_JSON": "",
+        "K_SERVICE": "synthetic-service", "ACCOUNT_REGION": "synthetic-region",
+        "STRATEGY_DISPLAY_NAME": "synthetic", "QSL_NOTIFY_LANG": lang,
+    }
+    for key, value in env.items():
+        monkeypatch.setenv(key, value)
+    monkeypatch.setattr(main, "_get_telegram_token", lambda: "synthetic-token")
+    monkeypatch.delenv("QSL_GLOBAL_TELEGRAM_CHAT_ID", raising=False)
+    monkeypatch.setenv("GLOBAL_TELEGRAM_CHAT_ID", "synthetic-chat")
+    store = MemoryStore()
+    monkeypatch.setattr(unified, "GcsJsonStore", lambda: store)
+    sent = []
+
+    def post(url, *, json, timeout):
+        assert url == "https://api.telegram.org/botsynthetic-token/sendMessage"
+        assert timeout == 15
+        sent.append(json)
+        return SimpleNamespace(status_code=200, json=lambda: {"ok": True})
+
+    monkeypatch.setattr(requests, "post", post)
+    response = main.app.test_client().post("/probe")
+    assert response.status_code == 500
+    payload = response.get_json()
+    assert payload["ok"] is False
+    assert payload["runtime_error_notification_attempted"] is True
+    assert "******21" in payload["error"]
+    assert len(sent) == 1
+    assert sent[0]["chat_id"] == "synthetic-chat"
+    assert ("health check failed" if lang == "en" else "健康检查失败") in sent[0]["text"]
+    assert "Requested account" not in sent[0]["text"]
+    surfaces = [response.get_data(as_text=True), json.dumps(sent), capsys.readouterr().out]
+    for surface in surfaces:
+        assert account not in surface
+        assert account[-4:] not in surface
+    if unified_enabled:
+        assert any("/_notification_markers/" in key for key in store.data)
+        assert any("/_notification_events/" in key for key in store.data)
+
+
+@pytest.mark.parametrize("account", ["87654321", "****4321"])
+@pytest.mark.parametrize("lang", ["en", "zh"])
+def test_cycle_alert_redacts_raw_and_legacy_accounts_at_log_and_transport(monkeypatch, account, lang):
+    from application.rebalance_service import _publish_cycle_notification
+
+    sent, logged = [], []
+
+    def post(url, *, json, timeout):
+        assert url == "https://api.telegram.org/botsynthetic-token/sendMessage"
+        sent.append(json)
+        return SimpleNamespace(status_code=200, json=lambda: {"ok": True})
+
+    monkeypatch.setattr(requests, "post", post)
+    result = {"account": account, "strategy_profile": "synthetic", "portfolio": {},
+              "execution": {}, "submitted_orders": [], "skipped_orders": [],
+              "execution_blocked": True, "execution_blocking_skips": [{"reason": "broker_rejected"}]}
+    settings = SimpleNamespace(tg_token="synthetic-token", tg_chat_id="synthetic-chat", notify_lang=lang)
+    assert _publish_cycle_notification(result, settings=settings, log_message=logged.append) is True
+    assert len(sent) == 1
+    assert logged and "******21" in "\n".join(logged)
+    for surface in (json.dumps(sent), "\n".join(logged)):
+        assert "87654321" not in surface
+        assert "4321" not in surface
+    assert result["account"] == account  # Broker-facing input remains intact.

tests/test_firstrade_client.py

@@ -91,8 +91,24 @@ def build_fake_client(live=False):
     ).connect()
 
 
-def test_mask_account_id_keeps_only_last_four_digits():
-    assert mask_account_id("12345678") == "****5678"
+@pytest.mark.parametrize("value, expected", [
+    (None, ""), ("", ""), ("7", "*"), ("78", "**"),
+    ("678", "*78"), ("5678", "**78"), ("12345678", "******78"),
+    ("synthetic-78", "**********78"), ("****5678", "******78"),
+])
+def test_mask_account_id_keeps_at_most_last_two_characters(value, expected):
+    assert mask_account_id(value) == expected
+    assert mask_account_id(expected) == expected
+
+
+def test_account_summaries_and_selection_errors_use_last_two_characters():
+    client = build_fake_client()
+    assert client.list_account_summaries()[0]["account"] == "******78"
+    with pytest.raises(FirstradePlatformError) as caught:
+        client.select_account("87654321")
+    assert "******21" in str(caught.value)
+    assert "4321" not in str(caught.value)
+    assert "87654321" not in str(caught.value)
 
 
 def test_notional_live_order_rejects_size_above_local_cap():