リリース後72時間を Antigravity Background Agent に見張らせる：6アプリ並行アップデートで固まった指標監視ワークフロー

# release_monitor/runner.py
from datetime import datetime, timedelta
from typing import TypedDict
 
class MetricResult(TypedDict):
    name: str
    value: float
    threshold: float
    breached: bool
    source: str
 
def collect_metrics(app_id: str, build: str) -> list[MetricResult]:
    return [
        crashlytics_crash_free(app_id, build),
        play_vitals_anr(app_id, build),
        admob_ecpm_dip(app_id, build),
        ga4_d1_retention(app_id, build),
    ]
 
def build_report(app_id: str, build: str) -> str:
    metrics = collect_metrics(app_id, build)
    breached = [m for m in metrics if m["breached"]]
 
    header = f"# {app_id} / build {build} / {datetime.utcnow().isoformat()}Z\n\n"
    if not breached:
        return header + "全4指標は警告ライン内です。次の集計まで待機します。\n"
 
    body = "\n".join(
        f"- ⚠️ **{m['name']}** = {m['value']:.3f} (threshold {m['threshold']:.3f}, source: {m['source']})"
        for m in breached
    )
    return header + "## 警告ライン超過\n\n" + body + "\n"

-- bigquery/d1_retention.sql
WITH first_open AS (
  SELECT
    user_pseudo_id,
    MIN(event_date) AS first_date
  FROM `dolice-wallpapers.analytics_xxxxx.events_*`
  WHERE event_name = 'first_open'
    AND _TABLE_SUFFIX BETWEEN @start_date AND @end_date
  GROUP BY user_pseudo_id
),
returned AS (
  SELECT DISTINCT user_pseudo_id
  FROM `dolice-wallpapers.analytics_xxxxx.events_*`
  WHERE event_name = 'session_start'
    AND _TABLE_SUFFIX BETWEEN @start_date AND @end_date
)
SELECT
  fo.first_date,
  COUNT(DISTINCT fo.user_pseudo_id) AS new_users,
  COUNT(DISTINCT r.user_pseudo_id) AS retained_d1,
  SAFE_DIVIDE(COUNT(DISTINCT r.user_pseudo_id), COUNT(DISTINCT fo.user_pseudo_id)) AS d1_rate
FROM first_open fo
LEFT JOIN returned r
  ON fo.user_pseudo_id = r.user_pseudo_id
WHERE r.user_pseudo_id IN (
  SELECT user_pseudo_id FROM `dolice-wallpapers.analytics_xxxxx.events_*`
  WHERE event_name = 'session_start'
    AND PARSE_DATE('%Y%m%d', event_date) = DATE_ADD(PARSE_DATE('%Y%m%d', fo.first_date), INTERVAL 1 DAY)
)
GROUP BY fo.first_date
ORDER BY fo.first_date DESC;

# AGENTS.md（抜粋・release-monitor リポジトリ）
 
このリポジトリは Antigravity Background Agent によるリリース後監視ジョブを動かす場所です。
 
## 認証
- `GOOGLE_APPLICATION_CREDENTIALS` に bq 用サービスアカウント鍵のパスをセット
- AdMob は `ADMOB_OAUTH_REFRESH_TOKEN` を環境変数で渡す（Reporting API スコープ付き）
- Crashlytics は Firebase REST API 経由なので、上記サービスアカウント鍵で兼用可能
 
## 実行
- `python release_monitor/runner.py --app=<bundle_id> --build=<version>`
- 1時間ごとに走らせる場合は `cron/release-monitor.cron` を参照
- 出力は `reports/<utc-timestamp>.md`
 
## 触ってはいけない場所
- `secrets/`（鍵類）。手動配置・手動更新のみ。
- `reports/archive/` 配下の過去レポート（追記禁止・新規作成のみ）

-- bigquery/crashlytics_top_issues.sql
SELECT
  issue_id,
  ANY_VALUE(issue_title) AS title,
  ANY_VALUE(issue_subtitle) AS subtitle,
  COUNT(DISTINCT installation_uuid) AS affected_installs,
  COUNT(*) AS crash_count,
  MIN(event_timestamp) AS first_seen,
  MAX(event_timestamp) AS last_seen,
  ANY_VALUE(application.display_version) AS version
FROM `dolice-wallpapers.firebase_crashlytics.<app_id>_ANDROID_REALTIME`
WHERE _PARTITIONTIME >= TIMESTAMP_SUB(CURRENT_TIMESTAMP(), INTERVAL 72 HOUR)
  AND application.display_version = @target_version
  AND event_type = 'crash'
GROUP BY issue_id
ORDER BY affected_installs DESC
LIMIT 10;

# release_monitor/crashlytics_summary.py
import json, subprocess, textwrap
 
def fetch_top_issues(app_id: str, version: str) -> list[dict]:
    out = subprocess.check_output([
        "bq", "query", "--use_legacy_sql=false", "--format=json",
        f"--parameter=target_version::{version}",
        open("bigquery/crashlytics_top_issues.sql").read(),
    ])
    return json.loads(out)
 
def summarize_for_agent(issues: list[dict]) -> str:
    if not issues:
        return "クラッシュなし。"
    lines = []
    for i, issue in enumerate(issues[:3], 1):
        lines.append(textwrap.dedent(f"""
        ### Issue {i}: {issue["title"]}
        - 影響インストール数: {issue["affected_installs"]}
        - クラッシュ件数: {issue["crash_count"]}
        - 初回検出: {issue["first_seen"]}
        - 推定要因: {issue["subtitle"]}
        """).strip())
    return "\n\n".join(lines)

# release_monitor/admob_ecpm.py
import os, requests, statistics
from datetime import date, timedelta
 
ADMOB_REPORT_URL = (
    "https://admob.googleapis.com/v1/accounts/{publisher_id}/networkReport:generate"
)
 
def fetch_ecpm(publisher_id: str, ad_unit_id: str, days_back: int) -> list[float]:
    end = date.today()
    start = end - timedelta(days=days_back)
    body = {
        "report_spec": {
            "date_range": {
                "start_date": {"year": start.year, "month": start.month, "day": start.day},
                "end_date": {"year": end.year, "month": end.month, "day": end.day},
            },
            "dimensions": ["DATE", "AD_UNIT"],
            "metrics": ["AD_REQUESTS", "MATCHED_REQUESTS", "IMPRESSIONS", "ESTIMATED_EARNINGS"],
            "dimension_filters": [{
                "dimension": "AD_UNIT",
                "matches_any": {"values": [ad_unit_id]},
            }],
        }
    }
    headers = {"Authorization": f"Bearer {os.environ['ADMOB_ACCESS_TOKEN']}"}
    res = requests.post(ADMOB_REPORT_URL.format(publisher_id=publisher_id), json=body, headers=headers)
    res.raise_for_status()
    rows = res.json().get("row", [])
    return [
        (int(r["metricValues"]["ESTIMATED_EARNINGS"]["microsValue"]) / 1_000_000)
        / max(int(r["metricValues"]["IMPRESSIONS"]["integerValue"]), 1) * 1000
        for r in rows if "metricValues" in r
    ]
 
def detect_dip(ecpm_series: list[float], window: int = 7, threshold: float = 0.20) -> bool:
    if len(ecpm_series) <= window:
        return False
    baseline = statistics.median(ecpm_series[:-1][-window:])
    latest = ecpm_series[-1]
    return baseline > 0 and (baseline - latest) / baseline >= threshold

# release_monitor/compose_report.py
from pathlib import Path
 
def compose(app_id: str, build: str) -> Path:
    crash_free = crashlytics_crash_free(app_id, build)
    anr = play_vitals_anr(app_id, build)
    ecpm = admob_ecpm_dip(app_id, build)
    d1 = ga4_d1_retention(app_id, build)
 
    md = [
        f"# Release Monitor / {app_id} / build {build}",
        "",
        "## サマリー（72h 集計）",
        f"| 指標 | 現在値 | 閾値 | 判定 |",
        f"| --- | --- | --- | --- |",
        f"| crash-free users | {crash_free['value']:.3%} | 99.5% | {'⚠️ 超過' if crash_free['breached'] else '✅ 安全'} |",
        f"| ANR rate | {anr['value']:.3%} | 0.47% | {'⚠️ 超過' if anr['breached'] else '✅ 安全'} |",
        f"| eCPM dip | {ecpm['value']:.1%} | 20% | {'⚠️ 超過' if ecpm['breached'] else '✅ 安全'} |",
        f"| D1 retention | {d1['value']:.3%} | base-5pt | {'⚠️ 超過' if d1['breached'] else '✅ 安全'} |",
        "",
        "## 推奨アクション",
        recommended_action(crash_free, anr, ecpm, d1),
    ]
    out = Path(f"reports/{app_id}-{build}.md")
    out.write_text("\n".join(md), encoding="utf-8")
    return out

# .github/workflows/staged-rollout.yml
name: Staged Rollout Check
on:
  schedule:
    - cron: "0 */3 * * *"   # 3時間ごと
  workflow_dispatch:
 
jobs:
  check:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - name: Run release monitor
        env:
          GOOGLE_APPLICATION_CREDENTIALS: ${{ secrets.GCP_SA_KEY_PATH }}
          ADMOB_ACCESS_TOKEN: ${{ secrets.ADMOB_TOKEN }}
        run: |
          python release_monitor/runner.py \
            --app=jp.dolice.wallpaper.beautifulhd \
            --build=${{ vars.CURRENT_BUILD }}
      - name: Post report as commit comment
        run: |
          gh api repos/${{ github.repository }}/commits/${{ github.sha }}/comments \
            -f body="$(cat reports/jp.dolice.wallpaper.beautifulhd-${{ vars.CURRENT_BUILD }}.md)"