Apply an already-saved HK stop with verified runtime controls

.github/workflows/runtime-target-lifecycle.yml

@@ -2,7 +2,7 @@ name: Runtime Target Lifecycle
 
 on:
   workflow_run:
-    workflows: ["Deploy Cloud Run"]
+    workflows: ["Deploy Cloud Run", "Stop HK Runtime"]
     types: [completed]
   workflow_dispatch:
     inputs:

.github/workflows/stop-hk-runtime.yml

@@ -0,0 +1,56 @@
+name: Stop HK Runtime
+
+on:
+  workflow_dispatch:
+    inputs:
+      stop_request:
+        description: "Exact existing HK identity from the settings console; no configuration or credentials."
+        required: true
+        type: string
+      confirm:
+        description: "STOP_ONLY: disable execution and pause its schedules; no cancellation or liquidation."
+        required: true
+        type: string
+
+permissions:
+  contents: read
+
+# Serialize with the platform's existing deployment writer, across both workflows.
+concurrency:
+  group: Deploy Cloud Run-${{ github.ref_name }}
+  cancel-in-progress: false
+
+jobs:
+  stop:
+    if: github.repository == 'QuantStrategyLab/LongBridgePlatform' && github.ref == 'refs/heads/main' && github.run_attempt == 1 && inputs.confirm == 'STOP_ONLY'
+    runs-on: ubuntu-latest
+    timeout-minutes: 10
+    environment: longbridge-hk
+    permissions:
+      contents: read
+      id-token: write
+    env:
+      RUNTIME_TARGET_ENABLED: ${{ vars.RUNTIME_TARGET_ENABLED }}
+      RUNTIME_TARGET_JSON: ${{ vars.RUNTIME_TARGET_JSON || secrets.RUNTIME_TARGET_JSON }}
+    steps:
+      - name: Checkout exact workflow source
+        uses: actions/checkout@v6
+        with:
+          ref: ${{ github.sha }}
+          persist-credentials: false
+      - name: Authenticate existing HK deployment identity
+        uses: google-github-actions/auth@v3
+        with:
+          workload_identity_provider: projects/252919773759/locations/global/workloadIdentityPools/github-actions/providers/github-main
+          service_account: longbridge-platform-deploy@longbridgequant.iam.gserviceaccount.com
+      - name: Set up cloud client
+        uses: google-github-actions/setup-gcloud@v3
+        with:
+          project_id: longbridgequant
+          version: ">= 416.0.0"
+      - name: Apply stop and verify service and Scheduler
+        run: python3 scripts/stop_hk_runtime.py
+      - name: Explain stop boundary
+        if: always()
+        run: |
+          echo 'This operation only disables the existing HK runtime and pauses its schedules. A successful run verifies these controls; in-flight requests, broker orders and account retirement remain unverified. Do not retry an uncertain write.' >> "$GITHUB_STEP_SUMMARY"

scripts/stop_hk_runtime.py

@@ -0,0 +1,166 @@
+"""Apply only the existing HK target's stop; cloud responses stay in memory.
+
+This disables future execution and its Scheduler jobs. It neither terminates
+in-flight requests nor cancels orders, liquidates positions or retires an account.
+"""
+
+import copy
+import json
+import os
+from pathlib import Path
+import re
+import subprocess
+from urllib.parse import urlsplit
+
+PROJECT = "longbridgequant"
+REGION = "asia-east2"
+SERVICE = "longbridge-quant-hk-service"
+IDENTITY_FIELDS = {"platform_id", "deployment_selector", "account_selector", "account_scope", "service_name"}
+
+
+class StopError(ValueError):
+    """Only fixed, non-sensitive failure categories may leave this adapter."""
+
+
+def _identity(request, environment):
+    try:
+        if (set(request) != {"target_id", "github", "runtime_target"}
+                or request["target_id"] != "longbridge/hk"
+                or request["github"] != {"repository": "QuantStrategyLab/LongBridgePlatform",
+                                          "variable_scope": "environment", "environment": "longbridge-hk"}
+                or environment.get("RUNTIME_TARGET_ENABLED") != "false"):
+            raise ValueError
+        identity = request["runtime_target"]
+        declared = json.loads(environment["RUNTIME_TARGET_JSON"])
+        if (not isinstance(identity, dict) or set(identity) != IDENTITY_FIELDS
+                or not isinstance(declared, dict)
+                or any(identity[key] != declared.get(key) for key in IDENTITY_FIELDS)
+                or identity["platform_id"] != "longbridge" or identity["service_name"] != SERVICE
+                or identity["account_scope"] != "HK"
+                or not isinstance(identity["deployment_selector"], str) or not identity["deployment_selector"].strip()
+                or not isinstance(identity["account_selector"], list) or not identity["account_selector"]
+                or any(not isinstance(item, str) or not item.strip() for item in identity["account_selector"])
+                or len(set(identity["account_selector"])) != len(identity["account_selector"])):
+            raise ValueError
+        return identity
+    except (ValueError, TypeError, KeyError):
+        raise StopError("stop_request_rejected") from None
+
+
+def _container(raw):
+    containers = raw.get("containers", [])
+    if len(containers) != 1 or not isinstance(containers[0], dict):
+        raise StopError("stop_source_unverified")
+    container = copy.deepcopy(containers[0])
+    values = container.get("env", [])
+    if (not isinstance(values, list) or any(not isinstance(item, dict) or not isinstance(item.get("name"), str) for item in values)
+            or len({item["name"] for item in values}) != len(values)):
+        raise StopError("stop_source_unverified")
+    container["env"] = sorted(values, key=lambda item: item["name"])
+    return container
+
+
+def execute_stop(request, environment, *, run=subprocess.run):
+    identity = _identity(request, environment)
+
+    def cloud(args, *, write=False):
+        try:
+            result = run(["gcloud", *args, f"--project={PROJECT}", "--format=json", "--quiet"],
+                         capture_output=True, text=True, timeout=45, check=False,
+                         env={**os.environ, "CLOUDSDK_CORE_DISABLE_FILE_LOGGING": "1", "CLOUDSDK_CORE_LOG_HTTP": "false"})
+            if result.returncode:
+                raise ValueError
+            return json.loads(result.stdout or "{}")
+        except (ValueError, OSError, subprocess.SubprocessError):
+            raise StopError("stop_write_unverified" if write else "stop_source_unverified") from None
+
+    def snapshot():
+        try:
+            service = cloud(["run", "services", "describe", SERVICE, f"--region={REGION}"])
+            status = service["status"]
+            traffic = [item for item in status.get("traffic", []) if item.get("percent", 0) > 0]
+            if (service["metadata"]["name"] != SERVICE or len(traffic) != 1 or traffic[0].get("percent") != 100
+                    or status.get("latestCreatedRevisionName") != status.get("latestReadyRevisionName")):
+                raise ValueError
+            revision = traffic[0]["revisionName"]
+            if revision != status.get("latestReadyRevisionName") or not re.fullmatch(r"[a-z][a-z0-9-]*", revision):
+                raise ValueError
+            container = _container(cloud(["run", "revisions", "describe", revision, f"--region={REGION}"])["spec"])
+            if _container(service["spec"]["template"]["spec"]) != container:
+                raise ValueError
+            values = {item["name"]: item.get("value") for item in container["env"]}
+            target = json.loads(values["RUNTIME_TARGET_JSON"])
+            if any(target.get(key) != value for key, value in identity.items()) or values.get("RUNTIME_TARGET_ENABLED") not in {"true", "false"}:
+                raise ValueError
+            url = urlsplit(status["url"])
+            if url.scheme != "https" or not url.hostname or url.username or url.password or url.query or url.fragment:
+                raise ValueError
+            jobs = cloud(["scheduler", "jobs", "list", f"--location={REGION}"])
+            if not isinstance(jobs, list):
+                raise ValueError
+            selected = []
+            for job in jobs:
+                uri = urlsplit(job.get("httpTarget", {}).get("uri", ""))
+                if uri.scheme == "https" and uri.netloc == url.netloc:
+                    if (not re.fullmatch(rf"projects/{PROJECT}/locations/{REGION}/jobs/[A-Za-z0-9_-]+", job.get("name", ""))
+                            or job.get("state") not in {"ENABLED", "PAUSED"}):
+                        raise ValueError
+                    selected.append({"name": job["name"], "state": job["state"], "uri": uri.geturl()})
+            if not selected or len({job["name"] for job in selected}) != len(selected):
+                raise ValueError
+            return {"container": container, "jobs": sorted(selected, key=lambda item: item["name"]), "url": url.geturl(), "revision": revision}
+        except StopError:
+            raise
+        except (ValueError, TypeError, KeyError, AttributeError):
+            raise StopError("stop_source_unverified") from None
+
+    before = snapshot()
+    # Guard stale source/configuration before writing; this is not a cloud CAS.
+    if snapshot() != before:
+        raise StopError("stop_source_changed")
+    expected = copy.deepcopy(before["container"])
+    enabled = next(item for item in expected["env"] if item["name"] == "RUNTIME_TARGET_ENABLED")
+    was_enabled = enabled["value"] == "true"
+    enabled["value"] = "false"
+    if was_enabled:
+        cloud(["run", "services", "update", SERVICE, f"--region={REGION}",
+               "--update-env-vars=RUNTIME_TARGET_ENABLED=false"], write=True)
+        after = snapshot()
+        if after["container"] != expected or after["url"] != before["url"] or after["jobs"] != before["jobs"]:
+            raise StopError("stop_readback_unverified")
+    for job in before["jobs"]:
+        if job["state"] == "ENABLED":
+            cloud(["scheduler", "jobs", "pause", job["name"], f"--location={REGION}"], write=True)
+    after = snapshot()
+    expected_jobs = [{**job, "state": "PAUSED"} for job in before["jobs"]]
+    if after["container"] != expected or after["url"] != before["url"] or after["jobs"] != expected_jobs:
+        raise StopError("stop_readback_unverified")
+    return {"platform_applied": True, "runtime_enabled": False, "scheduler_state": "paused",
+            "in_flight_state": "unknown", "retirement_complete": False, "no_order": True}
+
+
+def main():
+    try:
+        if (os.environ.get("GITHUB_EVENT_NAME") != "workflow_dispatch"
+                or os.environ.get("GITHUB_REF") != "refs/heads/main"
+                or os.environ.get("GITHUB_REPOSITORY") != "QuantStrategyLab/LongBridgePlatform"
+                or os.environ.get("GITHUB_RUN_ATTEMPT") != "1"):
+            raise StopError("stop_workflow_rejected")
+        path = Path(os.environ["GITHUB_EVENT_PATH"])
+        if path.is_symlink() or not path.is_file() or path.stat().st_size > 1024 * 1024:
+            raise StopError("stop_request_rejected")
+        event = json.loads(path.read_text())
+        if event["inputs"].get("confirm") != "STOP_ONLY":
+            raise StopError("stop_request_rejected")
+        request = json.loads(event["inputs"]["stop_request"])
+        print(json.dumps(execute_stop(request, os.environ), sort_keys=True))
+        return 0
+    except StopError as error:
+        print(json.dumps({"platform_applied": None, "error": str(error), "retirement_complete": False}))
+    except (ValueError, TypeError, KeyError, OSError):
+        print(json.dumps({"platform_applied": None, "error": "stop_request_rejected", "retirement_complete": False}))
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

tests/test_runtime_monitor_workflows.py

@@ -71,7 +71,7 @@ def test_cloud_run_deployment_requires_manual_dispatch_and_lifecycle_observes_co
     assert "if: github.event_name == 'workflow_dispatch'" in deploy
 
     assert "workflow_run:" in lifecycle
-    assert 'workflows: ["Deploy Cloud Run"]' in lifecycle
+    assert 'workflows: ["Deploy Cloud Run", "Stop HK Runtime"]' in lifecycle
     assert "types: [completed]" in lifecycle
     assert "github.event.workflow_run.conclusion" not in lifecycle
     assert 'observe-gcp: "true"' in lifecycle

tests/test_stop_hk_runtime.py

@@ -0,0 +1,141 @@
+"""Synthetic cloud adapter tests: no credentials, broker calls or real resources."""
+
+import copy
+import json
+import subprocess
+import unittest
+
+from scripts.stop_hk_runtime import StopError, execute_stop
+
+
+class Cloud:
+    def __init__(self):
+        self.identity = {"platform_id": "longbridge", "deployment_selector": "synthetic-hk",
+                         "account_selector": ["synthetic-account"], "account_scope": "HK",
+                         "service_name": "longbridge-quant-hk-service"}
+        self.target = {**self.identity, "strategy_profile": "synthetic-strategy"}
+        self.request = {"target_id": "longbridge/hk", "runtime_target": self.identity,
+                        "github": {"repository": "QuantStrategyLab/LongBridgePlatform",
+                                   "variable_scope": "environment", "environment": "longbridge-hk"}}
+        self.environment = {"RUNTIME_TARGET_ENABLED": "false", "RUNTIME_TARGET_JSON": json.dumps(self.target)}
+        self.container = {"image": "synthetic-image@sha256:" + "a" * 64, "env": [
+            {"name": "RUNTIME_TARGET_ENABLED", "value": "true"},
+            {"name": "RUNTIME_TARGET_JSON", "value": json.dumps(self.target)},
+            {"name": "UNCHANGED", "value": "synthetic-only"}]}
+        self.service = {"metadata": {"name": "longbridge-quant-hk-service"},
+                        "spec": {"template": {"spec": {"containers": [self.container]}}},
+                        "status": {"url": "https://synthetic.example", "latestReadyRevisionName": "synthetic-old",
+                                   "latestCreatedRevisionName": "synthetic-old",
+                                   "traffic": [{"revisionName": "synthetic-old", "percent": 100}]}}
+        prefix = "projects/longbridgequant/locations/asia-east2/jobs/"
+        self.jobs = [{"name": prefix + "synthetic-main", "state": "ENABLED",
+                      "httpTarget": {"uri": "https://synthetic.example/run"}},
+                     {"name": prefix + "synthetic-probe", "state": "PAUSED",
+                      "httpTarget": {"uri": "https://synthetic.example/probe"}},
+                     {"name": prefix + "synthetic-other", "state": "ENABLED",
+                      "httpTarget": {"uri": "https://other.example/run"}}]
+        self.calls = []
+        self.fail_pause = False
+        self.fail_update = False
+        self.corrupt_after_update = False
+
+    def run(self, command, **kwargs):
+        self.calls.append(command)
+        assert kwargs["capture_output"] and kwargs["timeout"] <= 60
+        args = command[1:]
+        if args[:3] == ["run", "services", "describe"]:
+            payload = self.service
+        elif args[:3] == ["run", "revisions", "describe"]:
+            payload = {"spec": {"containers": [self.container]}}
+        elif args[:3] == ["scheduler", "jobs", "list"]:
+            payload = self.jobs
+        elif args[:3] == ["run", "services", "update"]:
+            if self.fail_update:
+                raise subprocess.TimeoutExpired(command, 45, output="synthetic-private-error")
+            self.container["env"][0]["value"] = "false"
+            if self.corrupt_after_update:
+                self.container["env"][2]["value"] = "changed"
+            payload = self.service
+        elif args[:3] == ["scheduler", "jobs", "pause"]:
+            if self.fail_pause:
+                return subprocess.CompletedProcess(command, 1, "", "synthetic-private-error")
+            next(job for job in self.jobs if job["name"] == args[3])["state"] = "PAUSED"
+            payload = {}
+        else:
+            raise AssertionError(command)
+        return subprocess.CompletedProcess(command, 0, json.dumps(payload), "")
+
+    def writes(self):
+        return [call for call in self.calls if "update" in call or "pause" in call]
+
+
+class StopHkRuntimeTests(unittest.TestCase):
+    def test_success_disables_only_bound_service_and_pauses_its_jobs(self):
+        cloud = Cloud()
+        result = execute_stop(cloud.request, cloud.environment, run=cloud.run)
+        self.assertTrue(result["platform_applied"])
+        self.assertEqual(result["scheduler_state"], "paused")
+        self.assertEqual(result["in_flight_state"], "unknown")
+        self.assertFalse(result["retirement_complete"])
+        self.assertEqual(len(cloud.writes()), 2)
+        self.assertIn("--update-env-vars=RUNTIME_TARGET_ENABLED=false", cloud.writes()[0])
+        self.assertEqual(cloud.jobs[-1]["state"], "ENABLED")
+
+    def test_already_stopped_is_verified_without_writes(self):
+        cloud = Cloud()
+        cloud.container["env"][0]["value"] = "false"
+        cloud.jobs[0]["state"] = "PAUSED"
+        self.assertTrue(execute_stop(cloud.request, cloud.environment, run=cloud.run)["platform_applied"])
+        self.assertEqual(cloud.writes(), [])
+
+    def test_bad_request_or_enabled_desire_does_not_contact_cloud(self):
+        for mutation in [lambda c: c.request.update(target_id="longbridge/sg"),
+                         lambda c: c.environment.update(RUNTIME_TARGET_ENABLED="true"),
+                         lambda c: c.request.update(enabled=True),
+                         lambda c: c.request["github"].update(environment="longbridge-sg"),
+                         lambda c: c.request["runtime_target"].update(account_selector=["other"]),
+                         lambda c: c.environment.update(RUNTIME_TARGET_JSON="null")]:
+            with self.subTest(mutation=mutation):
+                cloud = Cloud()
+                mutation(cloud)
+                with self.assertRaises(StopError):
+                    execute_stop(cloud.request, cloud.environment, run=cloud.run)
+                self.assertEqual(cloud.calls, [])
+
+    def test_ambiguous_or_mismatched_cloud_prevents_writes(self):
+        for mutation in [lambda c: c.service["status"].update(traffic=[]),
+                         lambda c: c.service["status"].update(latestCreatedRevisionName="pending"),
+                         lambda c: c.container["env"][1].update(value='{"service_name":"other"}'),
+                         lambda c: c.jobs[0].update(state="UNKNOWN"),
+                         lambda c: c.jobs.clear(),
+                         lambda c: c.container["env"].append(copy.deepcopy(c.container["env"][0]))]:
+            with self.subTest(mutation=mutation):
+                cloud = Cloud()
+                mutation(cloud)
+                with self.assertRaises(StopError):
+                    execute_stop(cloud.request, cloud.environment, run=cloud.run)
+                self.assertEqual(cloud.writes(), [])
+
+    def test_unknown_update_is_not_retried_or_followed_by_other_writes(self):
+        cloud = Cloud()
+        cloud.fail_update = True
+        with self.assertRaisesRegex(StopError, "^stop_write_unverified$"):
+            execute_stop(cloud.request, cloud.environment, run=cloud.run)
+        self.assertEqual(len(cloud.writes()), 1)
+
+    def test_pause_failure_is_sanitized_and_stops(self):
+        cloud = Cloud()
+        cloud.fail_pause = True
+        with self.assertRaisesRegex(StopError, "^stop_write_unverified$"):
+            execute_stop(cloud.request, cloud.environment, run=cloud.run)
+        self.assertEqual(len(cloud.writes()), 2)
+
+    def test_unrelated_configuration_change_fails_readback(self):
+        cloud = Cloud()
+        cloud.corrupt_after_update = True
+        with self.assertRaisesRegex(StopError, "^stop_readback_unverified$"):
+            execute_stop(cloud.request, cloud.environment, run=cloud.run)
+
+
+if __name__ == "__main__":
+    unittest.main()