cleveragents-core/tools/_forgejo_cache.py

"""Local SQLite cache of Forgejo data for the auto-agents repo (defaults
to ``cleveragents/cleveragents-core``; overridable via ``FORGEJO_OWNER``
and ``FORGEJO_REPO``).

Used by ``count-master-merges.py`` and ``pr-stats.py`` to answer all
master-branch and PR-lifecycle questions from a local database, syncing only
deltas from the API instead of re-fetching the whole window on every run.

Design
------
- **Single SQLite file** at ``tools/.cache/forgejo.sqlite`` (gitignored).
- **Three data tables**:
    ``commits``       - master branch, keyed by full SHA, append-only
    ``pulls``         - every PR we've seen, keyed by PR number
    ``reachability``  - cached compare/master...sha results
  Plus ``sync_meta`` for cursors and ``schema_version`` for migrations.
- **Every row stores the raw API JSON** in a ``raw`` column so new fields can
  be surfaced without a cache rebuild; parsed columns are indexed
  projections for efficient queries.
- **Delta sync** walks the commits and PR endpoints from page 1 (newest
  first) and stops as soon as the current page is entirely already-cached
  (commits) or its max ``updated_at`` is older than the last sync
  (PRs). Initial seeding is just "delta sync from an empty cache" and
  continues until the API pages are exhausted.
- **PR enrichment** (``merged_by``) is lazy: on read, any terminal-state PR
  that lacks ``merged_by_login`` but needs it gets one PR-detail fetch,
  cached permanently.
- **Concurrency**: ``fcntl.flock`` on ``forgejo.sqlite.lock`` during syncs.
  SQLite handles concurrent reads natively.
- **Force-push detection**: sync_meta tracks ``master_head_sha``; if the
  previous HEAD is not seen anywhere in the first N pages of current master,
  we emit a warning. The user can pass ``--full`` to rebuild.

Public API (used by the scripts)
--------------------------------
::

    cache = ForgejoCache.open()
    cache.sync(token)                               # delta-sync forward
    commits = cache.commits_on_master(start, end)   # list of raw commit dicts
    closed  = cache.pulls_closed_in_window(start, end)
    merged  = cache.pulls_merged_in_window(start, end)
    opened  = cache.pulls_opened_in_window(start, end)
    sha_map = cache.sha_to_pr_map(start, end)
    reach   = cache.is_sha_reachable_from_master(sha, token)
    detail  = cache.enrich_pr(number, token)

All list/map queries return the raw PR/commit dicts exactly as the Forgejo
API returned them (deserialised from the ``raw`` column), so callers that
already expect raw API shape need no changes.
"""
from __future__ import annotations

import fcntl
import json
import os
import re
import sqlite3
import sys
import time
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Iterable, Sequence

# ─── Config ─────────────────────────────────────────────────────────────────

# Env-driven so fork-mode reporting (`FORGEJO_OWNER=drew` etc.) reads
# from the fork's data instead of the canonical repo. The defaults
# preserve the original behaviour: when run without env vars in a
# canonical-mode shell, the cache, every API call, and every cache file
# resolve exactly as before.
REPO_OWNER = os.environ.get("FORGEJO_OWNER", "cleveragents")
REPO_NAME = os.environ.get("FORGEJO_REPO", "cleveragents-core")
API_BASE = os.environ.get(
    "FORGEJO_API_BASE", "https://git.cleverthis.com/api/v1"
).rstrip("/")
REQUEST_TIMEOUT_SEC = 30
REQUEST_RETRIES = 3
RETRY_BACKOFF_SEC = 2.0

DEFAULT_CACHE_DIR = Path(__file__).resolve().parent / ".cache"
# Cache path is partitioned per-(owner, repo) so a fork-mode run never
# clobbers the canonical cache (and vice versa). The default
# ``cleveragents/cleveragents-core`` keeps the historical filename
# ``forgejo.sqlite`` for backward compatibility with existing checkouts.
if REPO_OWNER == "cleveragents" and REPO_NAME == "cleveragents-core":
    DEFAULT_CACHE_PATH = DEFAULT_CACHE_DIR / "forgejo.sqlite"
else:
    _safe_owner = re.sub(r"[^a-zA-Z0-9._-]+", "-", REPO_OWNER)
    _safe_repo = re.sub(r"[^a-zA-Z0-9._-]+", "-", REPO_NAME)
    DEFAULT_CACHE_PATH = (
        DEFAULT_CACHE_DIR / f"forgejo.{_safe_owner}.{_safe_repo}.sqlite"
    )
# Schema versions:
#   1 — original (commits, pulls, reachability, sync_meta)
#   2 — Tier 1B telemetry: merge_cycle, ci_gate_events, llm_activity
SCHEMA_VERSION = 3

# Safety ceiling on pagination during a single sync pass.
SYNC_PAGE_HARD_LIMIT = 4000  # 4000 * 50 = 200,000 rows; far above anything real.

# How many consecutive pages with no new rows before we declare the sync done.
SYNC_EMPTY_PAGE_TOLERANCE = 2


# ─── HTTP helper (mirrors count-master-merges.py) ──────────────────────────

def api_get(url: str, token: str) -> Any:
    """GET a Forgejo API URL with retries on transient network errors."""
    req = urllib.request.Request(url, headers={"Authorization": f"token {token}"})
    last_err: Exception | None = None
    for attempt in range(1, REQUEST_RETRIES + 1):
        try:
            with urllib.request.urlopen(req, timeout=REQUEST_TIMEOUT_SEC) as resp:
                return json.loads(resp.read())
        except urllib.error.HTTPError:
            # HTTP errors are authoritative: propagate so caller can decide
            # (e.g. 404 on compare endpoint means "not reachable").
            raise
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            last_err = e
            if attempt < REQUEST_RETRIES:
                time.sleep(RETRY_BACKOFF_SEC * attempt)
                continue
    raise RuntimeError(f"cannot reach {url} after {REQUEST_RETRIES} attempts: {last_err}")


# ─── Cache ─────────────────────────────────────────────────────────────────

SCHEMA_SQL = """
CREATE TABLE IF NOT EXISTS schema_version (
    version INTEGER PRIMARY KEY
);

CREATE TABLE IF NOT EXISTS commits (
    sha TEXT PRIMARY KEY,
    committer_date TEXT NOT NULL,
    author_name TEXT,
    committer_name TEXT,
    message TEXT,
    parents TEXT,
    raw TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_commits_committer_date ON commits(committer_date);

CREATE TABLE IF NOT EXISTS pulls (
    number INTEGER PRIMARY KEY,
    state TEXT,
    merged INTEGER,
    created_at TEXT,
    updated_at TEXT,
    closed_at TEXT,
    merged_at TEXT,
    base_ref TEXT,
    head_sha TEXT,
    merge_commit_sha TEXT,
    user_login TEXT,
    merged_by_login TEXT,
    additions INTEGER,
    deletions INTEGER,
    changed_files INTEGER,
    labels TEXT,
    has_detail INTEGER NOT NULL DEFAULT 0,
    raw TEXT NOT NULL
);
CREATE INDEX IF NOT EXISTS idx_pulls_closed_at ON pulls(closed_at);
CREATE INDEX IF NOT EXISTS idx_pulls_merged_at ON pulls(merged_at);
CREATE INDEX IF NOT EXISTS idx_pulls_created_at ON pulls(created_at);
CREATE INDEX IF NOT EXISTS idx_pulls_updated_at ON pulls(updated_at);
CREATE INDEX IF NOT EXISTS idx_pulls_head_sha ON pulls(head_sha);
CREATE INDEX IF NOT EXISTS idx_pulls_merge_commit_sha ON pulls(merge_commit_sha);
CREATE INDEX IF NOT EXISTS idx_pulls_state_merged ON pulls(state, merged);

CREATE TABLE IF NOT EXISTS reachability (
    merge_sha TEXT PRIMARY KEY,
    reachable INTEGER NOT NULL,
    checked_at TEXT NOT NULL,
    master_head_sha TEXT
);

CREATE TABLE IF NOT EXISTS sync_meta (
    key TEXT PRIMARY KEY,
    value TEXT
);

-- ─── Tier 1B telemetry tables ────────────────────────────────────────────
--
-- merge_cycle: one row per merge driver attempt (single-PR or train-merge).
--
--   Phase columns are wall-clock seconds; NULL means the phase did not run
--   for this terminal_state. Action columns let the canvas surface "what
--   the driver actually did" without having to grep logs.
CREATE TABLE IF NOT EXISTS merge_cycle (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    started_at      TEXT NOT NULL,
    ended_at        TEXT,
    pr_numbers      TEXT NOT NULL,        -- JSON array of PR numbers
    train_id        TEXT,                 -- e.g. "auto/train/<ts>" or NULL for single-PR
    bisect_depth    INTEGER NOT NULL DEFAULT 0,
    rebase_seconds  REAL,
    ci_seconds      REAL,
    merge_seconds   REAL,
    total_seconds   REAL,
    terminal_state  TEXT NOT NULL,        -- merged | merged_in_train | rebase-conflict-vs-master
                                          -- | ci-fail-on-rebased-sha | ci-timeout | bisect-budget-exhausted
                                          -- | restart-budget-exhausted | umbrella-pr-creation-failed
                                          -- | merge-409-race | crashed
    action_taken    TEXT,                 -- short slug e.g. "merged"|"released"|"escalated"
    action_detail   TEXT,                 -- free-form JSON or string with details
    last_released_at TEXT,                -- when the PR(s) was released back to the pool
    flake_rate      REAL,                 -- forward-fill from ci_gate_events for this cycle, if computed
    raw             TEXT                  -- optional JSON blob with extra fields for debug
);
CREATE INDEX IF NOT EXISTS idx_merge_cycle_started_at  ON merge_cycle(started_at);
CREATE INDEX IF NOT EXISTS idx_merge_cycle_ended_at    ON merge_cycle(ended_at);
CREATE INDEX IF NOT EXISTS idx_merge_cycle_terminal    ON merge_cycle(terminal_state);
CREATE INDEX IF NOT EXISTS idx_merge_cycle_pr_numbers  ON merge_cycle(pr_numbers);

-- ci_gate_events: one row per individual CI gate observation (per-PR per-run).
--
--   gate is the status_check_context name (e.g. "CI / lint*"), outcome is
--   "passed" | "failed" | "timeout" | "skipped". Used for flake-rate, gate
--   fall-through analysis, and feeding the future T5A duplicate-failure
--   clusterer.
CREATE TABLE IF NOT EXISTS ci_gate_events (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    observed_at TEXT NOT NULL,
    pr_number   INTEGER,
    head_sha    TEXT NOT NULL,
    gate        TEXT NOT NULL,
    outcome     TEXT NOT NULL,           -- passed | failed | timeout | skipped
    duration_s  REAL,
    cycle_id    INTEGER,                  -- FK to merge_cycle.id (nullable)
    raw         TEXT                      -- optional log/error excerpt
);
CREATE INDEX IF NOT EXISTS idx_ci_gate_events_observed ON ci_gate_events(observed_at);
CREATE INDEX IF NOT EXISTS idx_ci_gate_events_gate     ON ci_gate_events(gate);
CREATE INDEX IF NOT EXISTS idx_ci_gate_events_outcome  ON ci_gate_events(outcome);
CREATE INDEX IF NOT EXISTS idx_ci_gate_events_head_sha ON ci_gate_events(head_sha);
CREATE INDEX IF NOT EXISTS idx_ci_gate_events_pr       ON ci_gate_events(pr_number);
CREATE INDEX IF NOT EXISTS idx_ci_gate_events_cycle    ON ci_gate_events(cycle_id);

-- llm_activity: one row per LLM call by any agent in the auto-agents pipeline.
--
--   Both for cost accounting (T1A/T1C/T2C in the token-reduction plan) and
--   for activity-stream rendering. tokens_in/tokens_out/cached_tokens are
--   denormalised here for fast aggregation.
CREATE TABLE IF NOT EXISTS llm_activity (
    id              INTEGER PRIMARY KEY AUTOINCREMENT,
    started_at      TEXT NOT NULL,
    ended_at        TEXT,
    agent           TEXT NOT NULL,        -- e.g. "pr-review-worker", "task-implementor"
    pr_number       INTEGER,
    session_tag     TEXT,                 -- AUTO-* session tag, when applicable
    model           TEXT NOT NULL,
    tokens_in       INTEGER,
    tokens_out      INTEGER,
    cached_tokens   INTEGER,
    decision        TEXT,                 -- e.g. "approved", "request_changes", "ci_fix_pushed"
    outcome         TEXT,                 -- e.g. "success", "escalated", "timeout"
    cycle_id        INTEGER,              -- FK to merge_cycle.id (nullable)
    raw             TEXT                  -- optional JSON blob (full request/response trace)
);
CREATE INDEX IF NOT EXISTS idx_llm_activity_started   ON llm_activity(started_at);
CREATE INDEX IF NOT EXISTS idx_llm_activity_agent     ON llm_activity(agent);
CREATE INDEX IF NOT EXISTS idx_llm_activity_pr        ON llm_activity(pr_number);
CREATE INDEX IF NOT EXISTS idx_llm_activity_model     ON llm_activity(model);
CREATE INDEX IF NOT EXISTS idx_llm_activity_outcome   ON llm_activity(outcome);
CREATE INDEX IF NOT EXISTS idx_llm_activity_cycle     ON llm_activity(cycle_id);

-- ─── conflict_drive_cycles (Tier 1.5 — conflict driver telemetry) ────────
--
-- One row per (cycle, PR). The cycle-level totals (candidates_count,
-- resolved_count, escalated_count, timeout_count, push_rejected_count)
-- are duplicated across all PR rows from the same cycle for cheap
-- per-cycle aggregation; the per-PR specifics live in details_json.
--
-- ``outcome`` and ``failure_kind`` are promoted to top-level columns so
-- the retry-budget query at § 5.1 of conflict-drive-plan.md can use the
-- ``(pr_number, started_at)`` index directly without parsing JSON.
CREATE TABLE IF NOT EXISTS conflict_drive_cycles (
    cycle_id            TEXT NOT NULL,           -- shared across rows from one cycle
    pr_number           INTEGER NOT NULL,        -- top-level for indexed lookups
    started_at          TEXT NOT NULL,
    ended_at            TEXT NOT NULL,
    outcome             TEXT NOT NULL,           -- 'resolved' | 'resolved-no-conflict'
                                                 -- | 'unresolvable' | 'verification-fail'
                                                 -- | 'rebase-failed' | 'timeout'
                                                 -- | 'transport-error' | 'lease-violation'
    failure_kind        TEXT,                    -- 'definite' | 'transient' | NULL on success
    candidates_count    INTEGER NOT NULL,        -- cycle-level totals duplicated across PR rows
    resolved_count      INTEGER NOT NULL,
    escalated_count     INTEGER NOT NULL,
    timeout_count       INTEGER NOT NULL,
    push_rejected_count INTEGER NOT NULL,
    details_json        TEXT NOT NULL,           -- per-PR breakdown
    PRIMARY KEY (cycle_id, pr_number)
);
CREATE INDEX IF NOT EXISTS idx_cdc_pr_number_started ON conflict_drive_cycles(pr_number, started_at);
CREATE INDEX IF NOT EXISTS idx_cdc_started_at        ON conflict_drive_cycles(started_at);
"""


class ForgejoCache:
    """SQLite-backed cache with delta sync."""

    def __init__(self, db_path: Path):
        self.db_path = Path(db_path)
        self.db_path.parent.mkdir(parents=True, exist_ok=True)
        # timeout=30s raises SQLite's busy-timeout so writers wait/retry when
        # a sibling process (e.g. a subprocess opening the same DB) momentarily
        # holds a lock, instead of failing immediately. Python's default is 5s,
        # which isn't enough when a parent renderer spawns count-master-merges.
        self._conn = sqlite3.connect(str(self.db_path), timeout=30.0)
        self._conn.row_factory = sqlite3.Row
        self._conn.execute("PRAGMA journal_mode = WAL")
        self._conn.execute("PRAGMA synchronous = NORMAL")
        # Belt-and-suspenders: the PRAGMA is redundant with timeout=30.0 above,
        # but explicit documents intent and is useful if someone opens the
        # connection differently in the future.
        self._conn.execute("PRAGMA busy_timeout = 30000")
        self._conn.executescript(SCHEMA_SQL)
        self._migrate()

    @classmethod
    def open(cls, db_path: Path | None = None) -> "ForgejoCache":
        return cls(db_path or DEFAULT_CACHE_PATH)

    # ─── Schema management ──────────────────────────────────────────────

    def _migrate(self) -> None:
        row = self._conn.execute("SELECT MAX(version) AS v FROM schema_version").fetchone()
        current = row["v"] if row and row["v"] is not None else 0
        # The new tables in SCHEMA_SQL are all CREATE TABLE IF NOT EXISTS, so
        # they are added on first ``__init__`` after upgrade. Any future
        # migration that mutates an existing table goes here, gated on
        # ``current < N``.
        if current < SCHEMA_VERSION:
            self._conn.execute(
                "INSERT INTO schema_version(version) VALUES (?)", (SCHEMA_VERSION,)
            )
            self._conn.commit()

    # ─── Tier 1B telemetry helpers ─────────────────────────────────────

    def record_merge_cycle(self, cycle: dict[str, Any]) -> int:
        """Insert a ``merge_cycle`` row. Returns the new row id.

        ``cycle`` is a dict matching the table columns (extra keys are
        captured into the ``raw`` JSON blob). Used by the merge driver to
        record one full attempt (single-PR or train-merge).
        """
        cols = (
            "started_at",
            "ended_at",
            "pr_numbers",
            "train_id",
            "bisect_depth",
            "rebase_seconds",
            "ci_seconds",
            "merge_seconds",
            "total_seconds",
            "terminal_state",
            "action_taken",
            "action_detail",
            "last_released_at",
            "flake_rate",
        )
        # Anything not in cols goes into raw for forensics.
        raw_blob = {k: v for k, v in cycle.items() if k not in cols}
        values = [cycle.get(c) for c in cols]
        # pr_numbers MUST be a JSON-encoded array of ints — coerce here so
        # callers can pass either a list or a string.
        prs = values[cols.index("pr_numbers")]
        if isinstance(prs, (list, tuple)):
            values[cols.index("pr_numbers")] = json.dumps(list(prs))
        elif prs is None:
            values[cols.index("pr_numbers")] = "[]"
        # bisect_depth is NOT NULL DEFAULT 0 in the schema; supply 0 when the
        # caller doesn't pass it so the explicit-NULL doesn't violate the
        # constraint.
        if values[cols.index("bisect_depth")] is None:
            values[cols.index("bisect_depth")] = 0
        cur = self._conn.execute(
            f"INSERT INTO merge_cycle ({','.join(cols)}, raw) "
            f"VALUES ({','.join('?' for _ in cols)}, ?)",
            (*values, json.dumps(raw_blob) if raw_blob else None),
        )
        self._conn.commit()
        rid = cur.lastrowid
        assert rid is not None
        return rid

    def record_ci_gate_event(self, event: dict[str, Any]) -> int:
        """Insert one ``ci_gate_events`` row. Commits immediately — use
        :meth:`record_ci_gate_events_batch` for high-volume callers (the
        merge driver collects all 6 gate outcomes per PR ramp-up; an
        eager fsync per row is wasteful).
        """
        return self.record_ci_gate_events_batch([event])[-1]

    def record_ci_gate_events_batch(self, events: list[dict[str, Any]]) -> list[int]:
        """Insert N ``ci_gate_events`` rows in a single transaction (one
        fsync rather than N). Returns the list of new row ids in input
        order."""
        if not events:
            return []
        cols = (
            "observed_at",
            "pr_number",
            "head_sha",
            "gate",
            "outcome",
            "duration_s",
            "cycle_id",
        )
        rows: list[tuple] = []
        for event in events:
            raw_blob = {k: v for k, v in event.items() if k not in cols}
            rows.append(
                (
                    *[event.get(c) for c in cols],
                    json.dumps(raw_blob) if raw_blob else None,
                )
            )
        sql = (
            f"INSERT INTO ci_gate_events ({','.join(cols)}, raw) "
            f"VALUES ({','.join('?' for _ in cols)}, ?)"
        )
        ids: list[int] = []
        # executemany commits implicitly only inside a ``with self._conn``
        # block; we want a single explicit commit at the end so partial
        # batches don't half-write on error.
        try:
            for row in rows:
                cur = self._conn.execute(sql, row)
                rid = cur.lastrowid
                assert rid is not None
                ids.append(rid)
            self._conn.commit()
        except Exception:
            self._conn.rollback()
            raise
        return ids

    def record_llm_activity(self, activity: dict[str, Any]) -> int:
        """Insert one ``llm_activity`` row.

        Used by both the cost-tracking dashboards (1B) and the agent
        activity stream (Phase 2 deferred item) downstream of this. Callers
        emit one row per LLM invocation.
        """
        cols = (
            "started_at",
            "ended_at",
            "agent",
            "pr_number",
            "session_tag",
            "model",
            "tokens_in",
            "tokens_out",
            "cached_tokens",
            "decision",
            "outcome",
            "cycle_id",
        )
        raw_blob = {k: v for k, v in activity.items() if k not in cols}
        cur = self._conn.execute(
            f"INSERT INTO llm_activity ({','.join(cols)}, raw) "
            f"VALUES ({','.join('?' for _ in cols)}, ?)",
            (
                *[activity.get(c) for c in cols],
                json.dumps(raw_blob) if raw_blob else None,
            ),
        )
        self._conn.commit()
        rid = cur.lastrowid
        assert rid is not None
        return rid

    def merge_cycles_in_window(
        self, start: datetime, end: datetime
    ) -> list[dict[str, Any]]:
        """Return ``merge_cycle`` rows whose ``started_at`` is within
        [start, end). Used by the pr-velocity canvas renderer."""
        rows = self._conn.execute(
            "SELECT * FROM merge_cycle "
            "WHERE started_at >= ? AND started_at < ? "
            "ORDER BY started_at DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [dict(r) for r in rows]

    def ci_gate_events_in_window(
        self, start: datetime, end: datetime
    ) -> list[dict[str, Any]]:
        rows = self._conn.execute(
            "SELECT * FROM ci_gate_events "
            "WHERE observed_at >= ? AND observed_at < ? "
            "ORDER BY observed_at DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [dict(r) for r in rows]

    def llm_activity_in_window(
        self, start: datetime, end: datetime
    ) -> list[dict[str, Any]]:
        rows = self._conn.execute(
            "SELECT * FROM llm_activity "
            "WHERE started_at >= ? AND started_at < ? "
            "ORDER BY started_at DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [dict(r) for r in rows]

    # ─── conflict_drive_cycles helpers ─────────────────────────────────

    def record_conflict_drive_cycle(self, row: dict[str, Any]) -> None:
        """Insert one ``conflict_drive_cycles`` row.

        ``row`` must carry every NOT NULL column. ``details_json`` is
        encoded if a dict is passed; if the caller already JSON-encoded
        it, the string is stored as-is. ``outcome`` and ``failure_kind``
        live both as top-level indexed columns and inside ``details_json``;
        callers MUST keep them in sync (a single
        ``record_telemetry`` helper in ``conflict_drive.py`` is the
        intended source of truth — see plan § 9).
        """
        details = row.get("details_json")
        if isinstance(details, (dict, list)):
            details_str = json.dumps(details)
        else:
            details_str = details if details is not None else "{}"
        cols = (
            "cycle_id",
            "pr_number",
            "started_at",
            "ended_at",
            "outcome",
            "failure_kind",
            "candidates_count",
            "resolved_count",
            "escalated_count",
            "timeout_count",
            "push_rejected_count",
            "details_json",
        )
        values = list(row.get(c) for c in cols)
        values[cols.index("details_json")] = details_str
        # Use INSERT OR REPLACE so a same-cycle re-record (e.g. a retry
        # mid-cycle) doesn't trip the PRIMARY KEY (cycle_id, pr_number).
        # Cycle ids are intended to be unique-per-cycle; replace is the
        # safe default if a caller emits twice by accident.
        self._conn.execute(
            f"INSERT OR REPLACE INTO conflict_drive_cycles "
            f"({','.join(cols)}) VALUES ({','.join('?' for _ in cols)})",
            values,
        )
        self._conn.commit()

    def mark_conflict_drive_cycle_escalated(
        self, cycle_id: str, pr_number: int
    ) -> int:
        """Set ``escalated_count = 1`` on the row keyed by
        ``(cycle_id, pr_number)``.

        Telemetry semantics (plan § 9): ``escalated_count`` is a
        cycle-level event — it fires exactly once across the rolling
        24-hour retry budget when ``conflict_drive.run_one_cycle``
        decides the budget is exhausted and applies
        ``auto/needs-implementer``. ``record_conflict_drive_cycle``
        always writes ``escalated_count = 0`` because at row-write
        time the budget verdict has not yet been computed (the
        verdict needs the row that's being written). This helper is
        the second-phase update; calling it twice for the same row
        is idempotent.

        Returns the number of rows updated. Zero rows means the
        telemetry write earlier in the cycle either failed or used a
        different ``cycle_id`` — the caller may use this to decide
        whether to log a diagnostic. The caller is responsible for
        any logging; this helper stays silent so the cache module
        does not depend on a specific logger.
        """
        cur = self._conn.execute(
            "UPDATE conflict_drive_cycles SET escalated_count = 1 "
            "WHERE cycle_id = ? AND pr_number = ?",
            (cycle_id, pr_number),
        )
        self._conn.commit()
        return cur.rowcount or 0

    def count_conflict_drive_definite_failures_24h(
        self, pr_number: int, *, now: datetime | None = None
    ) -> int:
        """Return the number of ``failure_kind='definite'`` rows for
        ``pr_number`` whose ``started_at`` is within the trailing 24 h
        window. This is the retry-budget query at plan § 5.1 — used by
        ``conflict_drive.resolve_one`` to decide whether the next failure
        triggers escalation to ``auto/needs-implementer``.

        Uses the ``idx_cdc_pr_number_started`` index for an O(log N) lookup.
        ``now`` is injectable so unit tests can pin the window without
        time-traveling the wall clock.
        """
        cutoff = (now or datetime.now(timezone.utc)) - timedelta(hours=24)
        row = self._conn.execute(
            "SELECT COUNT(*) AS n FROM conflict_drive_cycles "
            "WHERE pr_number = ? AND failure_kind = 'definite' "
            "AND started_at > ?",
            (pr_number, cutoff.isoformat()),
        ).fetchone()
        return int(row["n"]) if row else 0

    def conflict_drive_cycles_in_window(
        self, start: datetime, end: datetime
    ) -> list[dict[str, Any]]:
        """Return ``conflict_drive_cycles`` rows whose ``started_at`` is
        within [start, end). Used by ``tools/render-pr-velocity.py`` for
        the "Conflict-resolution activity" section.

        Uses the ``idx_cdc_started_at`` index for an O(log N) seek.
        """
        rows = self._conn.execute(
            "SELECT * FROM conflict_drive_cycles "
            "WHERE started_at >= ? AND started_at < ? "
            "ORDER BY started_at DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [dict(r) for r in rows]

    def last_conflict_drive_cycle_for_pr(
        self, pr_number: int
    ) -> dict[str, Any] | None:
        """Return the most recent ``conflict_drive_cycles`` row for
        ``pr_number`` (highest ``started_at``), or ``None`` if no row
        exists.

        Used by ``tools/conflict_drive.py`` for the per-cycle cooldown
        check. Uses the ``idx_cdc_pr_number_started`` index, so the
        lookup is O(log N) — important because this fires once per
        candidate per cycle.
        """
        row = self._conn.execute(
            "SELECT * FROM conflict_drive_cycles "
            "WHERE pr_number = ? "
            "ORDER BY started_at DESC LIMIT 1",
            (pr_number,),
        ).fetchone()
        return dict(row) if row else None

    def last_released_at_for_pr(self, pr_number: int) -> str | None:
        """Most recent ``last_released_at`` from merge_cycle for one PR.

        Used by the merge driver's candidate filter to enforce the
        cooldown window without depending on labels (S4 simplification:
        labels persist for visibility, but eligibility is computed from
        telemetry).
        """
        # Pull cycles that mention this PR and sort by ended_at desc.
        # SQLite has no native JSON containment, so we use a LIKE filter
        # and then verify in Python (cheap; merge_cycle is small).
        rows = self._conn.execute(
            "SELECT pr_numbers, last_released_at FROM merge_cycle "
            "WHERE last_released_at IS NOT NULL "
            "  AND pr_numbers LIKE ? "
            "ORDER BY last_released_at DESC LIMIT 50",
            (f"%{pr_number}%",),
        ).fetchall()
        for r in rows:
            try:
                prs = json.loads(r["pr_numbers"])
            except (TypeError, ValueError):
                continue
            if pr_number in prs:
                return r["last_released_at"]
        return None

    # ─── sync_meta helpers ──────────────────────────────────────────────

    def _meta_get(self, key: str) -> str | None:
        row = self._conn.execute("SELECT value FROM sync_meta WHERE key = ?", (key,)).fetchone()
        return row["value"] if row else None

    def _meta_set(self, key: str, value: str) -> None:
        self._conn.execute(
            "INSERT INTO sync_meta(key, value) VALUES(?, ?) "
            "ON CONFLICT(key) DO UPDATE SET value = excluded.value",
            (key, value),
        )

    # ─── Locking ────────────────────────────────────────────────────────

    def _lock(self):
        """fcntl.flock on ``<db_path>.lock`` for exclusive sync."""
        lock_path = Path(str(self.db_path) + ".lock")
        lock_path.touch()
        fd = os.open(str(lock_path), os.O_RDWR)
        fcntl.flock(fd, fcntl.LOCK_EX)
        return fd

    @staticmethod
    def _unlock(fd: int) -> None:
        try:
            fcntl.flock(fd, fcntl.LOCK_UN)
        finally:
            os.close(fd)

    # ─── Commits sync ───────────────────────────────────────────────────

    def _upsert_commit(self, c: dict) -> bool:
        """Insert a commit if new. Returns True if a new row was inserted."""
        sha = c.get("sha")
        if not sha:
            return False
        row = self._conn.execute(
            "INSERT OR IGNORE INTO commits "
            "(sha, committer_date, author_name, committer_name, message, parents, raw) "
            "VALUES (?, ?, ?, ?, ?, ?, ?)",
            (
                sha,
                (c.get("commit", {}).get("committer", {}) or {}).get("date"),
                (c.get("commit", {}).get("author", {}) or {}).get("name"),
                (c.get("commit", {}).get("committer", {}) or {}).get("name"),
                (c.get("commit", {}) or {}).get("message"),
                json.dumps([p.get("sha") for p in (c.get("parents") or [])]),
                json.dumps(c),
            ),
        )
        return row.rowcount > 0

    def _sync_commits(self, token: str, full: bool, progress: bool) -> dict:
        """Pull master commits forward from HEAD until we reach cached content.

        When ``full=True`` the cache is truncated first (use after a force-push
        or to rebuild from scratch).
        """
        if full:
            self._conn.execute("DELETE FROM commits")
            self._conn.commit()

        prev_head = self._meta_get("master_head_sha")
        known_shas = {
            row["sha"] for row in self._conn.execute("SELECT sha FROM commits")
        }
        new_count = 0
        consecutive_all_cached = 0
        first_sha: str | None = None
        saw_prev_head = False

        for page in range(1, SYNC_PAGE_HARD_LIMIT + 1):
            commits = api_get(
                f"{API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/commits"
                f"?sha=master&limit=50&page={page}",
                token,
            )
            if not commits:
                break
            if first_sha is None and commits:
                first_sha = commits[0].get("sha")
            page_new = 0
            for c in commits:
                sha = c.get("sha")
                if sha == prev_head:
                    saw_prev_head = True
                if sha in known_shas:
                    continue
                if self._upsert_commit(c):
                    known_shas.add(sha)
                    new_count += 1
                    page_new += 1
            if progress and (page == 1 or page % 10 == 0):
                print(
                    f"#   commits: page {page}, +{page_new} new (running total {new_count})",
                    file=sys.stderr,
                )
            # Stop when we've hit a page that's entirely old content (meaning
            # we've caught up to the cache's existing HEAD).
            if page_new == 0:
                consecutive_all_cached += 1
                if consecutive_all_cached >= SYNC_EMPTY_PAGE_TOLERANCE:
                    break
            else:
                consecutive_all_cached = 0
            if len(commits) < 50:
                break

        self._conn.commit()
        if first_sha:
            self._meta_set("master_head_sha", first_sha)
            self._meta_set("last_commit_sync_at", datetime.now(timezone.utc).isoformat())

        return {
            "new_commits": new_count,
            "master_head_sha": first_sha,
            "force_push_suspected": (
                prev_head is not None and not saw_prev_head and new_count > 50
            ),
        }

    # ─── PR sync ────────────────────────────────────────────────────────

    def _upsert_pr(self, pr: dict, has_detail: bool | None = None) -> bool:
        """Insert or update a PR. Returns True if row was new or updated_at advanced."""
        num = pr.get("number")
        if num is None:
            return False
        existing = self._conn.execute(
            "SELECT updated_at, has_detail FROM pulls WHERE number = ?", (num,),
        ).fetchone()
        new_updated = pr.get("updated_at")
        if existing is not None and existing["updated_at"] == new_updated:
            # No change; but we may have just pulled detail — update has_detail if so.
            if has_detail and not existing["has_detail"]:
                self._conn.execute(
                    "UPDATE pulls SET has_detail = 1, merged_by_login = ?, "
                    "additions = ?, deletions = ?, changed_files = ?, raw = ? "
                    "WHERE number = ?",
                    (
                        (pr.get("merged_by") or {}).get("login"),
                        pr.get("additions"),
                        pr.get("deletions"),
                        pr.get("changed_files"),
                        json.dumps(pr),
                        num,
                    ),
                )
                return True
            return False

        self._conn.execute(
            "INSERT INTO pulls "
            "(number, state, merged, created_at, updated_at, closed_at, merged_at, "
            " base_ref, head_sha, merge_commit_sha, user_login, merged_by_login, "
            " additions, deletions, changed_files, labels, has_detail, raw) "
            "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) "
            "ON CONFLICT(number) DO UPDATE SET "
            "  state = excluded.state, merged = excluded.merged, "
            "  created_at = excluded.created_at, updated_at = excluded.updated_at, "
            "  closed_at = excluded.closed_at, merged_at = excluded.merged_at, "
            "  base_ref = excluded.base_ref, head_sha = excluded.head_sha, "
            "  merge_commit_sha = excluded.merge_commit_sha, "
            "  user_login = excluded.user_login, "
            "  merged_by_login = COALESCE(excluded.merged_by_login, pulls.merged_by_login), "
            "  additions = COALESCE(excluded.additions, pulls.additions), "
            "  deletions = COALESCE(excluded.deletions, pulls.deletions), "
            "  changed_files = COALESCE(excluded.changed_files, pulls.changed_files), "
            "  labels = excluded.labels, "
            "  has_detail = MAX(excluded.has_detail, pulls.has_detail), "
            "  raw = excluded.raw",
            (
                num,
                pr.get("state"),
                1 if pr.get("merged") else 0,
                pr.get("created_at"),
                pr.get("updated_at"),
                pr.get("closed_at"),
                pr.get("merged_at"),
                (pr.get("base") or {}).get("ref"),
                (pr.get("head") or {}).get("sha"),
                pr.get("merge_commit_sha"),
                (pr.get("user") or {}).get("login"),
                (pr.get("merged_by") or {}).get("login"),
                pr.get("additions"),
                pr.get("deletions"),
                pr.get("changed_files"),
                json.dumps([lbl.get("name") for lbl in (pr.get("labels") or [])]),
                1 if has_detail else 0,
                json.dumps(pr),
            ),
        )
        return True

    def _sync_pulls(self, token: str, full: bool, progress: bool) -> dict:
        """Sync all PRs by walking ``pulls?state=all&sort=recentupdate``.

        Stops when we see a page whose most-recent updated_at is older than
        our last-sync cursor (everything older is already cached and stable).
        """
        if full:
            self._conn.execute("DELETE FROM pulls")
            self._conn.commit()

        last_sync = self._meta_get("last_pull_sync_updated_at")
        if full or last_sync is None:
            last_sync_dt = None
        else:
            last_sync_dt = last_sync  # string comparison is safe on ISO 8601

        upserts = 0
        max_updated: str | None = None
        consecutive_stale_pages = 0
        for page in range(1, SYNC_PAGE_HARD_LIMIT + 1):
            prs = api_get(
                f"{API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/pulls"
                f"?state=all&sort=recentupdate&limit=50&page={page}",
                token,
            )
            if not prs:
                break
            page_new_or_updated = 0
            page_min_updated: str | None = None
            for pr in prs:
                upd = pr.get("updated_at")
                if upd is not None:
                    if page_min_updated is None or upd < page_min_updated:
                        page_min_updated = upd
                    if max_updated is None or upd > max_updated:
                        max_updated = upd
                if last_sync_dt is not None and upd is not None and upd <= last_sync_dt:
                    continue
                if self._upsert_pr(pr):
                    upserts += 1
                    page_new_or_updated += 1

            if progress and (page == 1 or page % 20 == 0):
                print(
                    f"#   pulls: page {page}, +{page_new_or_updated} upserts "
                    f"(running total {upserts})",
                    file=sys.stderr,
                )

            # Once the oldest updated_at on this page is older than our last
            # sync cursor, any subsequent page is entirely older still.
            if (
                last_sync_dt is not None
                and page_min_updated is not None
                and page_min_updated <= last_sync_dt
            ):
                break

            if page_new_or_updated == 0:
                consecutive_stale_pages += 1
                if consecutive_stale_pages >= SYNC_EMPTY_PAGE_TOLERANCE:
                    break
            else:
                consecutive_stale_pages = 0

            if len(prs) < 50:
                break

        self._conn.commit()
        if max_updated:
            self._meta_set("last_pull_sync_updated_at", max_updated)
        self._meta_set("last_pull_sync_at", datetime.now(timezone.utc).isoformat())
        return {"pulls_upserted": upserts, "max_updated_at": max_updated}

    # ─── PR enrichment (lazy, on read) ──────────────────────────────────

    def enrich_pr(self, number: int, token: str) -> dict:
        """Return the full PR dict, fetching detail on demand.

        The list endpoint omits ``merged_by``, ``additions``, ``deletions``,
        and ``changed_files``. We fetch the detail once per PR and persist it.
        """
        row = self._conn.execute(
            "SELECT raw, has_detail FROM pulls WHERE number = ?", (number,),
        ).fetchone()
        if row is None:
            # Not in cache at all — fetch and store.
            pr = api_get(
                f"{API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/pulls/{number}", token,
            )
            self._upsert_pr(pr, has_detail=True)
            self._conn.commit()
            return pr
        if row["has_detail"]:
            return json.loads(row["raw"])
        # Need to fetch detail.
        pr = api_get(
            f"{API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/pulls/{number}", token,
        )
        self._upsert_pr(pr, has_detail=True)
        self._conn.commit()
        return pr

    def backfill_merged_by(self, token: str, progress: bool = True) -> int:
        """One-shot: fetch detail for every in-cache closed+merged PR that
        lacks ``merged_by_login``. Called optionally after a fresh sync to
        warm the merged-by cache. Returns number of PRs enriched."""
        rows = self._conn.execute(
            "SELECT number FROM pulls "
            "WHERE state = 'closed' AND merged = 1 AND merged_by_login IS NULL "
            "AND has_detail = 0"
        ).fetchall()
        n = 0
        for i, row in enumerate(rows):
            self.enrich_pr(row["number"], token)
            n += 1
            if progress and (i + 1) % 50 == 0:
                print(f"#   merged_by backfill: {i+1}/{len(rows)}", file=sys.stderr)
        return n

    # ─── Reachability (compare endpoint) ────────────────────────────────

    def is_sha_reachable_from_master(
        self, sha: str, token: str, *, master_head_sha: str | None = None,
    ) -> bool:
        """Cached reachability check.

        Cache key is ``(merge_sha, master_head_sha)``; a positive result is
        permanent (a commit that was ever an ancestor of master stays so,
        absent a force-push), but negative results are re-checked if the
        master HEAD has advanced.
        """
        if not sha:
            return False
        row = self._conn.execute(
            "SELECT reachable, master_head_sha FROM reachability WHERE merge_sha = ?",
            (sha,),
        ).fetchone()
        if row is not None:
            if row["reachable"]:
                return True
            # Negative result: only trust it if master HEAD hasn't changed.
            cur_head = master_head_sha or self._meta_get("master_head_sha")
            if cur_head and row["master_head_sha"] == cur_head:
                return False
        # Miss or stale negative — do the API call.
        url = f"{API_BASE}/repos/{REPO_OWNER}/{REPO_NAME}/compare/master...{sha}"
        try:
            data = api_get(url, token)
        except urllib.error.HTTPError as e:
            if e.code == 404:
                reachable = False
            else:
                raise
        else:
            total = data.get("total_commits")
            if total is None:
                total = len(data.get("commits") or [])
            reachable = total == 0
        cur_head = master_head_sha or self._meta_get("master_head_sha")
        self._conn.execute(
            "INSERT INTO reachability(merge_sha, reachable, checked_at, master_head_sha) "
            "VALUES(?, ?, ?, ?) "
            "ON CONFLICT(merge_sha) DO UPDATE SET "
            "  reachable = excluded.reachable, "
            "  checked_at = excluded.checked_at, "
            "  master_head_sha = excluded.master_head_sha",
            (sha, 1 if reachable else 0, datetime.now(timezone.utc).isoformat(), cur_head),
        )
        self._conn.commit()
        return reachable

    # ─── Top-level sync ─────────────────────────────────────────────────

    def sync(
        self,
        token: str,
        *,
        full: bool = False,
        backfill_details: bool = False,
        progress: bool = True,
    ) -> dict:
        """Delta-sync the cache. Returns a summary dict."""
        lock_fd = self._lock()
        try:
            if progress:
                print("# syncing commits ...", file=sys.stderr)
            commits_summary = self._sync_commits(token, full, progress)
            if commits_summary.get("force_push_suspected"):
                print(
                    "# WARN: previous master HEAD not seen in recent commits — "
                    "possible force-push. Re-run with --full to rebuild from scratch.",
                    file=sys.stderr,
                )
            if progress:
                print("# syncing pulls ...", file=sys.stderr)
            pulls_summary = self._sync_pulls(token, full, progress)
            enriched = 0
            if backfill_details:
                if progress:
                    print("# back-filling merged_by for merged PRs ...", file=sys.stderr)
                enriched = self.backfill_merged_by(token, progress=progress)
            return {
                "commits": commits_summary,
                "pulls": pulls_summary,
                "enriched_prs": enriched,
            }
        finally:
            self._unlock(lock_fd)

    # ─── Query API (all return raw API-shaped dicts) ────────────────────

    def commits_on_master(
        self, start: datetime, end: datetime,
    ) -> list[dict]:
        rows = self._conn.execute(
            "SELECT raw FROM commits "
            "WHERE committer_date >= ? AND committer_date < ? "
            "ORDER BY committer_date DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [json.loads(r["raw"]) for r in rows]

    def pulls_closed_in_window(self, start: datetime, end: datetime) -> list[dict]:
        rows = self._conn.execute(
            "SELECT raw FROM pulls "
            "WHERE state = 'closed' AND closed_at IS NOT NULL "
            "  AND closed_at >= ? AND closed_at < ? "
            "ORDER BY closed_at DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [json.loads(r["raw"]) for r in rows]

    def pulls_merged_in_window(self, start: datetime, end: datetime) -> list[dict]:
        rows = self._conn.execute(
            "SELECT raw FROM pulls "
            "WHERE merged = 1 AND merged_at IS NOT NULL "
            "  AND merged_at >= ? AND merged_at < ? "
            "ORDER BY merged_at DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [json.loads(r["raw"]) for r in rows]

    def pulls_opened_in_window(self, start: datetime, end: datetime) -> list[dict]:
        rows = self._conn.execute(
            "SELECT raw FROM pulls "
            "WHERE created_at >= ? AND created_at < ? "
            "ORDER BY created_at DESC",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        return [json.loads(r["raw"]) for r in rows]

    def sha_to_pr_map(self, start: datetime, end: datetime) -> dict[str, dict]:
        """SHA -> PR dict for PRs merged in window (indexes both head.sha
        and merge_commit_sha), for count-master-merges Phase 1 cross-reference.
        """
        rows = self._conn.execute(
            "SELECT raw, head_sha, merge_commit_sha FROM pulls "
            "WHERE merged = 1 AND merged_at IS NOT NULL "
            "  AND merged_at >= ? AND merged_at < ?",
            (start.isoformat(), end.isoformat()),
        ).fetchall()
        out: dict[str, dict] = {}
        for r in rows:
            pr = json.loads(r["raw"])
            if r["head_sha"]:
                out[r["head_sha"]] = pr
            if r["merge_commit_sha"]:
                out[r["merge_commit_sha"]] = pr
        return out

    # ─── Introspection helpers ──────────────────────────────────────────

    def stats(self) -> dict:
        c = self._conn
        row = c.execute("SELECT COUNT(*) AS n, MIN(committer_date) AS oldest, "
                        "MAX(committer_date) AS newest FROM commits").fetchone()
        pull_row = c.execute(
            "SELECT COUNT(*) AS n, MIN(created_at) AS oldest_created, "
            "MAX(updated_at) AS latest_update FROM pulls"
        ).fetchone()
        reach_row = c.execute("SELECT COUNT(*) AS n FROM reachability").fetchone()
        return {
            "cache_path": str(self.db_path),
            "cache_size_bytes": self.db_path.stat().st_size if self.db_path.exists() else 0,
            "commits": {
                "count": row["n"],
                "oldest": row["oldest"],
                "newest": row["newest"],
            },
            "pulls": {
                "count": pull_row["n"],
                "oldest_created": pull_row["oldest_created"],
                "latest_update": pull_row["latest_update"],
            },
            "reachability_cached": reach_row["n"],
            "last_commit_sync_at": self._meta_get("last_commit_sync_at"),
            "last_pull_sync_at": self._meta_get("last_pull_sync_at"),
            "master_head_sha": self._meta_get("master_head_sha"),
        }

    def close(self) -> None:
        self._conn.close()