fix(runtime): enforce Cloud Run service and tag budget

.github/workflows/manual-strategy-switch.yml

@@ -70,7 +70,7 @@ on:
         required: false
         type: string
       service_name:
-        description: "Runtime service name. blank = platform default."
+        description: "Exact service name; max 33 characters for cand-{sha8} (13). blank = platform default."
         required: false
         type: string
       plugin_mode:
@@ -491,7 +491,12 @@ jobs:
           if [ "${SERVICE_TARGETS_MODE}" = "allow_create" ]; then
             args+=(--allow-create-service-target)
           fi
-          python3 python/scripts/build_runtime_switch.py "${args[@]}"
+          # The builder validates defaults, overrides and all retained inventory names.
+          # Forward only a successful cand-{sha8} (13-character tag) contract result.
+          if ! python3 python/scripts/build_runtime_switch.py "${args[@]}"; then
+            echo "::error::Switch target refused: check service_name syntax and the 46-character service+tag budget (cand tag: 13). No inventory was published."
+            exit 2
+          fi
           python3 python/scripts/runtime_settings.py validate "${target_file}"
           echo "TARGET_FILE=${target_file}" >> "$GITHUB_ENV"
 

docs/manual-strategy-switch-permission-control.md

@@ -64,3 +64,20 @@ verified.
    Single-operator mode removes the need for extra team reviewers, not investment authorization.
 
 See the Chinese note for the full checklist, fork guidance, and threat-model notes.
+
+## Cloud Run service naming
+
+The builder and manual workflow use a fixed `cand-{sha8}` tagged-URL contract:
+the tag is 13 characters and service plus tag must fit within 46 characters.
+Generated defaults and explicit `--service-name` overrides therefore need valid
+lowercase service syntax and at most 33 characters. Names are preserved exactly
+after the existing input whitespace trimming; no service is automatically renamed.
+
+Headroom below 1 is `TAG_IMPOSSIBLE`; headroom 1–12 is `TAG_TIGHT` for this
+contract (including 9–12); both refuse generation before output or publication.
+Every retained name alias in an emitted `CLOUD_RUN_SERVICE_TARGETS_JSON` inventory
+is also checked. Existing long service identities require a separately approved
+migration; passing this check proves only offline naming compatibility. Broker
+`build_cloud_run_env_sync_plan.py` callers must validate their resolved exact name
+under their own deployment contract; their code and live configuration are outside
+this QRS change.

docs/manual-strategy-switch-permission-control.zh-CN.md

@@ -108,3 +108,17 @@
 - deployment branch 限制只允许 `main`
 - token 90 天轮换
 - 独立记录每次切换的 Actions run URL
+
+## Cloud Run 服务命名
+
+生成器和手动 workflow 使用固定的 `cand-{sha8}` 标签约定：标签长 13 个字符，
+服务名与标签合计最多 46 个字符，因此生成的默认名和 `--service-name` 覆盖值
+均须符合小写服务名语法且最长 33 个字符。沿用输入首尾空白处理，保留服务名，
+不会自动重命名服务。
+
+标签剩余空间小于 1 时为 `TAG_IMPOSSIBLE`，1–12 时为此约定下的 `TAG_TIGHT`
+（包括 9–12），两者均在生成输出或发布前拒绝。输出的
+`CLOUD_RUN_SERVICE_TARGETS_JSON` 中保留项的所有名称别名也接受检查。已有长名
+服务需要另行批准迁移；通过此检查只证明离线命名兼容。各 broker 的
+`build_cloud_run_env_sync_plan.py` 应按各自部署约定检查最终名称；其代码和线上
+配置不属于本次 QRS 修改范围。

python/scripts/build_runtime_switch.py

@@ -15,6 +15,8 @@
     sys.path.insert(0, str(SCRIPT_DIR))
 ROOT = SCRIPT_DIR.parents[1]
 
+from cloud_run_name_budget import validate_service_inventory, validate_service_name  # noqa: E402
+
 from runtime_settings import (  # noqa: E402
     LIVE_CONTINUITY_STATES,
     SUPPORTED_PLATFORMS,
@@ -826,6 +828,7 @@ def _build_runtime_target(args: argparse.Namespace) -> dict[str, Any]:
     )
     account_selector = _split_csv(args.account_selector) or _account_selector_default(platform, account_scope)
     service_name = args.service_name.strip() if args.service_name else _default_service_name(platform, target_name)
+    validate_service_name(service_name)
     strategy_profile = args.strategy_profile.strip().lower()
     _validate_requested_execution_mode(strategy_profile, args.execution_mode)
     _validate_dca_platform(platform, strategy_profile)
@@ -1100,6 +1103,9 @@ def build_switch_target(args: argparse.Namespace) -> dict[str, Any]:
         target["plugin_mounts"] = top_level_mounts
     if repository_variables:
         target["repository_variables"] = repository_variables
+    for variables in (extra_variables, repository_variables):
+        if "CLOUD_RUN_SERVICE_TARGETS_JSON" in variables:
+            validate_service_inventory(variables["CLOUD_RUN_SERVICE_TARGETS_JSON"])
     errors = validate_target(target)
     if errors:
         raise ValueError("; ".join(errors))
@@ -1117,7 +1123,10 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--deployment-selector", default="")
     parser.add_argument("--account-selector", default="")
     parser.add_argument("--account-scope", default="")
-    parser.add_argument("--service-name", default="")
+    parser.add_argument(
+        "--service-name", default="",
+        help="Exact service name; must fit the 46-character budget with cand-{sha8} (13 characters).",
+    )
     parser.add_argument("--execution-windows-json", default="")
     parser.add_argument(
         "--decision-data-json",

python/scripts/cloud_run_name_budget.py

@@ -0,0 +1,54 @@
+"""Offline preflight for the fixed cand-{sha8} tagged-URL contract.
+
+Local Option B implementation: validate exact names, never rename or allocate tags.
+The combined Cloud Run service + traffic tag budget is 46 characters.
+"""
+from __future__ import annotations
+
+import json
+import re
+from typing import Any
+
+SERVICE_TAG_BUDGET = 46
+CANDIDATE_TAG_LENGTH = 13  # cand- + eight hexadecimal commit characters
+
+
+def validate_service_name(service_name: Any) -> None:
+    """Refuse invalid or over-budget names without exposing resource identities."""
+    if not isinstance(service_name, str) or not re.fullmatch(r"[a-z](?:[a-z0-9-]*[a-z0-9])?", service_name):
+        raise ValueError("service_name must use lowercase ASCII letters, digits and internal hyphens")
+    headroom = SERVICE_TAG_BUDGET - len(service_name)
+    if headroom < CANDIDATE_TAG_LENGTH:
+        classification = "TAG_IMPOSSIBLE" if headroom < 1 else "TAG_TIGHT"
+        raise ValueError(
+            f"service_name {classification}: length={len(service_name)}, tag_headroom={headroom}; "
+            f"cand-{{sha8}} requires {CANDIDATE_TAG_LENGTH} characters within the "
+            f"{SERVICE_TAG_BUDGET}-character service+tag budget. "
+            "Inventory generation refused; existing services must not be silently renamed."
+        )
+
+
+def validate_service_inventory(value: Any) -> None:
+    """Check every retained name alias before a complete inventory is emitted."""
+    if isinstance(value, str):
+        try:
+            value = json.loads(value)
+        except json.JSONDecodeError:
+            raise ValueError("service inventory must contain valid JSON") from None
+    entries = value.get("targets") if isinstance(value, dict) else value
+    if not isinstance(entries, list):
+        raise ValueError("service inventory targets must be an array")
+    for entry in entries:
+        if not isinstance(entry, dict):
+            raise ValueError("service inventory entries must be objects")
+        names = [entry[key] for key in ("service", "service_name", "cloud_run_service") if key in entry]
+        if "runtime_target" in entry:
+            runtime_target = entry["runtime_target"]
+            if not isinstance(runtime_target, dict):
+                raise ValueError("service inventory runtime_target must be an object")
+            if "service_name" in runtime_target:
+                names.append(runtime_target["service_name"])
+        if not names:
+            raise ValueError("service inventory entry requires service_name")
+        for name in names:
+            validate_service_name(name)

python/tests/test_cloud_run_name_budget.py

@@ -0,0 +1,132 @@
+from __future__ import annotations
+
+import contextlib
+import io
+import json
+import os
+import re
+import subprocess
+import sys
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import patch
+
+SCRIPTS = Path(__file__).resolve().parents[1] / "scripts"
+sys.path.insert(0, str(SCRIPTS))
+import build_runtime_switch as switch
+
+
+class CloudRunNameBudgetTest(unittest.TestCase):
+    def args(self, *extra, platform="ibkr", target="paper"):
+        return switch.build_parser().parse_args([
+            "--platform", platform, "--target-name", target,
+            "--strategy-profile", "tqqq_growth_income", "--execution-mode", "dry_run", *extra,
+        ])
+
+    def test_impossible_override_refuses_without_echo_or_output(self):
+        name = "interactive-brokers-" + "x" * 20 + "-service"
+        self.assertEqual(len(name), 48)
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "target.json"
+            stderr = io.StringIO()
+            with contextlib.redirect_stderr(stderr):
+                status = switch.main([
+                    "--platform", "ibkr", "--target-name", "paper",
+                    "--strategy-profile", "tqqq_growth_income", "--execution-mode", "dry_run",
+                    "--service-name", name, "--output", str(output),
+                ])
+            self.assertEqual(status, 2)
+            self.assertFalse(output.exists())
+            self.assertIn("TAG_IMPOSSIBLE", stderr.getvalue())
+            self.assertNotIn(name, stderr.getvalue())
+
+    def test_tight_names_refuse_fixed_candidate_contract(self):
+        for length in (34, 37, 45):
+            with self.subTest(length=length), self.assertRaisesRegex(ValueError, "TAG_TIGHT"):
+                switch.build_switch_target(self.args("--service-name", "s" * length))
+
+    def test_generated_and_fixed_names_use_same_gate(self):
+        for platform in ("ibkr", "longbridge"):
+            with self.subTest(platform=platform), self.assertRaisesRegex(ValueError, "TAG_IMPOSSIBLE"):
+                switch.build_switch_target(self.args(platform=platform, target="x" * 30))
+        with patch.dict(switch.DEFAULT_SERVICE_NAME, {"firstrade": "s" * 48}):
+            with self.assertRaisesRegex(ValueError, "TAG_IMPOSSIBLE"):
+                switch.build_switch_target(self.args(platform="firstrade"))
+
+    def test_invalid_names_fail_without_echo(self):
+        for name in (" ", "Bad-Service", "-bad", "bad-", "bad_name", "é-service", "bad\nname"):
+            with self.subTest(name=name):
+                with self.assertRaisesRegex(ValueError, "service_name") as caught:
+                    switch.build_switch_target(self.args("--service-name=" + name))
+                if name.strip():
+                    self.assertNotIn(name, str(caught.exception))
+
+    def test_short_names_pass_unchanged_at_exact_boundary(self):
+        for name in ("s", "demo-service", "s" * 33):
+            with self.subTest(name=name):
+                target = switch.build_switch_target(self.args("--service-name", name))
+                self.assertEqual(target["runtime_target"]["service_name"], name)
+        for platform in switch.SUPPORTED_PLATFORMS:
+            with self.subTest(platform=platform):
+                target = switch._build_runtime_target(self.args(platform=platform))
+                self.assertLessEqual(len(target["service_name"]), 33)
+
+    def test_retained_inventory_aliases_cannot_bypass_gate(self):
+        for alias in ("service", "service_name", "cloud_run_service", "runtime_target"):
+            bad = {alias: {"service_name": "s" * 48} if alias == "runtime_target" else "s" * 48}
+            with self.subTest(alias=alias), tempfile.TemporaryDirectory() as directory:
+                path = Path(directory) / "inventory.json"
+                path.write_text(json.dumps({"targets": [bad]}))
+                before = path.read_bytes()
+                with self.assertRaisesRegex(ValueError, "TAG_IMPOSSIBLE"):
+                    switch.build_switch_target(self.args(
+                        "--service-name", "demo-service", "--allow-create-service-target",
+                        "--existing-service-targets-json-file", str(path),
+                    ))
+                self.assertEqual(path.read_bytes(), before)
+
+    def test_workflow_only_forwards_validated_target(self):
+        root = SCRIPTS.parents[1]
+        workflow = (root / ".github/workflows/manual-strategy-switch.yml").read_text()
+        step = workflow.split("      - name: Build switch target\n", 1)[1].split("      - name:", 1)[0]
+        script = "\n".join(line[10:] for line in step.split("        run: |\n", 1)[1].splitlines())
+        for name, status in (("s" * 48, 2), ("demo-service", 0)):
+            with self.subTest(status=status), tempfile.TemporaryDirectory() as directory:
+                env = {**os.environ, **{key: "" for key in re.findall(r"\$\{([A-Z_]+)", script)}}
+                env.update(PLATFORM="ibkr", TARGET_NAME="paper", STRATEGY_PROFILE="tqqq_growth_income",
+                           EXECUTION_MODE="dry_run", PLUGIN_MODE="none", OPTION_OVERLAY_MODE="current",
+                           VARIABLE_SCOPE="default", SERVICE_NAME=name, RUNNER_TEMP=directory,
+                           GITHUB_ENV=str(Path(directory) / "env"))
+                result = subprocess.run(["bash", "-c", script], cwd=root, env=env,
+                                        capture_output=True, text=True, timeout=30)
+                self.assertEqual(result.returncode, status, result.stderr)
+                output = Path(directory) / "runtime-switch-target.json"
+                self.assertEqual(output.exists(), status == 0)
+                if status:
+                    self.assertIn("::error::Switch target refused", result.stdout)
+                    self.assertFalse(Path(env["GITHUB_ENV"]).exists())
+                    self.assertNotIn(name, result.stdout + result.stderr)
+                else:
+                    self.assertIn("TARGET_FILE=", Path(env["GITHUB_ENV"]).read_text())
+                    self.assertEqual(json.loads(output.read_text())["runtime_target"]["service_name"], name)
+
+    def test_separate_cli_processes_refuse_same_invalid_name(self):
+        with tempfile.TemporaryDirectory() as directory:
+            output = Path(directory) / "target.json"
+            command = [sys.executable, str(SCRIPTS / "build_runtime_switch.py"),
+                       "--platform", "ibkr", "--target-name", "paper", "--strategy-profile",
+                       "tqqq_growth_income", "--execution-mode", "dry_run", "--service-name",
+                       "s" * 48, "--output", str(output)]
+            processes = [subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+                         for _ in range(2)]
+            results = [(process, process.communicate(timeout=30)) for process in processes]
+            for process, (stdout, stderr) in results:
+                self.assertEqual(process.returncode, 2)
+                self.assertEqual(stdout, "")
+                self.assertIn("TAG_IMPOSSIBLE", stderr)
+            self.assertFalse(output.exists())
+
+
+if __name__ == "__main__":
+    unittest.main()