fix: Adopt shared Cloud Run tag preflight

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

@@ -288,6 +288,7 @@ jobs:
             exit 1
           fi
 
+      # Shared exact-name policy retains cbd and refuses incompatible zero-traffic diagnostics.
       - name: Validate fixed diagnostic traffic tag budget
         env:
           EXPECTED_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }}

NOTICE.md

@@ -12,3 +12,7 @@ not affiliated with, endorsed by, or supported by Firstrade Securities Inc.
 Keep this notice and the upstream license attribution when distributing this
 repository or derivative work.
 
+
+The Cloud Run tag preflight excerpt in `scripts/_vendor/cloud_run_tags.py` is
+from QuantStrategyLab/QuantPlatformKit (MIT). Its license and versioned source
+provenance are retained in `scripts/_vendor/`.

README.md

@@ -61,6 +61,16 @@ Direct runtime profiles can usually run from market history or portfolio state.
 4. Confirm rollback steps and artifact versions.
 5. Enable scheduled or live execution only after the above checks are clear.
 
+The sync plan and cached-balance staging preflight use the versioned shared
+Cloud Run tag contract before emitting a plan or staging a candidate. Service
+names are validated exactly, without renaming. The diagnostic keeps the fixed
+`cbd` tag and requires `len(service_name) + 3 <= 46`; incompatible names are
+refused rather than shortened or given traffic. Existing diagnostic ownership,
+zero-traffic, identity and readback checks still apply. The legacy `cloudbuild.yaml`
+image-only path validates the exact service name before building; it allocates no
+tag and gains no traffic authority. See the [vendored contract](scripts/_vendor/README.md)
+for provenance and update rules.
+
 ## Read-only account facts
 
 The existing **Diagnose cached-balance stage failure** manual workflow can inspect saved

cloudbuild.yaml

@@ -1,4 +1,10 @@
 steps:
+  - id: ServicePreflight
+    name: python:3.12-slim
+    entrypoint: python3
+    args:
+      - scripts/cloud_run_tag_preflight.py
+      - --service-name=$_SERVICE_NAME
   - id: Buildpack
     name: gcr.io/k8s-skaffold/pack
     entrypoint: pack

scripts/_vendor/LICENSE

@@ -0,0 +1,21 @@
+MIT License
+
+Copyright (c) 2026 QuantStrategyLab
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

scripts/_vendor/README.md

@@ -0,0 +1,27 @@
+# Cloud Run tag policy excerpt
+
+This directory vendors only `CONTRACT_VERSION`, `TAGGED_URL_BUDGET` and `headroom`
+from QuantPlatformKit `src/quant_platform_kit/operators/cloud_run_tags.py`, contract
+`cloud-run-tags/v1`, under its MIT license. These definitions are copied verbatim.
+The preflight excerpt has no dependencies, allocation journal or traffic API, so
+both the workflow's Python runner and the legacy Cloud Build path can use it
+without installing or advancing the runtime's pinned shared packages.
+
+The caller validates Cloud Run service syntax and retains the fixed `cbd` contract.
+It checks the original value before display masking, never renames or shortens a
+service, and refuses diagnostics when the three-character tag does not fit. The
+image-only Cloud Build entry point validates service syntax without requiring a
+tag. Passing preflight grants no deployment or traffic permission.
+
+When updating, copy the three definitions together from one exact reviewed QPK
+revision, update the source pin below, and run the caller boundary tests. Do not
+add allocation, ownership journals or micro-split fallback to these diagnostic or
+image-only callers. Existing staging baseline, identity, no-traffic, tag ownership
+and post-mutation readback checks remain required.
+
+Source pin: local contract snapshot from `qpk-cr-tag-helper-r1`, version
+`cloud-run-tags/v1`. The upstream Draft was unavailable at adoption, so this is the
+bounded local-copy fallback. Full source snapshot SHA-256:
+`ecb8762ddd8e76c1fb6bf4de2981347951ba9640442632b1e3aaa459d5fc8c05`.
+Only the three definitions listed above are vendored; no runtime dependency pin
+changes are needed. The delivery report retains the upstream availability readback.

scripts/_vendor/__init__.py

@@ -0,0 +1 @@
+"""Versioned offline operator policy excerpts; see README.md."""

scripts/_vendor/cloud_run_tags.py

@@ -0,0 +1,13 @@
+"""Pure preflight excerpt from QuantPlatformKit operators/cloud_run_tags.py.
+
+MIT License, Copyright (c) 2026 QuantStrategyLab.
+See README.md for provenance. Allocation and traffic APIs are intentionally absent.
+"""
+
+CONTRACT_VERSION = 'cloud-run-tags/v1'
+
+TAGGED_URL_BUDGET = 46
+
+def headroom(service_name: str) -> int:
+    """Measure the exact original name, before any display masking."""
+    return TAGGED_URL_BUDGET - len(service_name)

scripts/build_cloud_run_env_sync_plan.py

@@ -39,6 +39,8 @@ def _should_add_local_src(candidate: Path) -> bool:
     if candidate_str not in sys.path:
         sys.path.insert(0, candidate_str)
 
+from scripts.cloud_run_tag_preflight import validate_diagnostic
+
 from strategy_registry import (  # noqa: E402
     FIRSTRADE_PLATFORM,
     get_platform_profile_status_matrix,
@@ -376,11 +378,11 @@ def _load_target_entries(
         entries = [_coerce_mapping(item) for item in raw_entries]
         return entries, defaults, True
 
-    raw_services = str(env.get("CLOUD_RUN_SERVICES", "") or "").strip()
+    raw_services = str(env.get("CLOUD_RUN_SERVICES", "") or "")
     if not raw_services:
-        raw_services = str(env.get("CLOUD_RUN_SERVICE", "") or "").strip()
+        raw_services = str(env.get("CLOUD_RUN_SERVICE", "") or "")
     services = [
-        item.strip()
+        item
         for chunk in raw_services.replace(";", ",").replace("\n", ",").split(",")
         for item in [chunk]
         if item.strip()
@@ -404,11 +406,16 @@ def _build_target_plan(
     )
     runtime_target = _resolve_runtime_target(target, defaults, env, per_service_mode)
     if not service_name:
-        service_name = str(runtime_target.get("service_name") or "").strip()
+        service_name = runtime_target.get("service_name") or ""
     if not service_name:
         raise ValueError("Each Cloud Run sync target requires service/service_name")
 
-    runtime_target_service = str(runtime_target.get("service_name") or "").strip()
+    # Sync targets must support the fixed no-traffic diagnostic lifecycle.
+    # Check the exact configured value; never repair or rename it silently.
+    validate_diagnostic(service_name)
+    runtime_target_service = runtime_target.get("service_name")
+    if runtime_target_service is not None and runtime_target_service != "":
+        validate_diagnostic(runtime_target_service)
     if runtime_target_service and runtime_target_service != service_name:
         raise ValueError(
             f"Target {service_name} runtime_target.service_name={runtime_target_service!r} does not match"

scripts/cloud_run_tag_preflight.py

@@ -0,0 +1,53 @@
+"""Offline exact-name preflight; never allocate tags or authorize traffic."""
+from __future__ import annotations
+
+import argparse
+import re
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts._vendor.cloud_run_tags import CONTRACT_VERSION, TAGGED_URL_BUDGET, headroom
+
+DIAGNOSTIC_TAG = "cbd"
+MIN_DIAGNOSTIC_TAG_LENGTH = 3
+
+
+def validate_service(service_name: str) -> int:
+    """Validate the original identity before any formatting or masking."""
+    if (not isinstance(service_name, str) or len(service_name) > 63
+            or not re.fullmatch(r"[a-z](?:[a-z0-9-]*[a-z0-9])?", service_name)):
+        raise ValueError("INVALID_SERVICE_NAME")
+    return headroom(service_name)
+
+
+def validate_diagnostic(service_name: str, tag: str = DIAGNOSTIC_TAG) -> None:
+    room = validate_service(service_name)
+    if tag != DIAGNOSTIC_TAG:
+        raise ValueError("diagnostic_tag_mismatch")
+    if len(tag) < MIN_DIAGNOSTIC_TAG_LENGTH:
+        raise ValueError("diagnostic_tag_too_short")
+    # A shortened tag would break the fixed diagnostic URL/readback contract.
+    # Refuse even when the general allocator could fit a shorter tag.
+    if len(tag) > room:
+        raise ValueError("diagnostic_tag_name_budget_exceeded")
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--service-name", required=True)
+    args = parser.parse_args()
+    try:
+        validate_service(args.service_name)
+    except ValueError as exc:
+        print(str(exc), file=sys.stderr)
+        return 1
+    print("cloud_run_service_preflight_ok")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/verify_cached_diagnostic_stage.py

@@ -11,11 +11,19 @@
 from typing import Any
 
 
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from scripts import cloud_run_tag_preflight
+
+
 DIAGNOSTIC_GATE = "FIRSTRADE_CACHED_BALANCE_DIAGNOSTIC_ON_HTTP"
 RUNTIME_TARGET_KEYS = ("QSL_RUNTIME_TARGET_JSON", "RUNTIME_TARGET_JSON")
 DIAGNOSTIC_TAG = "cbd"
-MIN_TRAFFIC_TAG_LENGTH = 3
-MAX_SERVICE_AND_TRAFFIC_TAG_LENGTH = 46
+# Retain the diagnostic modules' public constants while sharing the policy values.
+MIN_TRAFFIC_TAG_LENGTH = cloud_run_tag_preflight.MIN_DIAGNOSTIC_TAG_LENGTH
+MAX_SERVICE_AND_TRAFFIC_TAG_LENGTH = cloud_run_tag_preflight.TAGGED_URL_BUDGET
 EXPECTED_PLATFORM_ID = "firstrade"
 GENERATED_TEMPLATE_ANNOTATIONS = {
     "run.googleapis.com/client-name",
@@ -67,10 +75,9 @@ def _validate_diagnostic_tag_budget(service_name: str, tag: str) -> None:
         raise ValueError("diagnostic_service_name_missing")
     if tag != DIAGNOSTIC_TAG:
         raise ValueError("diagnostic_tag_mismatch")
-    if len(tag) < MIN_TRAFFIC_TAG_LENGTH:
+    if len(tag) < cloud_run_tag_preflight.MIN_DIAGNOSTIC_TAG_LENGTH:
         raise ValueError("diagnostic_tag_too_short")
-    if len(service_name) + len(tag) > MAX_SERVICE_AND_TRAFFIC_TAG_LENGTH:
-        raise ValueError("diagnostic_tag_name_budget_exceeded")
+    cloud_run_tag_preflight.validate_diagnostic(service_name, tag)
 
 
 def _normalize_template_metadata(template: dict[str, Any]) -> None:

tests/test_cloud_run_tag_preflight.py

@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+import pytest
+
+from scripts import verify_cached_diagnostic_stage as stage
+
+ROOT = Path(__file__).resolve().parents[1]
+
+
+@pytest.mark.parametrize("name", ["Bad-Service", "service/other", " service", "service ", "service-", "1service", "s" * 64])
+def test_diagnostic_rejects_invalid_exact_name_without_echo(name, monkeypatch, capsys):
+    monkeypatch.setattr(sys, "argv", ["verify_cached_diagnostic_stage.py", "tag-budget",
+                                    "--expected-service", name, "--tag", "cbd"])
+    assert stage.main() == 1
+    output = capsys.readouterr()
+    assert output.out == ""
+    assert output.err == "cached_diagnostic_stage_blocked:INVALID_SERVICE_NAME\n"
+    assert name not in output.err
+
+
+@pytest.mark.parametrize("name", ["Bad-Service", " service ", "service-", "s" * 44, "s" * 46, "s" * 48])
+def test_plan_refuses_incompatible_exact_service_before_emitting_plan(name):
+    env = {**os.environ, "PLATFORM_CONFIG_JSON": '{"strategies":{}}',
+           "CLOUD_RUN_SERVICE_TARGETS_JSON": json.dumps([{
+               "service": name, "NOTIFY_LANG": "en", "runtime_target": {
+                   "platform_id": "firstrade", "strategy_profile": "tqqq_growth_income",
+                   "dry_run_only": True, "service_name": name}}])}
+    result = subprocess.run([sys.executable, str(ROOT / "scripts/build_cloud_run_env_sync_plan.py"), "--json"],
+                            env=env, text=True, capture_output=True)
+    assert result.returncode == 1
+    assert result.stdout == ""
+    assert result.stderr.strip() in {"INVALID_SERVICE_NAME", "diagnostic_tag_name_budget_exceeded"}
+    assert name not in result.stderr
+
+
+def test_cloudbuild_preflight_precedes_build_and_deploy():
+    text = (ROOT / "cloudbuild.yaml").read_text()
+    assert text.index("id: ServicePreflight") < text.index("id: Buildpack") < text.index("id: Deploy")
+    block = text.split("id: ServicePreflight", 1)[1].split("  - id:", 1)[0]
+    assert "scripts/cloud_run_tag_preflight.py" in block
+    assert "--service-name=$_SERVICE_NAME" in block
+    assert "--diagnostic" not in block  # Image-only caller must not allocate a traffic tag.
+
+
+@pytest.mark.parametrize(("length", "ok"), [(43, True), (44, False), (45, False), (46, False), (48, False)])
+def test_fixed_diagnostic_never_shortens_or_enables_traffic(length, ok):
+    name = "s" * length
+    if ok:
+        stage._validate_diagnostic_tag_budget(name, "cbd")
+    else:
+        with pytest.raises(ValueError, match="diagnostic_tag_name_budget_exceeded"):
+            stage._validate_diagnostic_tag_budget(name, "cbd")
+
+
+@pytest.mark.parametrize("mode", ["legacy", "runtime_identity", "runtime_fallback"])
+def test_plan_never_strips_an_exact_service_identity(mode):
+    name = " service-placeholder "
+    target = {"platform_id": "firstrade", "strategy_profile": "tqqq_growth_income",
+              "dry_run_only": True, "service_name": name}
+    env = {**os.environ, "PLATFORM_CONFIG_JSON": '{"strategies":{}}',
+           "NOTIFY_LANG": "en", "CLOUD_RUN_SERVICE_TARGETS_JSON": "",
+           "CLOUD_RUN_SERVICES": ""}
+    if mode == "legacy":
+        env.update(CLOUD_RUN_SERVICE=name, RUNTIME_TARGET_JSON=json.dumps(target))
+    else:
+        row = {"NOTIFY_LANG": "en", "runtime_target": target}
+        if mode == "runtime_identity":
+            row["service"] = name.strip()
+        env["CLOUD_RUN_SERVICE_TARGETS_JSON"] = json.dumps([row])
+    result = subprocess.run([sys.executable, str(ROOT / "scripts/build_cloud_run_env_sync_plan.py"), "--json"],
+                            env=env, text=True, capture_output=True)
+    assert result.returncode == 1
+    assert result.stdout == ""
+    assert result.stderr == "INVALID_SERVICE_NAME\n"
+
+
+@pytest.mark.parametrize(("name", "ok"), [("s" * 48, True), ("s" * 63, True),
+                                         ("s" * 64, False), ("bad/identity", False)])
+def test_execute_cloudbuild_preflight_arguments_without_cloud_calls(name, ok):
+    text = (ROOT / "cloudbuild.yaml").read_text()
+    block = text.split("id: ServicePreflight", 1)[1].split("  - id:", 1)[0]
+    args = [line.strip()[2:] for line in block.split("    args:", 1)[1].splitlines()
+            if line.strip().startswith("- ")]
+    args = [arg.replace("$_SERVICE_NAME", name) for arg in args]
+    result = subprocess.run([sys.executable, *args], cwd=ROOT, text=True, capture_output=True)
+    assert result.returncode == (0 if ok else 1)
+    assert result.stdout == ("cloud_run_service_preflight_ok\n" if ok else "")
+    assert result.stderr == ("" if ok else "INVALID_SERVICE_NAME\n")
+
+
+def test_plan_retains_exact_service_at_diagnostic_budget_boundary():
+    from scripts.build_cloud_run_env_sync_plan import build_sync_plan
+
+    name = "s" * 43
+    plan = build_sync_plan({"PLATFORM_CONFIG_JSON": '{"strategies":{}}',
+                           "CLOUD_RUN_SERVICE": name, "NOTIFY_LANG": "en",
+                           "RUNTIME_TARGET_JSON": json.dumps({"service_name": name,
+                               "platform_id": "firstrade", "strategy_profile": "tqqq_growth_income",
+                               "dry_run_only": True})})
+    assert plan["targets"][0]["service_name"] == name
+    assert json.loads(plan["targets"][0]["env"]["RUNTIME_TARGET_JSON"])["service_name"] == name