Crashlyticsの毎朝トリアージを5つのAntigravityサブエージェントに分担させた話

# agents/fetcher/main.py
import json
import os
from datetime import datetime, timedelta, timezone
from google.cloud import bigquery
 
PII_FIELDS = {"user_email", "custom_keys.user_id", "device.serial"}
 
def normalize_crash(row: dict) -> dict:
    """生クラッシュをPII除去後の正規化形式に変換する"""
    return {
        "issue_id": row["issue_id"],
        "title": row["issue_title"],
        "subtitle": row.get("issue_subtitle", ""),
        "app_id": row["app_id"],
        "app_version": row["application"]["display_version"],
        "os": f'{row["os"]["name"]} {row["os"]["display_version"]}',
        "device": f'{row["device"]["manufacturer"]} {row["device"]["model"]}',
        "occurred_at": row["event_timestamp"].isoformat(),
        "fatal": row["is_fatal"],
        "stack_top": _top_frames(row["exceptions"][0]["frames"], n=8),
    }
 
def _top_frames(frames: list, n: int) -> list:
    """先頭nフレームのみを抽出。アプリ自前のシンボルのみを優先する"""
    own_prefix = os.environ["APP_BUNDLE_PREFIX"]
    own = [f for f in frames if f.get("file", "").startswith(own_prefix)]
    rest = [f for f in frames if f not in own]
    chosen = (own + rest)[:n]
    return [{"file": f["file"], "line": f["line"], "symbol": f.get("symbol")} for f in chosen]
 
def fetch(app_ids: list[str], hours: int = 24) -> dict:
    client = bigquery.Client()
    since = datetime.now(timezone.utc) - timedelta(hours=hours)
    sql = """
    SELECT * FROM `crashlytics.firebase_crashlytics.*`
    WHERE event_timestamp > @since
      AND is_fatal = TRUE
      AND application.app_id IN UNNEST(@app_ids)
    LIMIT 50
    """
    job = client.query(
        sql,
        job_config=bigquery.QueryJobConfig(
            query_parameters=[
                bigquery.ScalarQueryParameter("since", "TIMESTAMP", since),
                bigquery.ArrayQueryParameter("app_ids", "STRING", app_ids),
            ]
        ),
    )
    crashes = [normalize_crash(dict(row)) for row in job.result()]
    return {"schema": "crash-batch-v1", "fetched_at": datetime.now(timezone.utc).isoformat(), "crashes": crashes}
 
if __name__ == "__main__":
    out = fetch(json.loads(os.environ["APP_IDS"]))
    print(json.dumps(out, ensure_ascii=False))

// agents/classifier/prompt.ts
export const CLASSIFIER_INSTRUCTIONS = `
You are a triage classifier for mobile app crashes.
For each crash, output one of:
  - "regression": file modified within last 30 days AND not previously reported
  - "known_open": same issue_id reported ≥3 times in last 90 days
  - "external": top frame in vendor SDK (Firebase, FBSDK, etc.)
  - "device_specific": occurs in ≥80% same device or OS
 
Append a priority_score 0-100 weighted by:
  - regression: base 70
  - known_open with recent volume spike: base 60
  - external: base 20
  - device_specific affecting current OS: base 50
 
Multiply by app weight (provided in input) for revenue-bearing apps.
Use codebase-search to verify file existence before classifying as regression.
`;

# agents/repro/runner.py
import subprocess
import re
import json
from pathlib import Path
 
class ReproResult:
    REPRODUCED = "reproduced"
    FLAKY = "flaky"
    NOT_REPRODUCED = "not_reproduced"
    SKIPPED = "skipped"
 
def run_ios_repro(script_path: Path, scheme: str, timeout: int = 300) -> dict:
    """xcodebuild testで再現スクリプトを3回実行し、揺らぎを判定する"""
    successes = 0
    failures = 0
    for attempt in range(3):
        try:
            res = subprocess.run(
                ["xcodebuild", "test",
                 "-scheme", scheme,
                 "-destination", "platform=iOS Simulator,name=iPhone 15",
                 "-only-testing", str(script_path)],
                capture_output=True, text=True, timeout=timeout,
            )
            # クラッシュ再現テストは「失敗するのが正解」
            if "Test Suite 'All tests' failed" in res.stdout:
                successes += 1
            else:
                failures += 1
        except subprocess.TimeoutExpired:
            failures += 1
 
    if successes == 3:
        return {"status": ReproResult.REPRODUCED, "successes": 3}
    if successes >= 1:
        return {"status": ReproResult.FLAKY, "successes": successes}
    return {"status": ReproResult.NOT_REPRODUCED, "successes": 0}

# agents/patch/prompt_builder.py
from pathlib import Path
import json
 
def build_patch_prompt(crash: dict, repro: dict) -> str:
    """再現済みクラッシュの修正プロンプトを組み立てる"""
    return f"""
ROLE: You are a senior iOS/Android engineer working with a 12-year indie developer.
 
CRASH:
{json.dumps(crash, ensure_ascii=False, indent=2)}
 
REPRO:
- script: {repro['script_path']}
- status: {repro['status']}
- successes: {repro['successes']} / 3
 
REQUIREMENTS:
1. Produce a minimal patch that makes the repro script pass (i.e., no crash).
2. Add a regression test in the same module as the crashing file.
3. DO NOT change unrelated public APIs.
4. DO NOT introduce new third-party dependencies.
5. If the fix requires architectural changes, output a "needs-discussion" marker instead of code.
 
OUTPUT FORMAT:
- patch: unified diff
- test: full new test file content
- risk_level: "low" | "medium" | "high"
- rollback_note: 1-2 sentences describing how to revert
"""

# agents/pr/decision.py
def decide_review_mode(patch: dict, crash: dict, app_meta: dict) -> str:
    """3つの条件を全て満たすときのみ自動承認モードを有効化する"""
    if patch["risk_level"] != "low":
        return "human_review_required"
    if app_meta["monthly_revenue_jpy"] > 100_000:
        return "human_review_required"
    if crash["app_version"].endswith(".0"):  # メジャー直後は慎重に
        return "human_review_required"
    return "auto_approve_candidate"