fix: preflight account snapshot delivery and constrain PAPER staging

.github/workflows/sync-cloud-run-env.yml

@@ -40,6 +40,11 @@ on:
         required: false
         type: string
         default: ""
+      account_snapshot_enabled:
+        description: "Optional PAPER account-snapshot endpoint switch. Blank preserves the current value."
+        required: false
+        type: string
+        default: ""
       target:
         description: "Legacy: configured or isolated verify targets. Image-only: exactly PAPER, HK, or SG."
         required: true
@@ -165,6 +170,7 @@ jobs:
       ACCOUNT_HISTORY_GCS_PREFIX: ${{ inputs.account_history_gcs_prefix }}
       ACCOUNT_HISTORY_TARGET_ID: ${{ inputs.account_history_target_id }}
       ACCOUNT_HISTORY_EXPECTED_SCOPE: ${{ inputs.account_history_expected_scope }}
+      ACCOUNT_SNAPSHOT_ENABLED_INPUT: ${{ inputs.account_snapshot_enabled }}
       CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
       CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }}
       GCP_ARTIFACT_REGISTRY_HOSTNAME: ${{ vars.GCP_ARTIFACT_REGISTRY_HOSTNAME }}
@@ -222,6 +228,21 @@ jobs:
             echo "History settings are invalid." >&2
             exit 1
           fi
+          snapshot_setting="${ACCOUNT_SNAPSHOT_ENABLED_INPUT:-}"
+          if [ -n "${snapshot_setting}" ] \
+            && [ "${snapshot_setting}" != "true" ] \
+            && [ "${snapshot_setting}" != "false" ]; then
+            echo "Account snapshot setting is invalid." >&2
+            exit 1
+          fi
+          if [ -n "${snapshot_setting}" ] && [ "${WORKFLOW_TARGET}" != "PAPER" ]; then
+            echo "Account snapshot setting is only admitted for PAPER." >&2
+            exit 1
+          fi
+          if [ -n "${snapshot_setting}" ] && [ "${history_count}" -ne 0 ]; then
+            echo "Account snapshot and history updates must be staged separately." >&2
+            exit 1
+          fi
           if [ "${SOURCE_COMMIT}" = "${GITHUB_SHA}" ] && [ "${history_count}" -eq 0 ]; then
             image_mode=same
           elif [ "${WORKFLOW_TARGET}" = "PAPER" ] && [ "${history_count}" -eq 4 ] \
@@ -251,13 +272,13 @@ jobs:
           fi
 
       - name: Set up Python for image-only admission
-        if: inputs.target == 'PAPER' && inputs.account_history_gcs_prefix != ''
+        if: inputs.target == 'PAPER'
         uses: actions/setup-python@v6
         with:
           python-version: "3.12"
 
       - name: Install frozen image-only admission dependencies
-        if: inputs.target == 'PAPER' && inputs.account_history_gcs_prefix != ''
+        if: inputs.target == 'PAPER'
         run: |
           set -euo pipefail
           python -m pip install --upgrade pip uv
@@ -314,6 +335,8 @@ jobs:
             exit 1
           fi
           history_arg=""
+          snapshot_arg=""
+          plan=""
           if [ "${archive_ref}" = "${approved_candidate}" ]; then
             git fetch --depth 1 origin "${approved_candidate}"
             if [ "$(git cat-file -t "${approved_candidate}")" != "commit" ] \
@@ -322,7 +345,7 @@ jobs:
               exit 1
             fi
           fi
-          if [ "${image_mode}" = "history" ]; then
+          if [ "${WORKFLOW_TARGET}" = "PAPER" ]; then
             plan="$(mktemp)"
             uv run --no-sync python "${GITHUB_WORKSPACE}/scripts/verify_deployed_runtime_target_admission.py" \
               --project="${GCP_PROJECT_ID}" \
@@ -331,8 +354,9 @@ jobs:
               --image-only-staging \
               --plan-out="${plan}"
             history_arg="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1])).get("history_arg") or "")' "${plan}")"
-            if [ -z "${history_arg}" ]; then
-              echo "History settings are required for this image." >&2
+            snapshot_arg="$(python3 -c 'import json,sys; print(json.load(open(sys.argv[1])).get("snapshot_arg") or "")' "${plan}")"
+            if [ -n "${history_arg}" ] && [ -n "${snapshot_arg}" ]; then
+              echo "Image-only environment update plan is invalid." >&2
               exit 1
             fi
           fi
@@ -356,10 +380,11 @@ jobs:
             --image="${image_repo}@${digest}" --no-traffic
             --update-labels="commit-sha=${image_tag},github-run-id=${GITHUB_RUN_ID}" --quiet
           )
-          if [ -n "${history_arg}" ]; then
-            update_command+=(--update-env-vars="${history_arg}")
+          env_update_arg="${history_arg:-${snapshot_arg}}"
+          if [ -n "${env_update_arg}" ]; then
+            update_command+=(--update-env-vars="${env_update_arg}")
           fi
-          if [ "${image_mode}" = "history" ]; then
+          if [ "${WORKFLOW_TARGET}" = "PAPER" ]; then
             revision_suffix="r${GITHUB_RUN_ID}"
             revision_name="${CLOUD_RUN_SERVICE}-${revision_suffix}"
             if [[ ! "${GITHUB_RUN_ID}" =~ ^[0-9]+$ ]] \
@@ -370,7 +395,7 @@ jobs:
             update_command+=(--revision-suffix="${revision_suffix}")
           fi
           "${update_command[@]}"
-          if [ "${image_mode}" = "history" ]; then
+          if [ "${WORKFLOW_TARGET}" = "PAPER" ]; then
             uv run --no-sync python "${GITHUB_WORKSPACE}/scripts/verify_deployed_runtime_target_admission.py" \
               --project="${GCP_PROJECT_ID}" \
               --region="${CLOUD_RUN_REGION}" \

docs/account_snapshot_history.md

@@ -13,7 +13,7 @@
 
 `PAPER` 只表示这份清单配置的范围，不能据此推断券商账户身份。脚本再要求期望 scope 精确为 `PAPER`，服务根必须是没有 userinfo、query、fragment 和额外 path 的 HTTPS `*.run.app`，并且只 GET 该源站的 `/account-snapshot`。GCS 前缀最后一段必须是 `account_snapshots`，不能落在 execution report 路径上。缺任何一项就失败，不补默认值。
 
-可选 QRS 发布仍默认关闭；只有 `ACCOUNT_FACTS_SYNC_ENABLED` 精确为 `true` 时才使用 `ACCOUNT_FACTS_SYNC_URL` 与专用 `ACCOUNT_FACTS_SYNC_TOKEN`。URL 必须是 HTTPS 的精确 `/api/account-facts/sync`，不接受 userinfo、端口、query、fragment 或其他路径；POST 不跟随重定向，设置超时并限制响应体大小。token 只作为该 workflow step 的环境变量和 Authorization header 使用，不打印。
+可选 QRS 发布仍默认关闭；只有 `ACCOUNT_FACTS_SYNC_ENABLED` 精确为 `true` 时才使用 `ACCOUNT_FACTS_SYNC_URL` 与专用 `ACCOUNT_FACTS_SYNC_TOKEN`。启用时会先校验 URL 和非空、无首尾空白的 token；配置错误在读取快照或创建日对象前失败，不消耗该日的 create-only 名额。URL 必须是 HTTPS 的精确 `/api/account-facts/sync`，不接受 userinfo、端口、query、fragment 或其他路径；POST 不跟随重定向，设置超时并限制响应体大小。token 只作为该 workflow step 的环境变量和 Authorization header 使用，不打印。
 
 OIDC 使用 heartbeat 里已经配置的 gcloud：`gcloud auth print-identity-token --audiences=<服务根> --quiet`。stdout 只留在内存，不打印，也不放进参数；失败只报短类别。HTTP 有超时、不跟随重定向、不重试。观察时钟在响应收齐之后读取。
 
@@ -25,7 +25,7 @@ OIDC 使用 heartbeat 里已经配置的 gcloud：`gcloud auth print-identity-to
 
 路径是 `{prefix}/{target_id}/{source_binding_id}/{YYYY-MM-DD}.json`。目标与来源绑定分成不同路径段，不把两段来源拼进同一个对象。`create_text` 只创建：已有对象时结果是 `already_recorded`，不覆盖、不重新拉取。存储结果不明则停止，不写占位点，也不重试。错误输出只有短类别。
 
-仅当本次 `create_text` 明确返回新建成功时，脚本才把完全相同的历史 JSON body POST 到 QRS。`already_recorded` 明确跳过发布，不能把这次新读取的内容冒充为已保存对象；`store_unknown` 不 POST。QRS 发布状态与历史记录状态分开输出：发布拒绝或结果未知不会撤销已写入历史；未知 POST 不自动重试。QRS `ok=true` 且回读的目标、观察日、观察结束时间匹配，只表示接收端确认保存，不证明页面已经展示或数据完成物理账户身份核验。接收端按其可信配置绑定目标与来源，调用方不传账户 key 或身份结论。
+仅当本次 `create_text` 明确返回新建成功时，脚本才把完全相同的历史 JSON body POST 到 QRS。`already_recorded` 明确跳过发布，不能把这次新读取的内容冒充为已保存对象；`store_unknown` 不 POST。QRS 发布状态与历史记录状态分开输出：发布拒绝或结果未知不会撤销已写入历史；未知 POST 不自动重试。现有流程没有已存对象的补送入口，且接收端默认拒绝超过 15 分钟观察窗口的记录，因此不能靠下一次日常运行可靠补送；配置预检不解决请求结果未知的情况。QRS `ok=true` 且回读的目标、观察日、观察结束时间匹配，只表示接收端确认保存，不证明页面已经展示或数据完成物理账户身份核验。接收端按其可信配置绑定目标与来源，调用方不传账户 key 或身份结论。
 
 这份记录不是 TWR，不是收益率，也不授予 live 权限。
 

scripts/record_daily_account_snapshot.py

@@ -78,6 +78,12 @@ def record_daily_account_snapshot(
 
     if str(env.get("ACCOUNT_HISTORY_RECORDING_ENABLED") or "").strip() != "true":
         return DailyAccountRecordResult("disabled", publish_status="disabled")
+    try:
+        sync_config = _account_facts_sync_config(env)
+    except _Rejected as rejected:
+        return DailyAccountRecordResult(
+            "error", rejected.category, "rejected", rejected.category
+        )
     try:
         config = _config(env)
     except _Rejected as rejected:
@@ -111,7 +117,7 @@ def record_daily_account_snapshot(
         )
     if created is True:
         publish_status, publish_category = _publish_account_facts(
-            env, body, http_post=http_post or _http_post
+            sync_config, body, http_post=http_post or _http_post
         )
         return DailyAccountRecordResult(
             "recorded", "", publish_status, publish_category
@@ -126,20 +132,14 @@ def record_daily_account_snapshot(
 
 
 def _publish_account_facts(
-    env: Mapping[str, str],
+    sync_config: tuple[str, str] | None,
     body: str,
     *,
     http_post: Callable[..., Any],
 ) -> tuple[str, str]:
-    if str(env.get("ACCOUNT_FACTS_SYNC_ENABLED") or "").strip() != "true":
+    if sync_config is None:
         return "disabled", ""
-    try:
-        url = _account_facts_sync_url(str(env.get("ACCOUNT_FACTS_SYNC_URL") or ""))
-        token = str(env.get("ACCOUNT_FACTS_SYNC_TOKEN") or "")
-        if not token or token != token.strip():
-            raise _Rejected("qrs_config_invalid")
-    except _Rejected as rejected:
-        return "rejected", rejected.category
+    url, token = sync_config
 
     encoded_body = body.encode("utf-8")
     if len(encoded_body) > MAX_ACCOUNT_FACTS_SYNC_BODY_BYTES:
@@ -223,6 +223,16 @@ def _account_facts_sync_url(value: str) -> str:
     return f"https://{host}{ACCOUNT_FACTS_SYNC_PATH}"
 
 
+def _account_facts_sync_config(env: Mapping[str, str]) -> tuple[str, str] | None:
+    if str(env.get("ACCOUNT_FACTS_SYNC_ENABLED") or "").strip() != "true":
+        return None
+    url = _account_facts_sync_url(str(env.get("ACCOUNT_FACTS_SYNC_URL") or ""))
+    token = str(env.get("ACCOUNT_FACTS_SYNC_TOKEN") or "")
+    if not token or token != token.strip():
+        raise _Rejected("qrs_config_invalid")
+    return url, token
+
+
 def _bounded_response_json(response: Any, max_bytes: int) -> Any:
     chunks: list[bytes] = []
     total = 0

scripts/verify_deployed_runtime_target_admission.py

@@ -41,6 +41,8 @@ class AdmissionError(ValueError):
     "ACCOUNT_HISTORY_TARGET_ID",
     "ACCOUNT_HISTORY_EXPECTED_SCOPE",
 )
+_ACCOUNT_SNAPSHOT_INPUT = "ACCOUNT_SNAPSHOT_ENABLED_INPUT"
+_ACCOUNT_SNAPSHOT_ENV = "LONGBRIDGE_ACCOUNT_SNAPSHOT_ENABLED"
 _UNSAFE_HISTORY = re.compile(r"[,=\s'\"`$\\|&<>]|^\-")
 _INGRESS = "run.googleapis.com/ingress"
 _SOURCE_DECLARATIONS = ("uv.lock", "pyproject.toml", "qsl.toml")
@@ -493,11 +495,11 @@ def prepare_image_only_staging(
     image_commit: str,
     run: Callable[[Sequence[str]], str],
 ) -> dict[str, Any]:
-    """Admit one PAPER no-traffic image. Lock texts come from image_commit, not the main worktree."""
+    """Admit a fixed history candidate or the signed main image for PAPER staging."""
 
-    if image_commit != APPROVED_PAPER_HISTORY_CANDIDATE:
-        raise AdmissionError("image source is not approved")
-    image_commit = APPROVED_PAPER_HISTORY_CANDIDATE
+    history, snapshot_value = _validate_image_only_source(
+        image_commit=image_commit, env=env, project=project
+    )
     admission = verify_service(service=service, service_json=service_json)
     _require_paper_target_identity(service=service, service_json=service_json)
     try:
@@ -539,17 +541,71 @@ def prepare_image_only_staging(
     )
     if serving != {source_revision}:
         raise AdmissionError(f"{service}: candidate UES revision differs from serving image")
-    history = history_update(env, workflow_target=str(env.get("WORKFLOW_TARGET") or ""), project_id=project)
+    history_arg = _history_arg(history)
+    snapshot_arg = (
+        f"{_ACCOUNT_SNAPSHOT_ENV}={snapshot_value}"
+        if snapshot_value is not None
+        else ""
+    )
+    changed_keys = set(history or ())
+    if snapshot_value is not None:
+        changed_keys.add(_ACCOUNT_SNAPSHOT_ENV)
     return {
-        "history_arg": _history_arg(history),
+        "history_arg": history_arg,
         "history_values": history or {},
+        "snapshot_arg": snapshot_arg,
+        "snapshot_value": snapshot_value,
         "image_commit": image_commit,
         "service_ingress": _ingress(service_json.get("metadata")),
         "serving_traffic": traffic,
-        "template_digest": _config_digest(configuration, skip=set(history or ())),
+        "template_digest": _config_digest(configuration, skip=changed_keys),
     }
 
 
+def _validate_image_only_source(
+    *, image_commit: str, env: Mapping[str, str], project: str
+) -> tuple[dict[str, str] | None, str | None]:
+    history = history_update(
+        env, workflow_target=str(env.get("WORKFLOW_TARGET") or ""), project_id=project
+    )
+    snapshot_value = _account_snapshot_update(env, workflow_target=str(env.get("WORKFLOW_TARGET") or ""))
+    if image_commit == APPROVED_PAPER_HISTORY_CANDIDATE:
+        if history is None or snapshot_value is not None:
+            raise AdmissionError("image source is not approved")
+    elif _is_exact_main_image(image_commit, env):
+        if history is not None:
+            raise AdmissionError("image source is not approved")
+    else:
+        raise AdmissionError("image source is not approved")
+    return history, snapshot_value
+
+
+def _account_snapshot_update(
+    env: Mapping[str, str], *, workflow_target: str
+) -> str | None:
+    value = str(env.get(_ACCOUNT_SNAPSHOT_INPUT) or "")
+    if value not in {"", "true", "false"}:
+        raise AdmissionError("account snapshot setting is invalid")
+    if value and workflow_target != "PAPER":
+        raise AdmissionError("account snapshot setting is only admitted for PAPER")
+    return value or None
+
+
+def _is_exact_main_image(image_commit: str, env: Mapping[str, str]) -> bool:
+    workflow_sha = str(env.get("GITHUB_SHA") or "")
+    return (
+        _SHA.fullmatch(image_commit) is not None
+        and image_commit == workflow_sha
+        and image_commit == str(env.get("GITHUB_WORKFLOW_SHA") or "")
+        and str(env.get("GITHUB_REPOSITORY") or "") == "QuantStrategyLab/LongBridgePlatform"
+        and str(env.get("APPROVED_REF") or "") == "main"
+        and str(env.get("GITHUB_REF_NAME") or "") == "main"
+        and str(env.get("GITHUB_REF") or "") == "refs/heads/main"
+        and str(env.get("GITHUB_WORKFLOW_REF") or "")
+        == "QuantStrategyLab/LongBridgePlatform/.github/workflows/sync-cloud-run-env.yml@refs/heads/main"
+    )
+
+
 def confirm_image_only_readback(
     *,
     service: str,
@@ -621,17 +677,21 @@ def confirm_image_only_readback(
     history = plan.get("history_values") or {}
     if not isinstance(history, Mapping):
         raise AdmissionError("staged readback is incomplete")
+    snapshot_value = plan.get("snapshot_value")
+    if snapshot_value not in {None, "true", "false"}:
+        raise AdmissionError("staged readback is incomplete")
+    skipped_keys = set(history)
+    if snapshot_value is not None:
+        skipped_keys.add(_ACCOUNT_SNAPSHOT_ENV)
     configuration = _configuration(revision)
-    if _config_digest(configuration, skip=set(history)) != plan.get("template_digest"):
+    if _config_digest(configuration, skip=skipped_keys) != plan.get("template_digest"):
         raise AdmissionError("staged revision configuration changed")
     literals = _literal_values(configuration)
     for key, value in history.items():
         if literals.get(str(key)) != value:
             raise AdmissionError("staged history settings do not match")
-
-
-def _history_count(env: Mapping[str, str]) -> int:
-    return sum(1 for key in _HISTORY_KEYS if str(env.get(key) or ""))
+    if snapshot_value is not None and literals.get(_ACCOUNT_SNAPSHOT_ENV) != snapshot_value:
+        raise AdmissionError("staged account snapshot setting does not match")
 
 
 def _write_plan(path: str, plan: Mapping[str, Any]) -> None:
@@ -659,13 +719,10 @@ def main(argv: Sequence[str] | None = None) -> int:
             print("Image-only staging admission failed.", file=sys.stderr)
             return 1
         try:
-            if (
-                str(os.environ.get("WORKFLOW_TARGET") or "") != "PAPER"
-                or _history_count(os.environ) != 4
-                or str(os.environ.get("SOURCE_COMMIT") or "") != APPROVED_PAPER_HISTORY_CANDIDATE
-            ):
-                raise AdmissionError("image source is not approved")
-            image_commit = APPROVED_PAPER_HISTORY_CANDIDATE
+            image_commit = str(os.environ.get("SOURCE_COMMIT") or "")
+            _validate_image_only_source(
+                image_commit=image_commit, env=os.environ, project=args.project
+            )
             service_json = _load_object(
                 _run_quiet(
                     [

tests/test_daily_account_snapshot.py

@@ -166,7 +166,14 @@ def _record(env=None, spies=None, now=NOW, now_reader=None):
 
 
 def test_disabled_switch_does_not_touch_token_http_or_store():
-    result, spies = _record(_env(ACCOUNT_HISTORY_RECORDING_ENABLED="false"))
+    result, spies = _record(
+        _env(
+            ACCOUNT_HISTORY_RECORDING_ENABLED="false",
+            ACCOUNT_FACTS_SYNC_ENABLED="true",
+            ACCOUNT_FACTS_SYNC_URL="",
+            ACCOUNT_FACTS_SYNC_TOKEN="",
+        )
+    )
 
     assert result.status == "disabled"
     assert spies.calls == []
@@ -515,9 +522,37 @@ def test_qrs_sync_url_must_be_exact_https_endpoint():
             ),
             spies=spies,
         )
-        assert result.status == "recorded"
+        assert result.status == "error"
+        assert result.category == "qrs_config_invalid"
         assert result.publish_status == "rejected"
-        assert [name for name, *_ in spies.calls].count("post") == 0
+        assert spies.calls == []
+        assert spies.stored == []
+
+
+@pytest.mark.parametrize(
+    "bad_config",
+    [
+        {"ACCOUNT_FACTS_SYNC_URL": ""},
+        {"ACCOUNT_FACTS_SYNC_URL": "https://qrs.example.test/other"},
+        {"ACCOUNT_FACTS_SYNC_TOKEN": ""},
+        {"ACCOUNT_FACTS_SYNC_TOKEN": " synthetic-qrs-token"},
+    ],
+)
+def test_enabled_qrs_invalid_config_is_rejected_before_snapshot_io(bad_config):
+    env = _env(
+        ACCOUNT_FACTS_SYNC_ENABLED="true",
+        ACCOUNT_FACTS_SYNC_URL=QRS_SYNC_URL,
+        ACCOUNT_FACTS_SYNC_TOKEN=QRS_TOKEN,
+    )
+    env.update(bad_config)
+    result, spies = _record(env=env)
+
+    assert result.status == "error"
+    assert result.category == "qrs_config_invalid"
+    assert result.publish_status == "rejected"
+    assert result.publish_category == "qrs_config_invalid"
+    assert spies.calls == []
+    assert spies.stored == []
 
 
 def test_new_source_or_utc_day_uses_a_new_object_segment():