fix(observation): enable the SG snapshot on its serving image

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

@@ -13,6 +13,7 @@ on:
           - legacy
           - image-only-no-traffic
           - image-only-notification-overlay
+          - sg-serving-snapshot
       approved_ref:
         description: "Exact workflow branch/tag name approved for image-only staging."
         required: false
@@ -2307,3 +2308,38 @@ jobs:
               --location="${scheduler_location}" \
               --quiet
           fi
+
+
+  sg-serving-snapshot:
+    name: Enable only the SG read-only snapshot
+    if: >-
+      github.event_name == 'workflow_dispatch' && github.ref == 'refs/heads/main' &&
+      github.run_attempt == 1 && inputs.deployment_mode == 'sg-serving-snapshot' &&
+      inputs.target == 'SG' && inputs.approved_ref == 'main' &&
+      inputs.account_snapshot_enabled == 'true' &&
+      inputs.source_commit == '' &&
+      inputs.account_history_recording_enabled == '' && inputs.account_history_gcs_prefix == '' &&
+      inputs.account_history_target_id == '' && inputs.account_history_expected_scope == ''
+    runs-on: ubuntu-latest
+    timeout-minutes: 12
+    environment: longbridge-sg
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: ${{ vars.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+      CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }}
+      CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }}
+    steps:
+      - uses: actions/checkout@v6
+        with:
+          persist-credentials: false
+      - uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }}
+          service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }}
+      - uses: google-github-actions/setup-gcloud@v3
+      - name: Stage and verify the serving image with the snapshot flag
+        env:
+          SG_SNAPSHOT_APPROVED_SERVING_SOURCE: ${{ secrets.SG_SNAPSHOT_APPROVED_SERVING_SOURCE }}
+        run: python3 scripts/enable_sg_serving_snapshot.py

config/hk_snapshot_verification.workflow.yaml

@@ -6,6 +6,7 @@ main:
         assign:
           - origin: ${sys.get_env("HK_SNAPSHOT_ORIGIN")}
           - expected_binding: ${sys.get_env("HK_SNAPSHOT_EXPECTED_BINDING")}
+          - expected_scope: ${sys.get_env("SNAPSHOT_VERIFICATION_SCOPE", "HK")}
           - requested_at: ${sys.now()}
           - archive_bucket: ${sys.get_env("HK_DIGEST_BUCKET", "")}
           - archive_prefix: ${sys.get_env("HK_DIGEST_PREFIX", "")}
@@ -14,7 +15,7 @@ main:
           - strategy_profile: ${sys.get_env("HK_DIGEST_STRATEGY_PROFILE", "")}
     - check_configuration:
         switch:
-          - condition: '${not text.match_regex(origin, "^https://[a-z0-9-]+[.][a-z0-9-]+[.]run[.]app$") or not text.match_regex(expected_binding, "^[a-f0-9]{64}$")}'
+          - condition: '${not text.match_regex(origin, "^https://[a-z0-9-]+[.][a-z0-9-]+[.]run[.]app$") or not text.match_regex(expected_binding, "^[a-f0-9]{64}$") or (expected_scope != "HK" and expected_scope != "SG")}'
             raise: hk_snapshot_configuration_invalid
     - read_snapshot:
         try:
@@ -39,7 +40,7 @@ main:
           - now: ${sys.now()}
     - check_binding_and_fields:
         switch:
-          - condition: '${response.code != 200 or get_type(snapshot) != "map" or map.get(snapshot, "schema_version") != "longbridge_account_snapshot.v1" or map.get(snapshot, "status") != "partial" or map.get(snapshot, "account_scope") != "HK"}'
+          - condition: '${response.code != 200 or get_type(snapshot) != "map" or map.get(snapshot, "schema_version") != "longbridge_account_snapshot.v1" or map.get(snapshot, "status") != "partial" or map.get(snapshot, "account_scope") != expected_scope}'
             raise: hk_snapshot_contract_invalid
           - condition: '${map.get(snapshot, "no_order") != true or map.get(snapshot, "live_authority_granted") != false or map.get(snapshot, "cash_complete") != true or map.get(snapshot, "positions_complete") != true}'
             raise: hk_snapshot_contract_invalid
@@ -59,7 +60,7 @@ main:
         switch:
           - condition: ${archive_bucket == "" and archive_prefix == "" and account_uid == "" and target_id == "" and strategy_profile == ""}
             next: receipt
-          - condition: '${not text.match_regex(archive_bucket, "^[a-z0-9][a-z0-9._-]{2,62}$") or not text.match_regex(archive_prefix, "^[A-Za-z0-9_/-]+$") or account_uid == "" or target_id == "" or strategy_profile == ""}'
+          - condition: '${expected_scope != "HK" or not text.match_regex(archive_bucket, "^[a-z0-9][a-z0-9._-]{2,62}$") or not text.match_regex(archive_prefix, "^[A-Za-z0-9_/-]+$") or account_uid == "" or target_id == "" or strategy_profile == ""}'
             raise: hk_digest_archive_configuration_invalid
     - start_projection:
         assign:
@@ -149,7 +150,7 @@ main:
           position_count: ${len(snapshot.positions)}
     - receipt:
         return:
-          account_scope: HK
+          account_scope: ${expected_scope}
           source_binding_verified: true
           read_only_snapshot_verified: true
           observed_at: ${snapshot.observed_finished_at}

docs/account-snapshot-history.md

@@ -197,6 +197,10 @@ producer 为每次观察写入 `{prefix}/{target_id}/{source_binding_id}/{observ
 
 这份记录不是 TWR，不是收益率，也不授予 live 权限。
 
+## SG 快照开关的单独发布
+
+`Deploy Cloud Run` 的 `sg-serving-snapshot` 模式从受保护的 SG Environment 读取已核准的现役来源。流程先用云端预检核对完整现役配置、停用的交易开关和当前固定流量，然后暂存同一镜像并单独开启快照。新版本的配置和身份核对通过后，才将流量切到该版本。SG 的日报采集还需要独立验证真实快照及私有归档。
+
 ## HK 独立只读观察模式（默认关闭，仅手动）
 
 新增的 `ACCOUNT_HISTORY_OBSERVATION_MODE=independent_get` 是与旧归档路线分开的 HK-only 模式。缺省仍为 `scheduler_archive`，旧模式继续保留一次 Scheduler 触发、受限 GCS listing、固定 generation、原字节发布及原有 SG/HK 受限恢复保证。选择独立模式后，失败不会回退到 Scheduler、`/probe`、`/run`、`/dry-run` 或策略 canary。

docs/read-only-account-observation.zh-CN.md

@@ -28,7 +28,7 @@
 
 `independent_get` 适用于可由当前执行器访问的已鉴权 Cloud Run 服务。内部 ingress 返回固定原因 `internal_observer_required`，在调用券商或发送账户事实前停止；不得为采集开放 ingress。内部服务应复用受支持的内部执行入口，另行验证只读采集、私有归档与接收闭环。此文档不表示该内部闭环已经完成。
 
-HK 内网连通性可用 `config/hk_snapshot_verification.workflow.yaml` 单独验证。它在同一 GCP 项目通过 OIDC 调用固定的 `GET /account-snapshot`，从受保护的部署配置读取服务地址和预期绑定。验证只返回时间、字段数量和绑定结果。Cloud Workflows 的调用日志须关闭，运行身份单独配置，调用者不能传入地址、方法或请求参数。未配置归档时只做手动验证。HK 生产归档使用受保护的 `HK_DIGEST_BUCKET`、`HK_DIGEST_PREFIX`、`HK_DIGEST_ACCOUNT_UID`、`HK_DIGEST_TARGET_ID` 和 `HK_DIGEST_STRATEGY_PROFILE`，每天在新加坡时间 06:00 前采集，按 06:00 分界写入唯一的账户候选对象。只保留有来源绑定的净资产和股票持仓，不从快照推断订单或成交。归档权限仅允许在 HK 候选路径创建对象，候选已存在时保留原对象。
+HK 内网连通性可用 `config/hk_snapshot_verification.workflow.yaml` 单独验证。它在同一 GCP 项目通过 OIDC 调用固定的 `GET /account-snapshot`，从受保护的部署配置读取服务地址和预期绑定。验证只返回时间、字段数量和绑定结果。Cloud Workflows 的调用日志须关闭，运行身份单独配置，调用者不能传入地址、方法或请求参数。未配置归档时只做手动验证。SG 的部署可在只读快照开启后将 `SNAPSHOT_VERIFICATION_SCOPE` 设为 `SG`，使用其独立身份和服务来源绑定。SG 验证模式不写入 HK 归档。HK 生产归档使用受保护的 `HK_DIGEST_BUCKET`、`HK_DIGEST_PREFIX`、`HK_DIGEST_ACCOUNT_UID`、`HK_DIGEST_TARGET_ID` 和 `HK_DIGEST_STRATEGY_PROFILE`，每天在新加坡时间 06:00 前采集，按 06:00 分界写入唯一的账户候选对象。只保留有来源绑定的净资产和股票持仓，不从快照推断订单或成交。归档权限仅允许在 HK 候选路径创建对象，候选已存在时保留原对象。
 
 单次操作最多一次快照 GET 和一次事实 POST，不自动重试未知结果。投影保留券商实际币种、金额、观察时间及来源；账户来源不匹配时拒收，不补零、不改旧数据时间。交易启停、风险预算和通知不由此观察器控制。
 

scripts/enable_sg_serving_snapshot.py

@@ -0,0 +1,132 @@
+"""Enable the SG account snapshot on the verified serving image only."""
+from __future__ import annotations
+
+import copy
+import json
+import os
+import re
+import subprocess
+import time
+from urllib.parse import urlencode
+from urllib.request import Request, urlopen
+
+FLAG = "LONGBRIDGE_ACCOUNT_SNAPSHOT_ENABLED"
+API = "https://run.googleapis.com/v2/"
+REVISION_METADATA = {
+    "name", "uid", "generation", "createTime", "updateTime", "deleteTime", "expireTime",
+    "launchStage", "service", "reconciling", "conditions", "observedGeneration", "logUri",
+    "satisfiesPzs", "scalingStatus", "containerStatuses", "creator", "etag",
+}
+TEMPLATE_METADATA = {"revision", "client", "clientVersion"}
+
+
+def _require(value):
+    if not value:
+        raise ValueError("sg_snapshot_release_readback_mismatch")
+
+
+def _environment(container):
+    return {item["name"]: item for item in container.get("env", [])}
+
+
+def _revision_path(service, revision):
+    if "/revisions/" in revision:
+        _require(revision.startswith(service + "/revisions/"))
+        return revision
+    _require(re.fullmatch(r"[a-z0-9-]+", revision))
+    return service + "/revisions/" + revision
+
+
+def _require_matching_revision(template, revision):
+    wanted = {k: v for k, v in template.items() if k not in TEMPLATE_METADATA}
+    actual = {k: v for k, v in revision.items() if k not in REVISION_METADATA | TEMPLATE_METADATA}
+    expected_container, = wanted["containers"]
+    actual_container, = actual["containers"]
+    if "name" not in expected_container:
+        actual["containers"] = [{k: v for k, v in actual_container.items() if k != "name"}]
+    _require(wanted == actual)
+
+
+def enable_snapshot(request, service, approved_source):
+    _require(re.fullmatch(r"[a-f0-9]{40}", approved_source))
+    baseline = request(service)
+    _require(baseline["name"] == service and baseline["ingress"] == "INGRESS_TRAFFIC_INTERNAL_ONLY")
+    current_traffic, = baseline["traffic"]
+    active, = [r for r in baseline["trafficStatuses"] if r.get("percent", 0) > 0]
+    _require(current_traffic["type"] == "TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION")
+    _require(current_traffic["percent"] == active["percent"] == 100)
+    _require(not current_traffic.get("tag") and not active.get("tag"))
+    old_path = _revision_path(service, active["revision"])
+    _require(old_path == _revision_path(service, current_traffic["revision"]))
+    old = request(old_path)
+    _require(old.get("labels", {}).get("commit-sha") == approved_source)
+    template = copy.deepcopy(baseline["template"])
+    container, = template["containers"]
+    _require_matching_revision(template, old)
+    env = _environment(container)
+    _require(re.search(r"@sha256:[a-f0-9]{64}$", container["image"]))
+    _require(env["RUNTIME_TARGET_ENABLED"].get("value") == "false")
+    target = json.loads(env["RUNTIME_TARGET_JSON"]["value"])
+    _require(target["platform_id"] == "longbridge" and target["account_scope"].upper() == "SG")
+    if FLAG in env:
+        _require(env[FLAG].get("value") == "false" and "valueSource" not in env[FLAG])
+        env[FLAG]["value"] = "true"
+    else:
+        container["env"].append({"name": FLAG, "value": "true"})
+    template.pop("revision", None)
+    body = {"name": service, "etag": baseline["etag"], "template": template}
+    request(service + "?" + urlencode({"updateMask": "template", "validateOnly": "true"}), "PATCH", body)
+    _require(request(service)["etag"] == baseline["etag"])
+    staged = request(service + "?updateMask=template", "PATCH", body)
+    _wait(request, staged)
+
+    live = request(service)
+    _require(live["traffic"] == baseline["traffic"] and live["trafficStatuses"] == baseline["trafficStatuses"])
+    _require(live["ingress"] == baseline["ingress"] and live["template"]["serviceAccount"] == template["serviceAccount"])
+    new_path = _revision_path(service, live["latestReadyRevision"])
+    _require(new_path != old_path and new_path == _revision_path(service, live["latestCreatedRevision"]))
+    ready = request(new_path)
+    _require_matching_revision(template, ready)
+    route = {**current_traffic, "revision": new_path.rsplit("/", 1)[-1], "percent": 100}
+    moved = request(service + "?updateMask=traffic", "PATCH", {"name": service, "etag": live["etag"], "traffic": [route]})
+    _wait(request, moved)
+
+    final = request(service)
+    current, = [r for r in final["trafficStatuses"] if r.get("percent", 0) > 0]
+    _require(current["percent"] == 100 and _revision_path(service, current["revision"]) == new_path)
+    _require(final["template"] == live["template"] and final["ingress"] == baseline["ingress"])
+
+
+def _wait(request, operation):
+    deadline = time.monotonic() + 240
+    while not operation.get("done") and time.monotonic() < deadline:
+        time.sleep(4)
+        operation = request(operation["name"])
+    _require(operation.get("done") and not operation.get("error"))
+
+
+def main():
+    project, region, service = (os.environ[k] for k in ("GCP_PROJECT_ID", "CLOUD_RUN_REGION", "CLOUD_RUN_SERVICE"))
+    _require(all(re.fullmatch(r"[a-z0-9-]+", value) for value in (project, region, service)))
+    token = subprocess.check_output(["gcloud", "auth", "print-access-token"], text=True,
+                                    stderr=subprocess.DEVNULL, timeout=30).strip()
+
+    def request(path, method="GET", payload=None):
+        headers = {"Authorization": "Bearer " + token, "Content-Type": "application/json"}
+        body = json.dumps(payload).encode() if payload is not None else None
+        with urlopen(Request(API + path, headers=headers, method=method, data=body), timeout=30) as response:
+            result = json.load(response)
+        _require(not result.get("error"))
+        return result
+
+    approved_source = os.environ["SG_SNAPSHOT_APPROVED_SERVING_SOURCE"]
+    enable_snapshot(request, f"projects/{project}/locations/{region}/services/{service}", approved_source)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        print(json.dumps({"sg_snapshot_release_verified": False, "error_type": type(error).__name__}))
+        raise SystemExit(1) from None
+    print('{"sg_snapshot_release_verified":true,"serving_image_preserved":true,"trading_target_still_disabled":true}')

tests/test_enable_sg_serving_snapshot.py

@@ -0,0 +1,83 @@
+import copy
+import json
+
+import pytest
+
+from scripts.enable_sg_serving_snapshot import FLAG, enable_snapshot
+
+SOURCE = "2" * 40
+SERVICE = "projects/example/locations/example/services/sg"
+IMAGE = "example.invalid/source@sha256:" + "1" * 64
+
+
+def fixture(*, enabled=False, drift=False, stale_template=False):
+    old_env = [
+        {"name": "RUNTIME_TARGET_ENABLED", "value": "true" if enabled else "false"},
+        {"name": "RUNTIME_TARGET_JSON", "value": json.dumps({"platform_id": "longbridge", "account_scope": "SG"})},
+    ]
+    template = {"containers": [{"image": IMAGE, "env": old_env}], "serviceAccount": "example",
+                "labels": {"commit-sha": SOURCE}}
+    baseline = {"name": SERVICE, "ingress": "INGRESS_TRAFFIC_INTERNAL_ONLY",
+                "template": template, "etag": "before",
+                "traffic": [{"type": "TRAFFIC_TARGET_ALLOCATION_TYPE_REVISION", "revision": "sg-old", "percent": 100}],
+                "trafficStatuses": [{"revision": "sg-old", "percent": 100}]}
+    old = copy.deepcopy(template)
+    old["containers"][0]["name"] = "generated-container-name"
+    if stale_template:
+        template["containers"][0]["command"] = ["/different-command"]
+    writes, live = [], copy.deepcopy(baseline)
+
+    def request(path, method="GET", body=None):
+        nonlocal live
+        if method == "PATCH":
+            writes.append((path, copy.deepcopy(body)))
+            if "validateOnly" in path:
+                return {"done": True}
+            if path.endswith("updateMask=template"):
+                assert body["etag"] == "before"
+                live = {**live, "template": body["template"], "etag": "staged",
+                        "latestReadyRevision": "sg-new", "latestCreatedRevision": "sg-new"}
+            else:
+                assert body["etag"] == "staged"
+                live = {**live, "traffic": body["traffic"], "trafficStatuses": [{"revision": "sg-new", "percent": 100}]}
+            return {"done": True}
+        if path == SERVICE:
+            return copy.deepcopy(live)
+        if path.endswith("sg-old"):
+            return copy.deepcopy(old)
+        assert path.endswith("sg-new")
+        data = {**old, "containers": copy.deepcopy(live["template"]["containers"])}
+        data["containers"][0]["name"] = "generated-container-name"
+        if drift:
+            data["containers"][0]["env"][0]["value"] = "true"
+        return data
+
+    return request, writes
+
+
+def test_reuses_the_serving_image_and_preserves_other_environment_values():
+    request, writes = fixture()
+    enable_snapshot(request, SERVICE, SOURCE)
+    assert len(writes) == 3
+    staged = writes[1][1]["template"]["containers"][0]
+    assert staged["image"] == IMAGE
+    assert [entry for entry in staged["env"] if entry["name"] == FLAG] == [{"name": FLAG, "value": "true"}]
+    assert staged["env"][0]["value"] == "false"
+    assert writes[2][1]["traffic"][0]["revision"] == "sg-new"
+
+
+@pytest.mark.parametrize(("options", "writes_before_stop"), [
+    ({"enabled": True}, 0), ({"stale_template": True}, 0), ({"drift": True}, 2),
+])
+def test_rejects_enabled_trading_or_staged_configuration_drift(options, writes_before_stop):
+    request, writes = fixture(**options)
+    with pytest.raises(ValueError, match="readback_mismatch"):
+        enable_snapshot(request, SERVICE, SOURCE)
+    assert len(writes) == writes_before_stop
+
+
+def test_rejects_a_serving_source_that_does_not_match_the_protected_pin():
+    request, writes = fixture()
+    with pytest.raises(ValueError, match="readback_mismatch"):
+        enable_snapshot(request, SERVICE, "3" * 40)
+    assert writes == []

tests/test_runtime_monitor_workflows.py

@@ -215,3 +215,10 @@ def test_internal_hk_archive_is_one_account_and_create_only() -> None:
     assert "ifGenerationMatch: 0" in source
     assert "actually_ran: false" in source
     assert "order_count:" not in source and "fill_count:" not in source
+
+
+def test_internal_sg_verification_uses_sg_identity_without_hk_archive() -> None:
+    source = (ROOT / "config/hk_snapshot_verification.workflow.yaml").read_text()
+    assert 'sys.get_env("SNAPSHOT_VERIFICATION_SCOPE", "HK")' in source
+    assert 'map.get(snapshot, "account_scope") != expected_scope' in source
+    assert 'expected_scope != "HK" or not text.match_regex(archive_bucket' in source