fix: admit scoped SG and HK account snapshot adoption

.github/workflows/execution-report-heartbeat.yml

@@ -122,9 +122,9 @@ jobs:
       - name: Set up gcloud
         uses: google-github-actions/setup-gcloud@v3
 
-      - name: Record and sync daily paper account snapshot
+      - name: Record and sync daily account snapshot
         id: account_history
-        if: ${{ !cancelled() && matrix.target.label == 'PAPER' && vars.ACCOUNT_HISTORY_RECORDING_ENABLED == 'true' }}
+        if: ${{ !cancelled() && contains(fromJSON('["PAPER","HK","SG"]'), matrix.target.label) && vars.ACCOUNT_HISTORY_RECORDING_ENABLED == 'true' }}
         continue-on-error: true
         env:
           ACCOUNT_HISTORY_RECORDING_ENABLED: ${{ vars.ACCOUNT_HISTORY_RECORDING_ENABLED }}

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

@@ -205,6 +205,7 @@ jobs:
           approved_candidate=0b939723c1db3ef59175535998b470cbcd4b8824
           approved_http_snapshot_candidate=d8314a61df697cae1dd03a78ddc5c2fc4179ec67
           approved_probe_snapshot_candidate=a2921d157efb887e9210fad6734ca040ea6e5293
+          approved_sghk_snapshot_candidate=922f338fea7c46391b50bb8316ac88154596916f
           history_count=0
           for history_value in \
             "${ACCOUNT_HISTORY_RECORDING_ENABLED:-}" \
@@ -222,14 +223,29 @@ jobs:
               esac
             fi
           done
-          if [ "${history_count}" -ne 0 ] && [ "${WORKFLOW_TARGET}" != "PAPER" ]; then
-            echo "History settings are only admitted for PAPER." >&2
+          if [ "${history_count}" -ne 0 ] && [ "${WORKFLOW_TARGET}" != "PAPER" ] \
+            && [ "${WORKFLOW_TARGET}" != "HK" ] && [ "${WORKFLOW_TARGET}" != "SG" ]; then
+            echo "History settings are not admitted for this target." >&2
             exit 1
           fi
           if [ "${history_count}" -ne 0 ] && [ "${history_count}" -ne 4 ]; then
             echo "History settings are invalid." >&2
             exit 1
           fi
+          if [ "${history_count}" -eq 4 ] && [ "${WORKFLOW_TARGET}" != "PAPER" ]; then
+            case "${WORKFLOW_TARGET}" in
+              HK) expected_target=hk ;;
+              SG) expected_target=sg ;;
+              *) echo "History settings are not admitted for this target." >&2; exit 1 ;;
+            esac
+            if [ "${ACCOUNT_HISTORY_RECORDING_ENABLED}" != "true" ] \
+              || [ "${ACCOUNT_HISTORY_TARGET_ID}" != "${expected_target}" ] \
+              || [ "${ACCOUNT_HISTORY_EXPECTED_SCOPE}" != "${WORKFLOW_TARGET}" ] \
+              || [ "${ACCOUNT_HISTORY_GCS_PREFIX}" != "gs://qsl-runtime-logs-shared/longbridge/account_snapshots" ]; then
+              echo "History settings do not match the approved account." >&2
+              exit 1
+            fi
+          fi
           snapshot_setting="${ACCOUNT_SNAPSHOT_ENABLED_INPUT:-}"
           if [ -n "${snapshot_setting}" ] \
             && [ "${snapshot_setting}" != "true" ] \
@@ -257,6 +273,10 @@ jobs:
             && [ -z "${snapshot_setting}" ] \
             && [ "${SOURCE_COMMIT}" = "${approved_probe_snapshot_candidate}" ]; then
             image_mode=probe
+          elif { [ "${WORKFLOW_TARGET}" = "HK" ] || [ "${WORKFLOW_TARGET}" = "SG" ]; } \
+            && [ "${history_count}" -eq 4 ] && [ -z "${snapshot_setting}" ] \
+            && [ "${SOURCE_COMMIT}" = "${approved_sghk_snapshot_candidate}" ]; then
+            image_mode=account-history
           else
             echo "Image source is not approved." >&2
             exit 1
@@ -281,13 +301,13 @@ jobs:
           fi
 
       - name: Set up Python for image-only admission
-        if: inputs.target == 'PAPER'
+        if: inputs.target == 'PAPER' || inputs.target == 'HK' || inputs.target == 'SG'
         uses: actions/setup-python@v6
         with:
           python-version: "3.12"
 
       - name: Install frozen image-only admission dependencies
-        if: inputs.target == 'PAPER'
+        if: inputs.target == 'PAPER' || inputs.target == 'HK' || inputs.target == 'SG'
         run: |
           set -euo pipefail
           python -m pip install --upgrade pip uv
@@ -311,6 +331,12 @@ jobs:
           approved_candidate=0b939723c1db3ef59175535998b470cbcd4b8824
           approved_http_snapshot_candidate=d8314a61df697cae1dd03a78ddc5c2fc4179ec67
           approved_probe_snapshot_candidate=a2921d157efb887e9210fad6734ca040ea6e5293
+          approved_sghk_snapshot_candidate=922f338fea7c46391b50bb8316ac88154596916f
+          case "${WORKFLOW_TARGET}" in
+            HK) expected_sghk_target=hk ;;
+            SG) expected_sghk_target=sg ;;
+            *) expected_sghk_target="" ;;
+          esac
           history_count=0
           for history_value in \
             "${ACCOUNT_HISTORY_RECORDING_ENABLED:-}" \
@@ -335,6 +361,14 @@ jobs:
             && [ -z "${snapshot_setting}" ] \
             && [ "${SOURCE_COMMIT}" = "${approved_probe_snapshot_candidate}" ]; then
             image_mode=probe
+          elif { [ "${WORKFLOW_TARGET}" = "HK" ] || [ "${WORKFLOW_TARGET}" = "SG" ]; } \
+            && [ "${history_count}" -eq 4 ] && [ -z "${snapshot_setting}" ] \
+            && [ "${ACCOUNT_HISTORY_RECORDING_ENABLED}" = "true" ] \
+            && [ "${ACCOUNT_HISTORY_TARGET_ID}" = "${expected_sghk_target}" ] \
+            && [ "${ACCOUNT_HISTORY_EXPECTED_SCOPE}" = "${WORKFLOW_TARGET}" ] \
+            && [ "${ACCOUNT_HISTORY_GCS_PREFIX}" = "gs://qsl-runtime-logs-shared/longbridge/account_snapshots" ] \
+            && [ "${SOURCE_COMMIT}" = "${approved_sghk_snapshot_candidate}" ]; then
+            image_mode=account-history
           else
             echo "Image source is not approved." >&2
             exit 1
@@ -348,6 +382,9 @@ jobs:
           elif [ "${image_mode}" = "probe" ]; then
             archive_ref="${approved_probe_snapshot_candidate}"
             image_tag="${approved_probe_snapshot_candidate}"
+          elif [ "${image_mode}" = "account-history" ]; then
+            archive_ref="${approved_sghk_snapshot_candidate}"
+            image_tag="${approved_sghk_snapshot_candidate}"
           else
             archive_ref="${approved_candidate}"
             image_tag="${approved_candidate}"
@@ -370,7 +407,7 @@ jobs:
               exit 1
             fi
           fi
-          if [ "${WORKFLOW_TARGET}" = "PAPER" ]; then
+          if [ "${WORKFLOW_TARGET}" = "PAPER" ] || [ "${image_mode}" = "account-history" ]; then
             plan="$(mktemp)"
             uv run --no-sync python "${GITHUB_WORKSPACE}/scripts/verify_deployed_runtime_target_admission.py" \
               --project="${GCP_PROJECT_ID}" \
@@ -409,7 +446,7 @@ jobs:
           if [ -n "${env_update_arg}" ]; then
             update_command+=(--update-env-vars="${env_update_arg}")
           fi
-          if [ "${WORKFLOW_TARGET}" = "PAPER" ]; then
+          if [ "${WORKFLOW_TARGET}" = "PAPER" ] || [ "${image_mode}" = "account-history" ]; then
             revision_suffix="r${GITHUB_RUN_ID}"
             revision_name="${CLOUD_RUN_SERVICE}-${revision_suffix}"
             if [[ ! "${GITHUB_RUN_ID}" =~ ^[0-9]+$ ]] \
@@ -420,7 +457,7 @@ jobs:
             update_command+=(--revision-suffix="${revision_suffix}")
           fi
           "${update_command[@]}"
-          if [ "${WORKFLOW_TARGET}" = "PAPER" ]; then
+          if [ "${WORKFLOW_TARGET}" = "PAPER" ] || [ "${image_mode}" = "account-history" ]; then
             uv run --no-sync python "${GITHUB_WORKSPACE}/scripts/verify_deployed_runtime_target_admission.py" \
               --project="${GCP_PROJECT_ID}" \
               --region="${CLOUD_RUN_REGION}" \

scripts/record_daily_account_snapshot.py

@@ -20,7 +20,8 @@
 SNAPSHOT_SCHEMA = "longbridge_account_snapshot.v1"
 SOURCE_KIND = "deployment_scope_token_version"
 ACCOUNT_FACTS_SYNC_PATH = "/api/account-facts/sync"
-EXPECTED_SCOPE = "PAPER"
+_EXPECTED_TARGETS = {"paper": ("PAPER", "paper"), "hk": ("HK", "live"), "sg": ("SG", "live")}
+_EXPECTED_GCS_PREFIX = "gs://qsl-runtime-logs-shared/longbridge/account_snapshots"
 SCHEDULER_SERVICE_ACCOUNT = "longbridge-platform-scheduler@longbridgequant.iam.gserviceaccount.com"
 OBSERVATION_WINDOW = timedelta(minutes=15)
 WAIT_SECONDS = 180.0
@@ -68,6 +69,7 @@ class _Config:
     bucket: str
     prefix_path: str
     target_id: str
+    expected_scope: str
     source_binding_id: str
     scheduler_job: str
     scheduler_resource: str
@@ -154,21 +156,28 @@ def _config(env: Mapping[str, str]) -> _Config:
     target_id = str(env.get("ACCOUNT_HISTORY_TARGET_ID") or "").strip()
     expected_scope = str(env.get("ACCOUNT_HISTORY_EXPECTED_SCOPE") or "").strip()
     source_binding_id = str(env.get("ACCOUNT_HISTORY_EXPECTED_SOURCE_BINDING_ID") or "").strip()
+    target_contract = _EXPECTED_TARGETS.get(target_id)
     if (
         _PROJECT_ID.fullmatch(project_id) is None
-        or target_id != "paper"
+        or project_id != "longbridgequant"
+        or target_contract is None
         or _TARGET_ID.fullmatch(target_id) is None
-        or expected_scope != EXPECTED_SCOPE
+        or expected_scope != target_contract[0]
+        or prefix != _EXPECTED_GCS_PREFIX
         or _BINDING_ID.fullmatch(source_binding_id) is None
     ):
         raise _Rejected("config_invalid")
     try:
         from application.runtime_target_manifest import load_runtime_target_manifest
 
-        matches = [target for target in load_runtime_target_manifest().targets if target.id == "paper"]
+        matches = [target for target in load_runtime_target_manifest().targets if target.id == target_id]
     except Exception:
         raise _Rejected("config_invalid") from None
-    if len(matches) != 1 or matches[0].mode != "paper":
+    if (
+        len(matches) != 1
+        or matches[0].mode != target_contract[1]
+        or matches[0].account_scope != expected_scope
+    ):
         raise _Rejected("config_invalid")
     service = matches[0].service
     region = matches[0].region
@@ -186,6 +195,7 @@ def _config(env: Mapping[str, str]) -> _Config:
         bucket=bucket,
         prefix_path=prefix_path,
         target_id=target_id,
+        expected_scope=expected_scope,
         source_binding_id=source_binding_id,
         scheduler_job=scheduler_job,
         scheduler_resource=scheduler_resource,
@@ -288,7 +298,7 @@ def _validate_scheduler_job(job: Mapping[str, Any], config: _Config) -> None:
         body_empty = body is None
     if (
         job.get("name") != config.scheduler_resource
-        or job.get("state") != "ENABLED"
+        or job.get("state") not in ({"ENABLED", "PAUSED"} if config.target_id == "hk" else {"ENABLED"})
         or target.get("httpMethod") != "POST"
         or target.get("uri") != f"{config.service_url}/probe"
         or not body_empty
@@ -394,7 +404,7 @@ def _list_candidates(
         remaining = deadline - monotonic()
         if remaining <= 0:
             raise _Rejected("observation_timeout")
-        prefix = f"{config.prefix_path}/paper/{config.source_binding_id}/{day.isoformat()}/"
+        prefix = f"{config.prefix_path}/{config.target_id}/{config.source_binding_id}/{day.isoformat()}/"
         try:
             blobs = client.list_blobs(
                 config.bucket,
@@ -497,7 +507,7 @@ def _validate_history_object(
         set(payload) != expected_fields
         or payload.get("schema_version") != HISTORY_SCHEMA
         or payload.get("snapshot_schema_version") != SNAPSHOT_SCHEMA
-        or payload.get("account_scope") != EXPECTED_SCOPE
+        or payload.get("account_scope") != config.expected_scope
         or payload.get("target_id") != config.target_id
         or payload.get("snapshot_atomic") is not False
         or not isinstance(binding, Mapping)
@@ -509,7 +519,7 @@ def _validate_history_object(
     started = _aware(payload.get("observed_started_at"))
     finished = _aware(payload.get("observed_finished_at"))
     expected_name = (
-        f"{config.prefix_path}/paper/{config.source_binding_id}/"
+        f"{config.prefix_path}/{config.target_id}/{config.source_binding_id}/"
         f"{started.date().isoformat()}/{_filename_for(finished)}"
     )
     if (

scripts/verify_deployed_runtime_target_admission.py

@@ -39,6 +39,8 @@ class AdmissionError(ValueError):
 APPROVED_PAPER_HTTP_SNAPSHOT_CANDIDATE = "d8314a61df697cae1dd03a78ddc5c2fc4179ec67"
 # Reviewed PAPER internal-probe snapshot producer. It carries no history or snapshot env update.
 APPROVED_PAPER_PROBE_SNAPSHOT_CANDIDATE = "a2921d157efb887e9210fad6734ca040ea6e5293"
+# Reviewed read-only SG/HK account-snapshot producer. It is not a PAPER candidate.
+APPROVED_SGHK_ACCOUNT_SNAPSHOT_CANDIDATE = "922f338fea7c46391b50bb8316ac88154596916f"
 # One already-staged PAPER revision that may hold history env while serving still runs an older image.
 _PAPER_HTTP_STAGED_SOURCE_REVISION = "longbridge-quant-paper-service-r36423178119"
 _PAPER_HTTP_STAGED_SOURCE_COMMIT = APPROVED_PAPER_HISTORY_CANDIDATE
@@ -398,20 +400,26 @@ def _literal_values(configuration: Mapping[str, Any]) -> dict[str, str]:
     return {item["name"]: item["value"] for item in configuration["env"] if "value" in item}
 
 
-def history_update(env: Mapping[str, str], *, workflow_target: str, project_id: str) -> dict[str, str] | None:
-    """Return an explicit PAPER history update, or None when the four inputs were omitted."""
+def history_update(
+    env: Mapping[str, str], *, workflow_target: str, project_id: str, allow_sghk: bool = False
+) -> dict[str, str] | None:
+    """Return one exact target history update, or None when the four inputs were omitted."""
 
     values = {key: str(env.get(key) or "") for key in _HISTORY_KEYS}
     if all(value == "" for value in values.values()):
         return None
-    if workflow_target != "PAPER":
-        raise AdmissionError("history settings are only admitted for PAPER")
     if any(not value or _UNSAFE_HISTORY.search(value) for value in values.values()):
         raise AdmissionError("history settings are invalid")
     if values["ACCOUNT_HISTORY_RECORDING_ENABLED"] != "true":
         raise AdmissionError("history settings are invalid")
-    if values["ACCOUNT_HISTORY_EXPECTED_SCOPE"] != "PAPER" or values["ACCOUNT_HISTORY_TARGET_ID"] != "paper":
-        raise AdmissionError("history settings are only admitted for PAPER")
+    target_pairs = {"PAPER": ("paper", "PAPER")}
+    if allow_sghk:
+        target_pairs.update({"HK": ("hk", "HK"), "SG": ("sg", "SG")})
+    expected = target_pairs.get(workflow_target)
+    if expected is None:
+        raise AdmissionError("history settings are only admitted for approved targets")
+    if (values["ACCOUNT_HISTORY_TARGET_ID"], values["ACCOUNT_HISTORY_EXPECTED_SCOPE"]) != expected:
+        raise AdmissionError("history settings do not match the selected target")
     if _PROJECT_ID.fullmatch(project_id) is None or _TARGET_ID.fullmatch(values["ACCOUNT_HISTORY_TARGET_ID"]) is None:
         raise AdmissionError("history settings are invalid")
     try:
@@ -420,9 +428,51 @@ def history_update(env: Mapping[str, str], *, workflow_target: str, project_id:
         raise AdmissionError("history settings are invalid") from None
     if prefix != values["ACCOUNT_HISTORY_GCS_PREFIX"]:
         raise AdmissionError("history settings are invalid")
+    if allow_sghk and prefix != "gs://qsl-runtime-logs-shared/longbridge/account_snapshots":
+        raise AdmissionError("history settings are not admitted for this candidate")
     return values
 
 
+def _require_sghk_candidate_identity(
+    *, target_label: str, service: str, project: str, region: str, service_json: Mapping[str, Any]
+) -> None:
+    expected = {"HK": ("hk", "HK"), "SG": ("sg", "SG")}.get(target_label)
+    if expected is None or project != "longbridgequant":
+        raise AdmissionError("candidate target does not match the approved account")
+    target_id, scope = expected
+    try:
+        from application.runtime_target_manifest import load_runtime_target_manifest
+
+        matches = [item for item in load_runtime_target_manifest().targets if item.id == target_id]
+    except Exception:
+        raise AdmissionError("candidate target does not match the approved account") from None
+    if (
+        len(matches) != 1
+        or matches[0].mode != "live"
+        or matches[0].account_scope != scope
+        or matches[0].service != service
+        or matches[0].region != region
+    ):
+        raise AdmissionError("candidate target does not match the approved account")
+    env = _container_env(service_json)
+    try:
+        runtime_target = json.loads(env.get("RUNTIME_TARGET_JSON") or env.get("QSL_RUNTIME_TARGET_JSON") or "{}")
+    except json.JSONDecodeError:
+        raise AdmissionError("runtime target identity does not match the approved account") from None
+    selector = runtime_target.get("account_selector") if isinstance(runtime_target, Mapping) else None
+    selectors = (selector,) if isinstance(selector, str) else tuple(selector) if isinstance(selector, list) else None
+    deployment_selector = str(runtime_target.get("deployment_selector") or "") if isinstance(runtime_target, Mapping) else ""
+    if (
+        not isinstance(runtime_target, Mapping)
+        or runtime_target.get("platform_id") != "longbridge"
+        or runtime_target.get("service_name") != service
+        or runtime_target.get("account_scope") != scope
+        or selectors != (scope,)
+        or deployment_selector != ("HK" if target_id == "hk" else "SG")
+    ):
+        raise AdmissionError("runtime target identity does not match the approved account")
+
+
 def _history_arg(values: Mapping[str, str] | None) -> str:
     if not values:
         return ""
@@ -668,7 +718,16 @@ def prepare_image_only_staging(
         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)
+    if image_commit == APPROVED_SGHK_ACCOUNT_SNAPSHOT_CANDIDATE:
+        _require_sghk_candidate_identity(
+            target_label=str(env.get("WORKFLOW_TARGET") or ""),
+            service=service,
+            project=project,
+            region=region,
+            service_json=service_json,
+        )
+    else:
+        _require_paper_target_identity(service=service, service_json=service_json)
     try:
         traffic = serving_traffic_rows(service_json)
     except ReconcileError:
@@ -746,7 +805,10 @@ 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
+        env,
+        workflow_target=str(env.get("WORKFLOW_TARGET") or ""),
+        project_id=project,
+        allow_sghk=image_commit == APPROVED_SGHK_ACCOUNT_SNAPSHOT_CANDIDATE,
     )
     snapshot_value = _account_snapshot_update(env, workflow_target=str(env.get("WORKFLOW_TARGET") or ""))
     if image_commit == APPROVED_PAPER_HISTORY_CANDIDATE:
@@ -762,6 +824,13 @@ def _validate_image_only_source(
             or snapshot_value is not None
         ):
             raise AdmissionError("image source is not approved")
+    elif image_commit == APPROVED_SGHK_ACCOUNT_SNAPSHOT_CANDIDATE:
+        if (
+            str(env.get("WORKFLOW_TARGET") or "") not in {"HK", "SG"}
+            or 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")