fix: scope runtime guard service log queries (#280)

- Prefer explicit per-environment Cloud Run services over the global target list
- Add region filters to Cloud Run log queries
- Cover explicit service and region filter behavior

Co-Authored-By: Codex <noreply@openai.com>

Author: Pigbibi

scripts/cloud_run_runtime_guard.py

@@ -46,6 +46,8 @@ def _load_services() -> list[str]:
         "CLOUD_RUN_SERVICE",
     ):
         services.extend(_split_values(os.environ.get(name)))
+    if services:
+        return list(dict.fromkeys(services))
 
     raw_targets = (os.environ.get("CLOUD_RUN_SERVICE_TARGETS_JSON") or "").strip()
     if raw_targets:
@@ -85,6 +87,17 @@ def _load_services() -> list[str]:
     return unique
 
 
+def _cloud_run_log_filter(service: str, since_text: str, region: str = "") -> str:
+    parts = [
+        'resource.type="cloud_run_revision"',
+        f'resource.labels.service_name="{service}"',
+    ]
+    if region:
+        parts.append(f'resource.labels.location="{region}"')
+    parts.append(f'timestamp >= "{since_text}"')
+    return " AND ".join(parts)
+
+
 def _scheduler_job_pattern_for_services(services: list[str]) -> str:
     candidates: list[str] = []
     for service in services:
@@ -442,11 +455,7 @@ def main() -> int:
     for service in services:
         service_since = _cloud_run_log_since(project, service, since) if ignore_pre_ready_logs else since
         service_since_text = _format_timestamp(service_since)
-        log_filter = (
-            'resource.type="cloud_run_revision" '
-            f'AND resource.labels.service_name="{service}" '
-            f'AND timestamp >= "{service_since_text}"'
-        )
+        log_filter = _cloud_run_log_filter(service, service_since_text, _region_for_service(service))
         try:
             entries = _run_gcloud_logging(project, log_filter, limit)
         except RuntimeError as exc:

tests/test_cloud_run_runtime_guard.py

@@ -8,6 +8,67 @@
 from scripts import cloud_run_runtime_guard as guard
 
 
+def _clear_runtime_guard_env(monkeypatch):
+    for name in (
+        "RUNTIME_GUARD_CLOUD_RUN_SERVICES",
+        "CLOUD_RUN_SERVICES",
+        "CLOUD_RUN_SERVICE",
+        "CLOUD_RUN_SERVICE_TARGETS_JSON",
+        "CLOUD_RUN_REGION",
+    ):
+        monkeypatch.delenv(name, raising=False)
+
+
+def test_load_services_prefers_explicit_service_over_target_list(monkeypatch):
+    _clear_runtime_guard_env(monkeypatch)
+    monkeypatch.setenv("CLOUD_RUN_SERVICE", "longbridge-quant-hk-service")
+    monkeypatch.setenv(
+        "CLOUD_RUN_SERVICE_TARGETS_JSON",
+        json.dumps(
+            {
+                "targets": [
+                    {"service": "longbridge-quant-paper-service"},
+                    {"service": "longbridge-quant-sg-service"},
+                ]
+            }
+        ),
+    )
+
+    assert guard._load_services() == ["longbridge-quant-hk-service"]
+
+
+def test_load_services_falls_back_to_target_list(monkeypatch):
+    _clear_runtime_guard_env(monkeypatch)
+    monkeypatch.setenv(
+        "CLOUD_RUN_SERVICE_TARGETS_JSON",
+        json.dumps(
+            {
+                "targets": [
+                    {"service": "longbridge-quant-paper-service"},
+                    {"runtime_target": {"service_name": "longbridge-quant-sg-service"}},
+                ]
+            }
+        ),
+    )
+
+    assert guard._load_services() == [
+        "longbridge-quant-paper-service",
+        "longbridge-quant-sg-service",
+    ]
+
+
+def test_cloud_run_log_filter_includes_region_when_available():
+    log_filter = guard._cloud_run_log_filter(
+        "longbridge-quant-paper-service",
+        "2026-07-01T12:00:00Z",
+        "asia-east1",
+    )
+
+    assert 'resource.labels.service_name="longbridge-quant-paper-service"' in log_filter
+    assert 'resource.labels.location="asia-east1"' in log_filter
+    assert 'timestamp >= "2026-07-01T12:00:00Z"' in log_filter
+
+
 def test_scheduler_job_pattern_includes_service_alias():
     pattern = guard._scheduler_job_pattern_for_services(["longbridge-quant-hk-service"])