fix: harden Firstrade runtime monitoring and deployment controls

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

Author: Pigbibi

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

@@ -1,10 +1,6 @@
 name: Deploy Cloud Run
 
 on:
-  workflow_run:
-    workflows: [CI]
-    types: [completed]
-    branches: [main]
   workflow_dispatch:
 
 permissions:
@@ -27,15 +23,8 @@ concurrency:
 jobs:
   deploy-cloud-run:
     name: Deploy Cloud Run
-    if: >
-      github.event_name == 'workflow_dispatch' ||
-      (
-        github.event_name == 'workflow_run' &&
-        github.event.workflow_run.conclusion == 'success' &&
-        github.event.workflow_run.event == 'push' &&
-        !contains(github.event.workflow_run.head_commit.message, 'chore(deps): align QPK pin') &&
-        !contains(github.event.workflow_run.head_commit.message, '[skip-cloud-run-deploy]')
-      )
+    # Production deployment and environment sync require an explicit operator dispatch.
+    if: github.event_name == 'workflow_dispatch'
     runs-on: ubuntu-latest
     timeout-minutes: 20
     permissions:
@@ -190,7 +179,7 @@ jobs:
         if: steps.deploy_config.outputs.enabled == 'true'
         uses: actions/checkout@v6
         with:
-          ref: ${{ github.event_name == 'workflow_run' && github.event.workflow_run.head_sha || github.sha }}
+          ref: ${{ github.sha }}
 
       - name: Validate deploy inputs
         if: steps.deploy_config.outputs.enabled == 'true'

scripts/cloud_run_runtime_guard.py

@@ -9,6 +9,7 @@
 import re
 import subprocess
 import sys
+import time
 import urllib.parse
 import urllib.request
 from typing import Any
@@ -24,6 +25,26 @@
     "URL_UNREACHABLE",
 )
 SCHEDULER_CLOUD_RUN_DEDUP_SECONDS = 120
+DEFAULT_LOG_QUERY_MAX_ATTEMPTS = 3
+DEFAULT_LOG_QUERY_RETRY_SECONDS = 1.0
+_RETRYABLE_LOG_QUERY_MARKERS = (
+    "http 429",
+    "http 500",
+    "http 502",
+    "http 503",
+    "http 504",
+    '"code": 429',
+    '"code": 500',
+    '"code": 502',
+    '"code": 503',
+    '"code": 504',
+    "internal error",
+    "unavailable",
+    "timed out",
+    "network connectivity",
+    "connection reset",
+    "rate limit",
+)
 
 
 def _split_values(raw: str | None) -> list[str]:
@@ -200,6 +221,31 @@ def _run_gcloud_json(args: list[str], context: str) -> Any:
         raise RuntimeError(f"gcloud {context} returned invalid JSON: {exc}") from exc
 
 
+def _log_query_retry_config() -> tuple[int, float]:
+    try:
+        attempts = int(
+            os.environ.get("RUNTIME_GUARD_LOG_QUERY_MAX_ATTEMPTS")
+            or DEFAULT_LOG_QUERY_MAX_ATTEMPTS
+        )
+    except ValueError:
+        attempts = DEFAULT_LOG_QUERY_MAX_ATTEMPTS
+    try:
+        retry_seconds = float(
+            os.environ.get("RUNTIME_GUARD_LOG_QUERY_RETRY_SECONDS")
+            or DEFAULT_LOG_QUERY_RETRY_SECONDS
+        )
+    except ValueError:
+        retry_seconds = DEFAULT_LOG_QUERY_RETRY_SECONDS
+    return max(1, min(attempts, 5)), max(0.0, min(retry_seconds, 10.0))
+
+
+def _is_retryable_log_query_error(detail: str) -> bool:
+    normalized = detail.lower()
+    if "403" in normalized or "permission_denied" in normalized:
+        return False
+    return any(marker in normalized for marker in _RETRYABLE_LOG_QUERY_MARKERS)
+
+
 def _run_gcloud_logging(project: str, log_filter: str, limit: int) -> list[dict[str, Any]]:
     command = [
         "gcloud",
@@ -211,17 +257,24 @@ def _run_gcloud_logging(project: str, log_filter: str, limit: int) -> list[dict[
         "--format=json",
         f"--limit={limit}",
     ]
-    result = subprocess.run(command, text=True, capture_output=True, check=False)
-    if result.returncode != 0:
-        detail = (result.stderr or result.stdout or "").strip()
-        raise RuntimeError(detail or "gcloud logging read failed")
-    if not result.stdout.strip():
-        return []
-    try:
-        payload = json.loads(result.stdout)
-    except json.JSONDecodeError as exc:
-        raise RuntimeError(f"gcloud returned invalid JSON: {exc}") from exc
-    return payload if isinstance(payload, list) else []
+    max_attempts, retry_seconds = _log_query_retry_config()
+    last_detail = ""
+    for attempt in range(1, max_attempts + 1):
+        result = _run_gcloud(command)
+        if result.returncode == 0:
+            if not result.stdout.strip():
+                return []
+            try:
+                payload = json.loads(result.stdout)
+            except json.JSONDecodeError as exc:
+                raise RuntimeError(f"gcloud returned invalid JSON: {exc}") from exc
+            return payload if isinstance(payload, list) else []
+        last_detail = (result.stderr or result.stdout or "").strip()
+        if attempt >= max_attempts or not _is_retryable_log_query_error(last_detail):
+            break
+        time.sleep(retry_seconds * attempt)
+    suffix = f" after {max_attempts} attempt(s)" if max_attempts > 1 else ""
+    raise RuntimeError((last_detail or "gcloud logging read failed") + suffix)
 
 
 def _parse_timestamp(value: Any) -> dt.datetime | None:

tests/test_cloud_run_runtime_guard.py

@@ -55,6 +55,54 @@ def fake_run_gcloud(command):
     ]
 
 
+def test_cloud_run_log_query_retries_transient_google_error(monkeypatch):
+    attempts = []
+    sleeps = []
+
+    def fake_run_gcloud(command):
+        attempts.append(command)
+        if len(attempts) == 1:
+            return subprocess.CompletedProcess(
+                command,
+                1,
+                stdout="",
+                stderr='HttpError: {"error": {"code": 500, "status": "INTERNAL"}}',
+            )
+        return subprocess.CompletedProcess(command, 0, stdout="[]", stderr="")
+
+    monkeypatch.setenv("RUNTIME_GUARD_LOG_QUERY_MAX_ATTEMPTS", "3")
+    monkeypatch.setenv("RUNTIME_GUARD_LOG_QUERY_RETRY_SECONDS", "0")
+    monkeypatch.setattr(guard, "_run_gcloud", fake_run_gcloud)
+    monkeypatch.setattr(guard.time, "sleep", lambda seconds: sleeps.append(seconds))
+
+    assert guard._run_gcloud_logging("project-1", 'resource.type="cloud_run_revision"', 10) == []
+    assert len(attempts) == 2
+    assert sleeps == [0.0]
+
+
+def test_cloud_run_log_query_does_not_retry_permission_error(monkeypatch):
+    attempts = []
+
+    def fake_run_gcloud(command):
+        attempts.append(command)
+        return subprocess.CompletedProcess(
+            command,
+            1,
+            stdout="",
+            stderr="ERROR: permission_denied (403)",
+        )
+
+    monkeypatch.setattr(guard, "_run_gcloud", fake_run_gcloud)
+
+    try:
+        guard._run_gcloud_logging("project-1", 'resource.type="cloud_run_revision"', 10)
+    except RuntimeError as exc:
+        assert "permission_denied" in str(exc)
+    else:
+        raise AssertionError("permission errors must fail without retry")
+    assert len(attempts) == 1
+
+
 def test_cloud_run_log_since_uses_latest_ready_revision(monkeypatch):
     monkeypatch.setenv("CLOUD_RUN_REGION", "us-central1")
     observed = []

tests/test_sync_cloud_run_env_workflow.py

@@ -3,6 +3,14 @@
 from pathlib import Path
 
 
+def test_sync_cloud_run_env_workflow_requires_manual_dispatch():
+    workflow_path = Path(__file__).resolve().parents[1] / ".github/workflows/sync-cloud-run-env.yml"
+    workflow = workflow_path.read_text(encoding="utf-8")
+
+    assert "workflow_run:" not in workflow
+    assert "if: github.event_name == 'workflow_dispatch'" in workflow
+
+
 def test_sync_cloud_run_env_workflow_uses_sync_plan_script():
     workflow_path = Path(__file__).resolve().parents[1] / ".github/workflows/sync-cloud-run-env.yml"
     workflow = workflow_path.read_text(encoding="utf-8")