fix: archive cached account facts before publishing

application/session_check_service.py

@@ -359,6 +359,45 @@ def persist_funds_snapshot(
     return True
 
 
+def archive_cached_account_facts(
+    observation: Mapping[str, Any],
+    *,
+    env_reader: Callable[[str, str | None], str | None] = os.getenv,
+) -> None:
+    """Archive a validated cached-only observation using its original time.
+
+    The endpoint validates identity and publishable facts before calling this.
+    Never issue another broker read or turn missing fields into zero values.
+    """
+    if not _flag("FIRSTRADE_PERSIST_ACCOUNT_SNAPSHOT", "false", env_reader):
+        return
+    store = build_gcs_state_store_from_env(env_reader)
+    if store is None:
+        raise ValueError("funds_archive_unavailable")
+    observed = datetime.fromisoformat(observation["observed_started_at"].replace("Z", "+00:00"))
+    if observed.tzinfo is None:
+        raise ValueError("funds_archive_unavailable")
+    observed = observed.astimezone(timezone.utc)
+    balances = observation["balances"]
+    snapshot = {
+        "as_of": observed.isoformat(),
+        "observed_finished_at": observation["observed_finished_at"],
+        "account": observation["broker_account_id"],
+        "session_reused": True,
+        "source": "cached_account_facts",
+        "balance_metrics": {
+            "total_equity": balances.get("provider_equity"),
+            "cash_balance": balances.get("cash_balance"),
+            "available_cash": balances.get("available_cash"),
+            "buying_power": balances.get("buying_power"),
+        },
+        "coverage": dict(observation.get("coverage") or {}),
+    }
+    if observation.get("positions_status") == "available":
+        snapshot["positions"] = observation["positions"]
+    persist_funds_snapshot(store=store, snapshot=snapshot, now=observed)
+
+
 def run_session_check(
     *,
     credentials: FirstradeCredentials | None = None,

docs/account-archive-history.md

@@ -1,5 +1,23 @@
 # Saved account history
 
+## Daily cached snapshot archive
+
+When `FIRSTRADE_PERSIST_ACCOUNT_SNAPSHOT=true`, `/account-facts-sync` writes
+its validated cached-only observation to the existing funds latest/history
+paths before publishing to the console. It uses the original UTC observation
+start time for both the snapshot and history path, preserving decimal text,
+null balances, requested position coverage and the observation finish time.
+No extra broker read, session renewal, order or notification is performed.
+
+A missing store or failed archive write returns HTTP 503 with the fixed
+`funds_archive_unavailable` reason and skips console publishing. A later console
+failure leaves the real archive available for inspection. The existing two-object
+write is not atomic; inspect partial or unknown write outcomes before retrying.
+Disabled archive configurations retain their existing console-only behavior.
+These new observations do not reconstruct missed historical days. Backfill needs
+actual saved broker observations with verified account identity and original
+timestamps; transaction history alone cannot establish past daily valuations.
+
 ## Manual session renewal
 
 Manual session renewal selects the configured account, or the sole returned

main.py

@@ -66,7 +66,9 @@
     validate_reconciliation_candidate,
     validate_reconciliation_preconditions,
 )
-from application.session_check_service import run_forced_session_renewal, run_session_check
+from application.session_check_service import (
+    archive_cached_account_facts, run_forced_session_renewal, run_session_check,
+)
 from notifications.telegram import build_sender
 from quant_platform_kit.common.runtime_reports import (
     append_runtime_report_error,
@@ -124,7 +126,7 @@ def get_project_id() -> str | None:
     "account_identity_mismatch", "account_list_empty", "balances_unavailable",
     "positions_unavailable", "readonly_client_unavailable", "sync_unavailable",
     "sync_rejected", "sync_ack_invalid", "configuration_incomplete",
-    "account_facts_unavailable", "runtime_target_mismatch",
+    "account_facts_unavailable", "runtime_target_mismatch", "funds_archive_unavailable",
 })
 _ACCOUNT_FACTS_LOG_STAGES = frozenset({"cache", "userinfo", "account_list", "account_read", "unclassified"})
 
@@ -1135,6 +1137,12 @@ def account_facts_sync():
             except Exception:
                 pass
 
+    try:
+        archive_cached_account_facts(observation)
+    except Exception:
+        _log_account_facts_failure("funds_archive_unavailable")
+        return jsonify({"ok": False, "error": "funds_archive_unavailable"}), 503
+
     try:
         publish_firstrade_account_snapshot(payload, config)
     except AccountFactsPublishError as exc:

tests/test_cached_funds_archive.py

@@ -0,0 +1,111 @@
+"""Cached-only observations must also reach the configured funds archive."""
+from copy import deepcopy
+
+import pytest
+
+import main
+from application.account_facts_publisher import AccountFactsPublishError
+from tests.test_account_facts_endpoint import FakeClient, _configure
+
+
+@pytest.fixture
+def archive_route(monkeypatch):
+    _configure(monkeypatch, FIRSTRADE_PERSIST_ACCOUNT_SNAPSHOT="true")
+    client = FakeClient()
+    observation = {
+        "status": "available", "platform": "firstrade",
+        "account_selector_status": "matched", "broker_account_id": "synthetic-native-id",
+        "observed_started_at": "2026-10-08T23:59:59+00:00",
+        "observed_finished_at": "2026-10-09T00:00:03+00:00",
+        "balances": {"currency": "USD", "provider_equity": "100.123456",
+                     "cash_balance": None, "available_cash": "25.01"},
+        "positions_status": "available", "positions": [],
+        "coverage": {"positions_requested": True, "positions_count": 0},
+    }
+    writes, publishes = [], []
+
+    class Store:
+        def write_json(self, key, value):
+            writes.append((key, deepcopy(value)))
+            return True
+
+    monkeypatch.setattr(main, "READ_ONLY_ACCOUNT_FACTS_CLIENT_BUILDER", lambda: client)
+    monkeypatch.setattr(main, "collect_firstrade_account_facts", lambda *a, **k: observation)
+    monkeypatch.setattr(main, "publish_firstrade_account_snapshot", lambda *a: publishes.append(a))
+    # Patch the existing state-store boundary; no cloud or broker traffic.
+    import application.session_check_service as service
+    monkeypatch.setattr(service, "build_gcs_state_store_from_env", lambda *a: Store())
+    return observation, writes, publishes, client, service
+
+
+def invoke():
+    return main.app.test_client().post(
+        "/account-facts-sync", headers={"Authorization": "Bearer synthetic-facts-token"})
+
+
+def test_cached_sync_archives_original_observation_before_publish(archive_route):
+    observation, writes, publishes, client, _ = archive_route
+    response = invoke()
+    assert response.status_code == 200
+    assert len(writes) == 2
+    key, saved = writes[1]
+    assert key.endswith("/funds/history/2026/10/08/20261008T235959Z.json")
+    assert saved["as_of"] == observation["observed_started_at"]
+    assert saved["balance_metrics"]["total_equity"] == "100.123456"
+    assert saved["balance_metrics"]["cash_balance"] is None
+    assert saved["positions"] == []
+    assert saved["coverage"] == observation["coverage"]
+    assert len(publishes) == 1
+    assert client.closed
+    assert response.json == {"ok": True, "stored": True}
+    assert "synthetic-native-id" not in response.get_data(as_text=True)
+
+
+@pytest.mark.parametrize("failure", ["missing", "write"])
+def test_archive_failure_blocks_success_and_qrs_publish(archive_route, monkeypatch, failure, capsys):
+    _, writes, publishes, client, service = archive_route
+    class FailedStore:
+        def write_json(self, *a):
+            raise RuntimeError("private-provider-text")
+    monkeypatch.setattr(service, "build_gcs_state_store_from_env",
+                        lambda *a: None if failure == "missing" else FailedStore())
+    response = invoke()
+    assert response.status_code == 503
+    assert response.json == {"ok": False, "error": "funds_archive_unavailable"}
+    assert publishes == []
+    assert client.closed
+    assert "private-provider-text" not in capsys.readouterr().out
+    assert "private-provider-text" not in response.get_data(as_text=True)
+
+
+def test_qrs_failure_keeps_real_archive(archive_route, monkeypatch):
+    _, writes, _, _, _ = archive_route
+    def reject(*a):
+        raise AccountFactsPublishError("sync_unavailable")
+    monkeypatch.setattr(main, "publish_firstrade_account_snapshot", reject)
+    assert invoke().status_code == 502
+    assert len(writes) == 2
+
+
+def test_invalid_identity_writes_nothing(archive_route):
+    observation, writes, publishes, _, _ = archive_route
+    observation["broker_account_id"] = "different-synthetic-id"
+    assert invoke().status_code == 503
+    assert writes == publishes == []
+
+
+def test_disabled_archive_has_no_storage_access(archive_route, monkeypatch):
+    _, writes, publishes, _, service = archive_route
+    monkeypatch.setenv("FIRSTRADE_PERSIST_ACCOUNT_SNAPSHOT", "false")
+    monkeypatch.setattr(service, "build_gcs_state_store_from_env",
+                        lambda *a: pytest.fail("archive disabled"))
+    assert invoke().status_code == 200
+    assert writes == []
+    assert len(publishes) == 1
+
+
+def test_unrequested_positions_stay_absent(archive_route):
+    observation, writes, _, _, _ = archive_route
+    observation.update(positions_status="not_requested", positions=None)
+    assert invoke().status_code == 200
+    assert "positions" not in writes[1][1]