Bound diagnostic tags and verify the real diagnostic CLI entry

.github/workflows/diagnose-cached-stage-failure.yml

@@ -93,7 +93,7 @@ jobs:
             echo "The checked-out source and current main ref must match expected_sha." >&2
             exit 1
           fi
-          python3 scripts/diagnose_cached_stage_failure.py \
+          python3 -m scripts.diagnose_cached_stage_failure \
             --validate-window "${FAILURE_WINDOW_START}" "${FAILURE_WINDOW_END}"
 
       - name: Authenticate with the existing deployment WIF identity
@@ -149,7 +149,7 @@ jobs:
             fi
           fi
 
-          python3 scripts/diagnose_cached_stage_failure.py \
+          python3 -m scripts.diagnose_cached_stage_failure \
             --service "${service_file}" --revisions "${revisions_file}" \
             --expected-service "${EXPECTED_SERVICE}" --iam-policy "${iam_file}" \
             --expected-project "${GCP_PROJECT_ID}" --expected-region "${CLOUD_RUN_REGION}" \

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

@@ -69,7 +69,7 @@ jobs:
       GCP_ARTIFACT_REGISTRY_REPOSITORY: cloud-run-source-deploy
       CACHED_DIAGNOSTIC_SOURCE_SHA: 3317c0282ca5e70a55b97084572e5013a8eeae3f
       EXPECTED_SERVING_SOURCE_SHA: e0043ca860a36c1790ddbb866cb848e298a3d3c7
-      CACHED_DIAGNOSTIC_TAG: cached-balance-diagnostic
+      CACHED_DIAGNOSTIC_TAG: cb
     steps:
       - name: Validate isolated stage inputs
         env:
@@ -123,6 +123,14 @@ jobs:
             exit 1
           fi
 
+      - name: Validate fixed diagnostic traffic tag budget
+        env:
+          EXPECTED_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }}
+        run: |
+          set -euo pipefail
+          python3 scripts/verify_cached_diagnostic_stage.py tag-budget \
+            --expected-service "${EXPECTED_SERVICE}" --tag "${CACHED_DIAGNOSTIC_TAG}"
+
       - name: Checkout fixed cached-diagnostic candidate
         uses: actions/checkout@v6
         with:

scripts/diagnose_cached_stage_failure.py

@@ -10,6 +10,11 @@
 from pathlib import Path
 from typing import Any
 
+from scripts.verify_cached_diagnostic_stage import (
+    DIAGNOSTIC_TAG,
+    MAX_SERVICE_AND_TRAFFIC_TAG_LENGTH,
+)
+
 REASONS = (
     "permission",
     "act_as",
@@ -35,6 +40,10 @@
     "startup": re.compile(r"startup probe|container failed to start|startup[^\n]*failed", re.I),
     "permission": re.compile(r"permission denied|permission_denied|not authorized|forbidden|\b403\b", re.I),
 }
+_COMBINED_TAG_NAME_LIMIT = re.compile(
+    r"traffic[^\n]*tag[^\n]*too long|combined traffic tag and service name cannot exceed 46",
+    re.I,
+)
 
 
 def _read_json(path: str | None) -> Any:
@@ -67,6 +76,10 @@ def _classify(texts: list[str]) -> list[str]:
     return [reason for reason in REASONS if _REASON_PATTERNS[reason].search(joined)] or ["unknown"]
 
 
+def _has_combined_tag_name_limit(texts: list[str]) -> bool:
+    return bool(_COMBINED_TAG_NAME_LIMIT.search("\n".join(texts)))
+
+
 def _revision_for_service(service: Any, revisions: Any) -> dict[str, Any] | None:
     if not isinstance(service, dict) or not isinstance(revisions, list):
         return None
@@ -277,16 +290,30 @@ def summarize(
     if audit_error_count and audit_status == "ok":
         failure_source = "audit"
         failure_categories = audit_categories
+        failure_texts = audit_texts
     elif latest_ready is False:
         failure_source = "revision"
         failure_categories = revision_categories
+        failure_texts = revision_texts
     else:
         failure_source = "none"
         failure_categories = ["unknown"]
+        failure_texts = []
+    combined_tag_name_error = _has_combined_tag_name_limit(failure_texts)
+    target_matches = bool(isinstance(metadata, dict) and metadata.get("name") == expected_service)
+    tag_budget_ok = bool(
+        target_matches
+        and isinstance(metadata, dict)
+        and isinstance(metadata.get("name"), str)
+        and len(metadata["name"]) + len(DIAGNOSTIC_TAG) <= MAX_SERVICE_AND_TRAFFIC_TAG_LENGTH
+    )
     result = {
         "schema_version": "firstrade_cached_stage_diagnostic.v1",
         "service_readable": service_ok,
-        "target_matches": bool(isinstance(metadata, dict) and metadata.get("name") == expected_service),
+        "target_matches": target_matches,
+        "diagnostic_tag_budget_ok": tag_budget_ok,
+        "failure_subcategory": "combined_traffic_tag_service_name_length" if combined_tag_name_error else "none",
+        "combined_traffic_tag_service_name_length_error_observed": combined_tag_name_error,
         "service_ready": _ready(service),
         "traffic_row_count": len(traffic_rows) if traffic_rows is not None else None,
         "positive_traffic_row_count": len(positive),

scripts/verify_cached_diagnostic_stage.py

@@ -13,7 +13,8 @@
 
 DIAGNOSTIC_GATE = "FIRSTRADE_CACHED_BALANCE_DIAGNOSTIC_ON_HTTP"
 RUNTIME_TARGET_KEYS = ("QSL_RUNTIME_TARGET_JSON", "RUNTIME_TARGET_JSON")
-DIAGNOSTIC_TAG = "cached-balance-diagnostic"
+DIAGNOSTIC_TAG = "cb"
+MAX_SERVICE_AND_TRAFFIC_TAG_LENGTH = 46
 EXPECTED_PLATFORM_ID = "firstrade"
 GENERATED_TEMPLATE_ANNOTATIONS = {
     "run.googleapis.com/client-name",
@@ -60,6 +61,15 @@ def _active_traffic(service: dict[str, Any]) -> tuple[list[dict[str, Any]], list
     return active_traffic, traffic_rows
 
 
+def _validate_diagnostic_tag_budget(service_name: str, tag: str) -> None:
+    if not isinstance(service_name, str) or not service_name:
+        raise ValueError("diagnostic_service_name_missing")
+    if tag != DIAGNOSTIC_TAG:
+        raise ValueError("diagnostic_tag_mismatch")
+    if len(service_name) + len(tag) > MAX_SERVICE_AND_TRAFFIC_TAG_LENGTH:
+        raise ValueError("diagnostic_tag_name_budget_exceeded")
+
+
 def _normalize_template_metadata(template: dict[str, Any]) -> None:
     metadata = template.get("metadata")
     if not isinstance(metadata, dict):
@@ -685,19 +695,26 @@ def verify_readback(state_path: Path, expected_image: str) -> None:
 
 def main() -> int:
     parser = argparse.ArgumentParser()
-    parser.add_argument("phase", choices=("active-revision", "capture", "verify", "scheduler-hash"))
+    parser.add_argument(
+        "phase", choices=("active-revision", "capture", "verify", "scheduler-hash", "tag-budget")
+    )
     parser.add_argument("--state", type=Path)
     parser.add_argument("--revision", type=Path)
     parser.add_argument("--expected-service")
     parser.add_argument("--expected-source-sha")
     parser.add_argument("--expected-image")
+    parser.add_argument("--tag")
     args = parser.parse_args()
     try:
         if args.phase == "active-revision":
             service = _read_service()
             active, _ = _active_traffic(service)
             print(active[0]["revisionName"])
             return 0
+        if args.phase == "tag-budget":
+            _validate_diagnostic_tag_budget(args.expected_service or "", args.tag or "")
+            print("diagnostic_tag_budget_ok")
+            return 0
         if args.phase == "scheduler-hash":
             jobs = json.load(sys.stdin)
             print(_scheduler_jobs_hash(jobs))

tests/test_cached_diagnostic_stage.py

@@ -33,6 +33,49 @@ def test_container_difference_categories_keep_private_details_closed():
     ]
 
 
+@pytest.mark.parametrize(
+    ("service_name_length", "error"),
+    [(44, None), (45, "diagnostic_tag_name_budget_exceeded")],
+)
+def test_diagnostic_traffic_tag_budget_is_checked_without_exposing_name(service_name_length, error):
+    service_name = "s" * service_name_length
+    if error:
+        with pytest.raises(ValueError, match=error):
+            stage._validate_diagnostic_tag_budget(service_name, stage.DIAGNOSTIC_TAG)
+    else:
+        stage._validate_diagnostic_tag_budget(service_name, stage.DIAGNOSTIC_TAG)
+
+
+def test_diagnostic_traffic_tag_budget_rejects_a_different_tag():
+    with pytest.raises(ValueError, match="diagnostic_tag_mismatch"):
+        stage._validate_diagnostic_tag_budget("service-placeholder", "long-tag")
+
+
+def test_tag_budget_cli_reports_only_closed_result(monkeypatch, capsys):
+    monkeypatch.setattr(
+        stage.sys,
+        "argv",
+        ["verify_cached_diagnostic_stage.py", "tag-budget", "--expected-service", "private-service", "--tag", "cb"],
+    )
+    assert stage.main() == 0
+    output = capsys.readouterr()
+    assert output.out == "diagnostic_tag_budget_ok\n"
+    assert "private-service" not in output.out + output.err
+
+
+def test_tag_budget_cli_rejects_long_name_without_echoing_it(monkeypatch, capsys):
+    service_name = "s" * 45
+    monkeypatch.setattr(
+        stage.sys,
+        "argv",
+        ["verify_cached_diagnostic_stage.py", "tag-budget", "--expected-service", service_name, "--tag", "cb"],
+    )
+    assert stage.main() == 1
+    output = capsys.readouterr()
+    assert output.err == "cached_diagnostic_stage_blocked:diagnostic_tag_name_budget_exceeded\n"
+    assert service_name not in output.out + output.err
+
+
 def test_primary_name_shape_classifies_names_images_counts_and_dependencies():
     serving = {"containers": [
         {"name": "hidden-serving-app-1", "image": "registry.invalid/team/hidden-serving-app:stable"},

tests/test_cached_stage_failure_diagnostic.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
 import json
+import os
+import subprocess
+import sys
 from datetime import UTC, datetime, timedelta
 from pathlib import Path
 
@@ -85,6 +88,9 @@ def test_summary_exposes_only_closed_statuses_counts_and_categories():
         "schema_version": "firstrade_cached_stage_diagnostic.v1",
         "service_readable": True,
         "target_matches": True,
+        "diagnostic_tag_budget_ok": True,
+        "failure_subcategory": "none",
+        "combined_traffic_tag_service_name_length_error_observed": False,
         "service_ready": True,
         "traffic_row_count": 1,
         "positive_traffic_row_count": 1,
@@ -153,6 +159,53 @@ def test_missing_logging_permission_is_reported_without_guessing_failure_categor
     assert summary["failure_categories"] == ["permission"]
 
 
+def test_combined_traffic_tag_length_is_closed_subcategory_and_budget_stays_private():
+    message = (
+        "traffic[].tag: traffic tag [TAG] and service name [SERVICE] together are too long. "
+        "Combined traffic tag and service name cannot exceed 46 characters."
+    )
+    service, revisions, policy, jobs, audit = _evidence(message)
+    summary = diagnostic.summarize(
+        service=service,
+        revisions=revisions,
+        expected_service=PRIVATE_SERVICE,
+        expected_project=PRIVATE_PROJECT,
+        expected_region=PRIVATE_REGION,
+        audit_window_start=WINDOW_START,
+        audit_window_end=WINDOW_END,
+        policy=policy,
+        jobs=jobs,
+        audit_entries=audit,
+        audit_status="ok",
+    )
+    assert summary["failure_categories"] == ["invalid_name"]
+    assert summary["failure_subcategory"] == "combined_traffic_tag_service_name_length"
+    assert summary["combined_traffic_tag_service_name_length_error_observed"] is True
+    assert summary["diagnostic_tag_budget_ok"] is True
+    assert PRIVATE_SERVICE not in json.dumps(summary)
+
+
+def test_tag_budget_false_is_reported_as_boolean_without_service_name():
+    service, revisions, policy, jobs, audit = _evidence()
+    long_name = "s" * 45
+    service["metadata"]["name"] = long_name
+    summary = diagnostic.summarize(
+        service=service,
+        revisions=revisions,
+        expected_service=long_name,
+        expected_project=PRIVATE_PROJECT,
+        expected_region=PRIVATE_REGION,
+        audit_window_start=WINDOW_START,
+        audit_window_end=WINDOW_END,
+        policy=policy,
+        jobs=jobs,
+        audit_entries=audit,
+        audit_status="ok",
+    )
+    assert summary["diagnostic_tag_budget_ok"] is False
+    assert long_name not in json.dumps(summary)
+
+
 def test_unreadable_metadata_stays_unknown_and_never_serializes_input():
     summary = diagnostic.summarize(
         service={"private": "private-value-placeholder"},
@@ -244,3 +297,87 @@ def test_workflow_is_opt_in_read_only_and_reuses_existing_identity():
     assert "docker build" not in workflow
     assert "docker push" not in workflow
     assert "actions/upload-artifact" not in workflow
+    assert workflow.count("python3 -m scripts.diagnose_cached_stage_failure") == 2
+    assert "python3 scripts/diagnose_cached_stage_failure.py" not in workflow
+
+
+def test_module_cli_runs_from_repository_root_without_pythonpath(tmp_path):
+    repo_root = Path(__file__).resolve().parents[1]
+    service, revisions, policy, jobs, audit = _evidence()
+    now = datetime.now(UTC).replace(microsecond=0)
+    audit[0]["timestamp"] = (now - timedelta(minutes=5)).isoformat().replace("+00:00", "Z")
+    inputs = {
+        "service": service,
+        "revisions": revisions,
+        "iam-policy": policy,
+        "scheduler-jobs": jobs,
+        "audit-entries": audit,
+    }
+    paths = {}
+    for name, value in inputs.items():
+        path = tmp_path / f"{name}.json"
+        path.write_text(json.dumps(value), encoding="utf-8")
+        paths[name] = str(path)
+
+    window_end = now.isoformat().replace("+00:00", "Z")
+    window_start = (now - timedelta(minutes=10)).isoformat().replace("+00:00", "Z")
+    env = {key: value for key, value in os.environ.items() if key != "PYTHONPATH"}
+
+    validated = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "scripts.diagnose_cached_stage_failure",
+            "--validate-window",
+            window_start,
+            window_end,
+        ],
+        cwd=repo_root,
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert validated.returncode == 0, validated.stderr
+    assert validated.stdout == ""
+
+    summarized = subprocess.run(
+        [
+            sys.executable,
+            "-m",
+            "scripts.diagnose_cached_stage_failure",
+            "--service",
+            paths["service"],
+            "--revisions",
+            paths["revisions"],
+            "--expected-service",
+            PRIVATE_SERVICE,
+            "--expected-project",
+            PRIVATE_PROJECT,
+            "--expected-region",
+            PRIVATE_REGION,
+            "--window-start",
+            window_start,
+            "--window-end",
+            window_end,
+            "--iam-policy",
+            paths["iam-policy"],
+            "--scheduler-jobs",
+            paths["scheduler-jobs"],
+            "--audit-entries",
+            paths["audit-entries"],
+            "--audit-status",
+            "ok",
+        ],
+        cwd=repo_root,
+        env=env,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    assert summarized.returncode == 0, summarized.stderr
+    summary = json.loads(summarized.stdout)
+    assert summary["failure_source"] == "audit"
+    assert summary["failure_categories"] == ["image"]
+    assert PRIVATE_SERVICE not in summarized.stdout
+    assert PRIVATE_MESSAGE not in summarized.stdout

tests/test_sync_cloud_run_env_workflow.py

@@ -259,6 +259,11 @@ def test_cached_balance_diagnostic_stage_is_opt_in_and_separate_from_deploy_and_
     assert "verify_cached_diagnostic_stage.py active-revision" in stage_job
     assert "verify_cached_diagnostic_stage.py scheduler-hash" in stage_job
     assert "--no-traffic --tag=\"${CACHED_DIAGNOSTIC_TAG}\"" in stage_job
+    assert "CACHED_DIAGNOSTIC_TAG: cb" in stage_job
+    budget_check = stage_job.index("Validate fixed diagnostic traffic tag budget")
+    image_build = stage_job.index("Build and push the fixed diagnostic image")
+    assert budget_check < image_build
+    assert "verify_cached_diagnostic_stage.py tag-budget" in stage_job
     assert "--ingress=" not in stage_job
     assert "--service-account=" not in stage_job
     assert "--set-env-vars" not in stage_job