From 93098717ef1d3475aa2d6ae7d556e1510772e117 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Fri, 2 Oct 2026 20:35:42 +0800 Subject: [PATCH] Diagnose failed cached balance staging with bounded read-only evidence Co-Authored-By: Codex --- .../diagnose-cached-stage-failure.yml | 158 ++++++++ scripts/diagnose_cached_stage_failure.py | 379 ++++++++++++++++++ tests/test_cached_stage_failure_diagnostic.py | 246 ++++++++++++ 3 files changed, 783 insertions(+) create mode 100644 .github/workflows/diagnose-cached-stage-failure.yml create mode 100644 scripts/diagnose_cached_stage_failure.py create mode 100644 tests/test_cached_stage_failure_diagnostic.py diff --git a/.github/workflows/diagnose-cached-stage-failure.yml b/.github/workflows/diagnose-cached-stage-failure.yml new file mode 100644 index 0000000..773c56b --- /dev/null +++ b/.github/workflows/diagnose-cached-stage-failure.yml @@ -0,0 +1,158 @@ +name: Diagnose cached-balance stage failure + +on: + workflow_dispatch: + inputs: + expected_sha: + description: Full merged main SHA containing this read-only diagnostic + required: true + type: string + approved_ref: + description: Must be refs/heads/main + required: true + type: string + run_readonly_diagnostic: + description: Read current Cloud Run/control metadata and bounded failure logs + required: true + default: false + type: boolean + failure_window_start: + description: UTC RFC3339 start time for the failed stage deployment + required: true + type: string + failure_window_end: + description: UTC RFC3339 end time for the failed stage deployment (maximum 30 minutes after start) + required: true + type: string + +permissions: + contents: read + +env: + GCP_PROJECT_ID: firstradequant + GCP_WORKLOAD_IDENTITY_PROVIDER: projects/1088907247379/locations/global/workloadIdentityPools/github-actions/providers/github-main + GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT: firstrade-platform-deploy@firstradequant.iam.gserviceaccount.com + +# Share the existing deployment lock while this bounded read-only inspection runs. +concurrency: + group: Deploy Cloud Run-${{ github.ref_name }} + cancel-in-progress: false + +jobs: + diagnose: + name: Read-only failed-stage diagnosis + if: github.event_name == 'workflow_dispatch' && inputs.run_readonly_diagnostic == true + runs-on: ubuntu-latest + timeout-minutes: 10 + permissions: + contents: read + id-token: write + env: + CLOUD_RUN_REGION: ${{ vars.CLOUD_RUN_REGION }} + CLOUD_RUN_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }} + SCHEDULER_LOCATION: ${{ vars.CLOUD_SCHEDULER_LOCATION || vars.CLOUD_RUN_REGION }} + FAILURE_WINDOW_START: ${{ inputs.failure_window_start }} + FAILURE_WINDOW_END: ${{ inputs.failure_window_end }} + steps: + - name: Validate fixed target, source, and bounded time window + env: + EXPECTED_SHA: ${{ inputs.expected_sha }} + APPROVED_REF: ${{ inputs.approved_ref }} + DISPATCH_SHA: ${{ github.sha }} + run: | + set -euo pipefail + if [ "${APPROVED_REF}" != "refs/heads/main" ] \ + || [ "${DISPATCH_SHA}" != "${EXPECTED_SHA}" ] \ + || ! [[ "${EXPECTED_SHA}" =~ ^[0-9a-f]{40}$ ]]; then + echo "Diagnostic source must be the exact approved main commit." >&2 + exit 1 + fi + if [ -z "${CLOUD_RUN_REGION:-}" ] || [ -z "${CLOUD_RUN_SERVICE:-}" ] \ + || [ -z "${SCHEDULER_LOCATION:-}" ]; then + echo "The fixed service or Scheduler target is unavailable." >&2 + exit 1 + fi + if ! [[ "${CLOUD_RUN_SERVICE}" =~ ^[a-z]([-a-z0-9]{0,61}[a-z0-9])?$ ]]; then + echo "The configured Cloud Run service target has an invalid shape." >&2 + exit 1 + fi + - name: Checkout approved main source + uses: actions/checkout@v6 + with: + ref: ${{ inputs.expected_sha }} + + - name: Verify checked-out source and bounded window + env: + EXPECTED_SHA: ${{ inputs.expected_sha }} + APPROVED_REF: ${{ inputs.approved_ref }} + run: | + set -euo pipefail + checked_out_sha="$(git rev-parse HEAD)" + approved_sha="$(git ls-remote --exit-code origin "${APPROVED_REF}" | awk 'NR == 1 { print $1 }')" + if [ "${checked_out_sha}" != "${EXPECTED_SHA}" ] || [ "${approved_sha}" != "${EXPECTED_SHA}" ]; then + echo "The checked-out source and current main ref must match expected_sha." >&2 + exit 1 + fi + python3 scripts/diagnose_cached_stage_failure.py \ + --validate-window "${FAILURE_WINDOW_START}" "${FAILURE_WINDOW_END}" + + - name: Authenticate with the existing deployment WIF identity + uses: google-github-actions/auth@v3 + with: + workload_identity_provider: ${{ env.GCP_WORKLOAD_IDENTITY_PROVIDER }} + service_account: ${{ env.GCP_WORKLOAD_IDENTITY_SERVICE_ACCOUNT }} + + - name: Set up gcloud + uses: google-github-actions/setup-gcloud@v3 + with: + project_id: ${{ env.GCP_PROJECT_ID }} + version: ">= 416.0.0" + + - name: Read private Cloud Run, IAM, Scheduler, and audit evidence + env: + EXPECTED_SERVICE: ${{ secrets.CLOUD_RUN_SERVICE }} + run: | + set -euo pipefail + umask 077 + private_dir="${RUNNER_TEMP}/firstrade-stage-diagnostic" + mkdir -m 700 "${private_dir}" + trap 'rm -f "${private_dir}"/*; rmdir "${private_dir}"' EXIT + + service_file="${private_dir}/service.json" + revisions_file="${private_dir}/revisions.json" + iam_file="${private_dir}/iam.json" + scheduler_file="${private_dir}/scheduler.json" + audit_file="${private_dir}/audit.json" + + gcloud run services describe "${CLOUD_RUN_SERVICE}" \ + --project="${GCP_PROJECT_ID}" --region="${CLOUD_RUN_REGION}" --format=json \ + >"${service_file}" 2>"${private_dir}/service.err" || true + gcloud run revisions list --service="${CLOUD_RUN_SERVICE}" \ + --project="${GCP_PROJECT_ID}" --region="${CLOUD_RUN_REGION}" --limit=20 --format=json \ + >"${revisions_file}" 2>"${private_dir}/revisions.err" || true + gcloud run services get-iam-policy "${CLOUD_RUN_SERVICE}" \ + --project="${GCP_PROJECT_ID}" --region="${CLOUD_RUN_REGION}" --format=json \ + >"${iam_file}" 2>"${private_dir}/iam.err" || true + gcloud scheduler jobs list --project="${GCP_PROJECT_ID}" \ + --location="${SCHEDULER_LOCATION}" --format=json \ + >"${scheduler_file}" 2>"${private_dir}/scheduler.err" || true + + identity_filter="protoPayload.resourceName=\"projects/${GCP_PROJECT_ID}/locations/${CLOUD_RUN_REGION}/services/${CLOUD_RUN_SERVICE}\" OR (protoPayload.resourceName=\"namespaces/${GCP_PROJECT_ID}/services/${CLOUD_RUN_SERVICE}\" AND resource.labels.project_id=\"${GCP_PROJECT_ID}\" AND resource.labels.location=\"${CLOUD_RUN_REGION}\" AND resource.labels.service_name=\"${CLOUD_RUN_SERVICE}\") OR (resource.labels.project_id=\"${GCP_PROJECT_ID}\" AND resource.labels.location=\"${CLOUD_RUN_REGION}\" AND resource.labels.service_name=\"${CLOUD_RUN_SERVICE}\")" + filter="(${identity_filter}) AND timestamp >= \"${FAILURE_WINDOW_START}\" AND timestamp <= \"${FAILURE_WINDOW_END}\" AND protoPayload.serviceName=\"run.googleapis.com\" AND (protoPayload.methodName:\"UpdateService\" OR protoPayload.methodName:\"CreateService\" OR protoPayload.methodName:\"ReplaceService\")" + audit_status=ok + if ! gcloud logging read "${filter}" --project="${GCP_PROJECT_ID}" \ + --limit=100 --format=json >"${audit_file}" 2>"${private_dir}/audit.err"; then + if grep -Eqi 'permission denied|permission_denied|not authorized|forbidden|\b403\b' "${private_dir}/audit.err"; then + audit_status=permission_denied + else + audit_status=unavailable + fi + fi + + python3 scripts/diagnose_cached_stage_failure.py \ + --service "${service_file}" --revisions "${revisions_file}" \ + --expected-service "${EXPECTED_SERVICE}" --iam-policy "${iam_file}" \ + --expected-project "${GCP_PROJECT_ID}" --expected-region "${CLOUD_RUN_REGION}" \ + --window-start "${FAILURE_WINDOW_START}" --window-end "${FAILURE_WINDOW_END}" \ + --scheduler-jobs "${scheduler_file}" --audit-entries "${audit_file}" \ + --audit-status "${audit_status}" diff --git a/scripts/diagnose_cached_stage_failure.py b/scripts/diagnose_cached_stage_failure.py new file mode 100644 index 0000000..29bf0be --- /dev/null +++ b/scripts/diagnose_cached_stage_failure.py @@ -0,0 +1,379 @@ +"""Summarize private Cloud Run staging evidence without printing source values.""" + +from __future__ import annotations + +import argparse +import json +import re +import sys +from datetime import UTC, datetime +from pathlib import Path +from typing import Any + +REASONS = ( + "permission", + "act_as", + "invalid_name", + "image", + "port", + "startup", + "missing_secret", +) +_REASON_PATTERNS = { + "act_as": re.compile(r"iam\.serviceaccounts\.actas|actas|service account user", re.I), + "missing_secret": re.compile(r"secret[^\n]*(not found|does not exist|missing)|secretkeyref", re.I), + "invalid_name": re.compile( + r"invalid[^\n]*name|name[^\n]*invalid|must (start|end) with|" + r"traffic[^\n]*tag[^\n]*too long|combined traffic tag and service name cannot exceed 46", + re.I, + ), + "image": re.compile( + r"image[^\n]*(pull|manifest|digest|not found|resolve)|failed to (resolve|pull|fetch)[^\n]*image", + re.I, + ), + "port": re.compile(r"port[^\n]*(listen|start|bind|set)|listen on the port", re.I), + "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), +} + + +def _read_json(path: str | None) -> Any: + if not path: + return None + try: + return json.loads(Path(path).read_text(encoding="utf-8")) + except (OSError, UnicodeError, json.JSONDecodeError): + return None + + +def _ready(resource: Any) -> bool | None: + status = resource.get("status") if isinstance(resource, dict) else None + conditions = status.get("conditions") if isinstance(status, dict) else None + if not isinstance(conditions, list): + return None + ready = [item for item in conditions if isinstance(item, dict) and item.get("type") == "Ready"] + if len(ready) != 1: + return None + state = ready[0].get("state", ready[0].get("status")) + if state is True or state == "True" or state == "CONDITION_SUCCEEDED": + return True + if state is False or state == "False" or state == "CONDITION_FAILED": + return False + return None + + +def _classify(texts: list[str]) -> list[str]: + joined = "\n".join(texts) + return [reason for reason in REASONS if _REASON_PATTERNS[reason].search(joined)] or ["unknown"] + + +def _revision_for_service(service: Any, revisions: Any) -> dict[str, Any] | None: + if not isinstance(service, dict) or not isinstance(revisions, list): + return None + status = service.get("status") + latest_name = status.get("latestCreatedRevisionName") if isinstance(status, dict) else None + if not isinstance(latest_name, str) or not latest_name: + return None + matches = [ + item + for item in revisions + if isinstance(item, dict) + and isinstance(item.get("metadata"), dict) + and item["metadata"].get("name") == latest_name + ] + return matches[0] if len(matches) == 1 else None + + +def _condition_error_texts(resource: Any) -> list[str]: + status = resource.get("status") if isinstance(resource, dict) else None + conditions = status.get("conditions") if isinstance(status, dict) else None + if not isinstance(conditions, list): + return [] + texts = [] + for condition in conditions: + if not isinstance(condition, dict): + continue + state = condition.get("state", condition.get("status")) + if state not in ("CONDITION_FAILED", "False", False): + continue + for key in ("reason", "message"): + value = condition.get(key) + if isinstance(value, str): + texts.append(value) + return texts + + +def _audit_identity_matches( + entry: dict[str, Any], *, project: str, region: str, service: str +) -> bool: + payload = entry.get("protoPayload") + payload = payload if isinstance(payload, dict) else {} + resource_name = payload.get("resourceName") + labels = entry.get("resource") + labels = labels.get("labels") if isinstance(labels, dict) else None + exact_labels = ( + isinstance(labels, dict) + and labels.get("project_id") == project + and labels.get("location") == region + and labels.get("service_name") == service + ) + v2_name = f"projects/{project}/locations/{region}/services/{service}" + v1_name = f"namespaces/{project}/services/{service}" + if resource_name is not None: + if not isinstance(resource_name, str): + return False + if resource_name == v2_name: + return True + return resource_name == v1_name and exact_labels + return exact_labels + + +def _audit_error_texts( + entries: Any, + *, + project: str, + region: str, + service: str, + window_start: str, + window_end: str, +) -> tuple[int | None, int | None, list[str]]: + if not isinstance(entries, list): + return None, None, [] + try: + start_time = datetime.fromisoformat(window_start.replace("Z", "+00:00")) + end_time = datetime.fromisoformat(window_end.replace("Z", "+00:00")) + except (TypeError, ValueError): + return None, None, [] + if ( + start_time.tzinfo is None + or end_time.tzinfo is None + or end_time < start_time + ): + return None, None, [] + failed = [] + matching_count = 0 + for entry in entries: + if not isinstance(entry, dict): + continue + if not _audit_identity_matches(entry, project=project, region=region, service=service): + continue + timestamp = entry.get("timestamp") + if not isinstance(timestamp, str): + continue + try: + observed_at = datetime.fromisoformat(timestamp.replace("Z", "+00:00")) + except ValueError: + continue + if observed_at.tzinfo is None or not start_time <= observed_at <= end_time: + continue + payload = entry.get("protoPayload") + payload = payload if isinstance(payload, dict) else {} + method = payload.get("methodName") + if not isinstance(method, str) or not any( + marker in method for marker in ("UpdateService", "CreateService", "ReplaceService") + ): + continue + matching_count += 1 + status = payload.get("status") + status = status if isinstance(status, dict) else entry.get("status") + if not isinstance(status, dict): + continue + code = status.get("code") + message = status.get("message") + is_error = (isinstance(code, int) and code != 0) or (isinstance(message, str) and bool(message)) + if is_error: + failed.append((code, message if isinstance(message, str) else "")) + texts = [message for _, message in failed if message] + if any(code == 7 for code, _ in failed): + texts.append("permission_denied") + return matching_count, len(failed), texts + + +def _control_summary(policy: Any, jobs: Any) -> dict[str, Any]: + if ( + isinstance(policy, dict) + and isinstance(policy.get("bindings"), list) + and all(isinstance(binding, dict) for binding in policy["bindings"]) + ): + binding_count: int | None = len(policy["bindings"]) + else: + binding_count = None + if isinstance(jobs, list) and all(isinstance(item, dict) for item in jobs): + states = {"enabled": 0, "paused": 0, "other": 0} + for job in jobs: + state = job.get("state") + if state == "ENABLED": + states["enabled"] += 1 + elif state == "PAUSED": + states["paused"] += 1 + else: + states["other"] += 1 + job_count: int | None = len(jobs) + else: + states = {"enabled": None, "paused": None, "other": None} + job_count = None + return { + "iam_policy_readable": binding_count is not None, + "iam_binding_count": binding_count, + "scheduler_readable": job_count is not None, + "scheduler_job_count": job_count, + "scheduler_state_counts": states, + } + + +def summarize( + *, + service: Any, + revisions: Any, + expected_service: str, + expected_project: str, + expected_region: str, + audit_window_start: str, + audit_window_end: str, + policy: Any, + jobs: Any, + audit_entries: Any, + audit_status: str, +) -> dict[str, Any]: + service_ok = isinstance(service, dict) + metadata = service.get("metadata") if service_ok else None + status = service.get("status") if service_ok else None + status = status if isinstance(status, dict) else {} + traffic = status.get("traffic") + traffic_rows = traffic if isinstance(traffic, list) and all(isinstance(row, dict) for row in traffic) else None + positive = [] + if traffic_rows is not None: + for row in traffic_rows: + percent = row.get("percent", 0) + if isinstance(percent, int) and not isinstance(percent, bool) and percent > 0: + positive.append((row, percent)) + traffic_valid = ( + traffic_rows is not None + and len(positive) == 1 + and positive[0][1] == 100 + and isinstance(status.get("latestReadyRevisionName"), str) + and positive[0][0].get("revisionName") == status.get("latestReadyRevisionName") + ) + latest = _revision_for_service(service, revisions) + revision_texts = _condition_error_texts(latest) + latest_ready = _ready(latest) + revision_categories = ( + _classify(revision_texts) + if revision_texts + else [] + if latest_ready is True + else ["unknown"] + ) + audit_entries_count, audit_error_count, audit_texts = _audit_error_texts( + audit_entries, + project=expected_project, + region=expected_region, + service=expected_service, + window_start=audit_window_start, + window_end=audit_window_end, + ) + audit_categories = _classify(audit_texts) if audit_error_count else [] + audit_categories = audit_categories if audit_status == "ok" else ["unknown"] + if audit_error_count and audit_status == "ok": + failure_source = "audit" + failure_categories = audit_categories + elif latest_ready is False: + failure_source = "revision" + failure_categories = revision_categories + else: + failure_source = "none" + failure_categories = ["unknown"] + result = { + "schema_version": "firstrade_cached_stage_diagnostic.v1", + "service_readable": service_ok, + "target_matches": bool(isinstance(metadata, dict) and metadata.get("name") == expected_service), + "service_ready": _ready(service), + "traffic_row_count": len(traffic_rows) if traffic_rows is not None else None, + "positive_traffic_row_count": len(positive), + "traffic_is_single_ready_revision_at_100_percent": bool(traffic_valid), + "latest_created_revision_found": latest is not None, + "revision_list_readable": isinstance(revisions, list) + and all(isinstance(item, dict) for item in revisions), + "latest_created_revision_ready": latest_ready, + "latest_revision_error_categories": revision_categories, + **_control_summary(policy, jobs), + "audit_query_status": audit_status, + "audit_entry_count": audit_entries_count if audit_status == "ok" else None, + "audit_error_count": audit_error_count if audit_status == "ok" else None, + "audit_error_categories": audit_categories, + "failure_source": failure_source, + "failure_categories": failure_categories, + } + return result + + +def _validate_window(start: str, end: str) -> None: + pattern = r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d+)?Z" + if not re.fullmatch(pattern, start) or not re.fullmatch(pattern, end): + raise ValueError("time_window_invalid") + try: + start_time = datetime.fromisoformat(start.replace("Z", "+00:00")) + end_time = datetime.fromisoformat(end.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("time_window_invalid") from exc + seconds = (end_time - start_time).total_seconds() + if seconds < 0 or seconds > 1800 or end_time > datetime.now(UTC): + raise ValueError("time_window_invalid") + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--service") + parser.add_argument("--revisions") + parser.add_argument("--expected-service") + parser.add_argument("--expected-project") + parser.add_argument("--expected-region") + parser.add_argument("--window-start") + parser.add_argument("--window-end") + parser.add_argument("--iam-policy") + parser.add_argument("--scheduler-jobs") + parser.add_argument("--audit-entries") + parser.add_argument("--audit-status", choices=("ok", "permission_denied", "unavailable")) + parser.add_argument("--validate-window", nargs=2, metavar=("START", "END")) + args = parser.parse_args() + if args.validate_window: + try: + _validate_window(*args.validate_window) + except ValueError: + print("time_window_invalid", file=sys.stderr) + return 2 + return 0 + required = ( + args.service, + args.revisions, + args.expected_service, + args.expected_project, + args.expected_region, + args.window_start, + args.window_end, + args.iam_policy, + args.scheduler_jobs, + args.audit_entries, + args.audit_status, + ) + if any(value is None for value in required): + parser.error("summary inputs are required") + summary = summarize( + service=_read_json(args.service), + revisions=_read_json(args.revisions), + expected_service=args.expected_service, + expected_project=args.expected_project, + expected_region=args.expected_region, + audit_window_start=args.window_start, + audit_window_end=args.window_end, + policy=_read_json(args.iam_policy), + jobs=_read_json(args.scheduler_jobs), + audit_entries=_read_json(args.audit_entries), + audit_status=args.audit_status, + ) + print(json.dumps(summary, sort_keys=True, separators=(",", ":"))) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_cached_stage_failure_diagnostic.py b/tests/test_cached_stage_failure_diagnostic.py new file mode 100644 index 0000000..628c714 --- /dev/null +++ b/tests/test_cached_stage_failure_diagnostic.py @@ -0,0 +1,246 @@ +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta +from pathlib import Path + +import pytest + +from scripts import diagnose_cached_stage_failure as diagnostic + + +PRIVATE_SERVICE = "private-service-placeholder" +PRIVATE_NAME = "private-revision-placeholder" +PRIVATE_MESSAGE = "secret placeholder: failed to pull image for private-service-placeholder" +PRIVATE_PROJECT = "private-project-placeholder" +PRIVATE_REGION = "private-region-placeholder" +WINDOW_START = "2026-10-02T00:00:00Z" +WINDOW_END = "2026-10-02T00:10:00Z" + + +def _evidence(message: str = PRIVATE_MESSAGE): + service = { + "metadata": {"name": PRIVATE_SERVICE}, + "status": { + "conditions": [{"type": "Ready", "state": "CONDITION_SUCCEEDED"}], + "latestCreatedRevisionName": PRIVATE_NAME, + "latestReadyRevisionName": PRIVATE_NAME, + "traffic": [{"revisionName": PRIVATE_NAME, "percent": 100}], + }, + } + revisions = [ + { + "metadata": {"name": PRIVATE_NAME}, + "status": { + "conditions": [ + {"type": "Ready", "state": "CONDITION_FAILED", "reason": "ContainerFailure", "message": message} + ] + }, + "spec": {"containers": [{"env": [{"name": "PRIVATE_KEY", "value": "private-value-placeholder"}]}]}, + } + ] + policy = { + "bindings": [ + {"role": "roles/run.invoker", "members": ["serviceAccount:private@example.invalid"]} + ] + } + jobs = [{"name": "private-job-placeholder", "state": "PAUSED"}] + audit = [ + { + "timestamp": "2026-10-02T00:05:00Z", + "resource": { + "labels": { + "project_id": PRIVATE_PROJECT, + "location": PRIVATE_REGION, + "service_name": PRIVATE_SERVICE, + } + }, + "protoPayload": { + "methodName": "google.cloud.run.v2.Services.UpdateService", + "resourceName": f"projects/{PRIVATE_PROJECT}/locations/{PRIVATE_REGION}/services/{PRIVATE_SERVICE}", + "request": {"env": "private-value-placeholder"}, + "status": {"code": 13, "message": message}, + }, + } + ] + return service, revisions, policy, jobs, audit + + +def test_summary_exposes_only_closed_statuses_counts_and_categories(): + summary = diagnostic.summarize( + service=_evidence()[0], + revisions=_evidence()[1], + expected_service=PRIVATE_SERVICE, + expected_project=PRIVATE_PROJECT, + expected_region=PRIVATE_REGION, + audit_window_start=WINDOW_START, + audit_window_end=WINDOW_END, + policy=_evidence()[2], + jobs=_evidence()[3], + audit_entries=_evidence()[4], + audit_status="ok", + ) + + assert summary == { + "schema_version": "firstrade_cached_stage_diagnostic.v1", + "service_readable": True, + "target_matches": True, + "service_ready": True, + "traffic_row_count": 1, + "positive_traffic_row_count": 1, + "traffic_is_single_ready_revision_at_100_percent": True, + "latest_created_revision_found": True, + "revision_list_readable": True, + "latest_created_revision_ready": False, + "latest_revision_error_categories": ["image"], + "iam_policy_readable": True, + "iam_binding_count": 1, + "scheduler_readable": True, + "scheduler_job_count": 1, + "scheduler_state_counts": {"enabled": 0, "paused": 1, "other": 0}, + "audit_query_status": "ok", + "audit_entry_count": 1, + "audit_error_count": 1, + "audit_error_categories": ["image"], + "failure_source": "audit", + "failure_categories": ["image"], + } + serialized = json.dumps(summary) + for private_value in (PRIVATE_SERVICE, PRIVATE_NAME, PRIVATE_MESSAGE, "private-value-placeholder", "private@example.invalid"): + assert private_value not in serialized + + +@pytest.mark.parametrize( + ("message", "expected"), + [ + ("permission denied: PERMISSION_DENIED", ["permission"]), + ("iam.serviceAccounts.actAs is required", ["act_as"]), + ("revision name is invalid", ["invalid_name"]), + ("failed to pull image manifest", ["image"]), + ("container failed to listen on the port", ["port"]), + ("startup probe failed", ["startup"]), + ("secret version was not found", ["missing_secret"]), + ( + "traffic[].tag: traffic tag [TAG] and service name [SERVICE] together are too long. " + "Combined traffic tag and service name cannot exceed 46 characters.", + ["invalid_name"], + ), + ("opaque upstream diagnostic", ["unknown"]), + ], +) +def test_error_text_is_reduced_to_fixed_reason(message, expected): + assert diagnostic._classify([message]) == expected + + +def test_missing_logging_permission_is_reported_without_guessing_failure_category(): + service, revisions, policy, jobs, audit = _evidence("permission denied") + 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="permission_denied", + ) + assert summary["audit_query_status"] == "permission_denied" + assert summary["audit_entry_count"] is None + assert summary["failure_source"] == "revision" + assert summary["failure_categories"] == ["permission"] + + +def test_unreadable_metadata_stays_unknown_and_never_serializes_input(): + summary = diagnostic.summarize( + service={"private": "private-value-placeholder"}, + revisions={"private": PRIVATE_NAME}, + expected_service=PRIVATE_SERVICE, + expected_project=PRIVATE_PROJECT, + expected_region=PRIVATE_REGION, + audit_window_start=WINDOW_START, + audit_window_end=WINDOW_END, + policy=None, + jobs=None, + audit_entries=None, + audit_status="unavailable", + ) + assert summary["service_ready"] is None + assert summary["latest_created_revision_found"] is False + assert summary["revision_list_readable"] is False + assert summary["iam_policy_readable"] is False + assert summary["scheduler_readable"] is False + assert summary["audit_query_status"] == "unavailable" + assert PRIVATE_NAME not in json.dumps(summary) + + +def test_audit_events_must_match_exact_project_region_service_and_time_window(): + service, revisions, policy, jobs, audit = _evidence() + same_prefix = json.loads(json.dumps(audit[0])) + same_prefix["protoPayload"]["resourceName"] = ( + f"projects/{PRIVATE_PROJECT}/locations/{PRIVATE_REGION}/services/{PRIVATE_SERVICE}-suffix" + ) + same_prefix["resource"]["labels"]["service_name"] = f"{PRIVATE_SERVICE}-suffix" + other_region = json.loads(json.dumps(audit[0])) + other_region["protoPayload"]["resourceName"] = ( + f"projects/{PRIVATE_PROJECT}/locations/other-region/services/{PRIVATE_SERVICE}" + ) + other_region["resource"]["labels"]["location"] = "other-region" + outside_window = json.loads(json.dumps(audit[0])) + outside_window["timestamp"] = "2026-10-02T00:10:01Z" + 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, same_prefix, other_region, outside_window], + audit_status="ok", + ) + assert summary["audit_entry_count"] == 1 + assert summary["audit_error_count"] == 1 + assert summary["audit_error_categories"] == ["image"] + + +def test_failure_window_is_utc_and_limited_to_thirty_minutes(): + end = datetime.now(UTC).replace(microsecond=0) + start = end - timedelta(minutes=30) + diagnostic._validate_window(start.isoformat().replace("+00:00", "Z"), end.isoformat().replace("+00:00", "Z")) + with pytest.raises(ValueError, match="time_window_invalid"): + diagnostic._validate_window( + (end - timedelta(minutes=31)).isoformat().replace("+00:00", "Z"), + end.isoformat().replace("+00:00", "Z"), + ) + + +def test_workflow_is_opt_in_read_only_and_reuses_existing_identity(): + workflow = (Path(__file__).resolve().parents[1] / ".github/workflows/diagnose-cached-stage-failure.yml").read_text( + encoding="utf-8" + ) + assert "default: false" in workflow + assert "inputs.run_readonly_diagnostic == true" in workflow + assert "refs/heads/main" in workflow + assert "secrets.CLOUD_RUN_SERVICE" in workflow + assert "vars.CLOUD_RUN_REGION" in workflow + assert "GCP_PROJECT_ID: firstradequant" in workflow + assert "google-github-actions/auth@v3" in workflow + assert "firstrade-platform-deploy@firstradequant.iam.gserviceaccount.com" in workflow + assert "github-main" in workflow + assert "gcloud run services describe" in workflow + assert "gcloud run revisions list" in workflow + assert "gcloud run services get-iam-policy" in workflow + assert "gcloud scheduler jobs list" in workflow + assert "gcloud logging read" in workflow + assert 'projects/${GCP_PROJECT_ID}/locations/${CLOUD_RUN_REGION}/services/${CLOUD_RUN_SERVICE}' in workflow + assert "resource.labels.project_id=\\\"${GCP_PROJECT_ID}\\\"" in workflow + assert "resource.labels.location=\\\"${CLOUD_RUN_REGION}\\\"" in workflow + assert "gcloud run deploy" not in workflow + assert "docker build" not in workflow + assert "docker push" not in workflow + assert "actions/upload-artifact" not in workflow