From 9ac7b70e733a122a570307ef85eea99b7cb24df4 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 10 Sep 2026 01:20:04 +0800 Subject: [PATCH 1/2] Apply an already-saved HK runtime stop with verified controls Co-Authored-By: Codex --- .../workflows/runtime-target-lifecycle.yml | 2 +- .github/workflows/stop-hk-runtime.yml | 56 ++++++ scripts/stop_hk_runtime.py | 166 ++++++++++++++++++ tests/test_stop_hk_runtime.py | 141 +++++++++++++++ 4 files changed, 364 insertions(+), 1 deletion(-) create mode 100644 .github/workflows/stop-hk-runtime.yml create mode 100644 scripts/stop_hk_runtime.py create mode 100644 tests/test_stop_hk_runtime.py diff --git a/.github/workflows/runtime-target-lifecycle.yml b/.github/workflows/runtime-target-lifecycle.yml index c133081..73df7f4 100644 --- a/.github/workflows/runtime-target-lifecycle.yml +++ b/.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: diff --git a/.github/workflows/stop-hk-runtime.yml b/.github/workflows/stop-hk-runtime.yml new file mode 100644 index 0000000..44a4bde --- /dev/null +++ b/.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" diff --git a/scripts/stop_hk_runtime.py b/scripts/stop_hk_runtime.py new file mode 100644 index 0000000..896ee10 --- /dev/null +++ b/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()) diff --git a/tests/test_stop_hk_runtime.py b/tests/test_stop_hk_runtime.py new file mode 100644 index 0000000..d7ffc0a --- /dev/null +++ b/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() From b6b7c889f8013366b5154c979aa65e50ef86a7ac Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Thu, 10 Sep 2026 01:25:02 +0800 Subject: [PATCH 2/2] Update lifecycle trigger assertion for HK stop readback Co-Authored-By: Codex --- tests/test_runtime_monitor_workflows.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_runtime_monitor_workflows.py b/tests/test_runtime_monitor_workflows.py index ea2d73e..1c0359b 100644 --- a/tests/test_runtime_monitor_workflows.py +++ b/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