cleveragents-core/tools/_pr_comments_cache.py

"""Persistent on-disk cache for PR/issue comments with delta fetch.

What this is
------------

Live test 2026-05-13 showed every implementer cycle on PR #30
re-fetched 1340+ accumulated comments via the
``/repos/{owner}/{repo}/issues/{N}/comments`` endpoint, hit the
20-page pagination cap, truncated the most recent comments off the
end, and burned ~30s on the fetch alone. Across 5 attempts in a
single PR cycle that's ~150s of identical work plus stale data.

This module caches the comment list to disk and supports a delta
fetch via Forgejo's ``?since=<ISO timestamp>`` query parameter. The
common case becomes: 5-10 NEW comments since the last fetch (or
zero) rather than the full 1340.

Cache layout
------------

``/tmp/cleveragents-comment-cache/pr-{N}.json`` — per-PR JSON file.
Schema::

    {
      "schema_version": 1,
      "pr_number": 30,
      "fetched_at": "2026-05-13T10:00:00+00:00",
      "since_cursor": "2026-05-13T09:59:00+00:00",
      "comments": [
        {"id": 1, "created_at": "...", "body": "...", ...},
        ...
      ],
      "any_partial_fetch": false
    }

``fetched_at`` is the wall-clock time of the last write — it drives
the staleness check only. ``since_cursor`` is the ``created_at`` of
the newest cached comment; it is what we pass as ``?since=`` on the
next delta fetch, so the delta backfills forward from where the
cache actually ends. A page-capped seed holds only the OLDEST N
comments, so its newest cached comment is NOT the newest on the PR —
using a wall-clock ``?since=`` would skip the un-fetched middle
forever. The list is append-only by id. ``any_partial_fetch`` is
True when the cache was seeded from a page-capped fetch and has not
yet been backfilled to completeness by a later delta.

Invalidation
------------

Comments edited or deleted upstream are NOT detected in v1. Two
defenses:

- ``CACHE_MAX_STALENESS_S`` (default 24h) — when the cache is older
  than this, we discard it and do a full refetch. Bounds the
  staleness window.
- Force-refresh via ``invalidate(pr_number)`` for operator-side
  manual interventions.

Edits and deletions are rare in this codebase's PR workflow; the
v1 trade-off (skip detection, do periodic refresh) is intentional.

Concurrency
-----------

The cache file is written atomically (write-to-tmp + rename). Two
dispatchers reading the SAME PR's comments concurrently each
compute the same delta and the last writer wins — idempotent. A
modest amount of duplicate-fetch work is the cost; the alternative
(file locking) would serialise readers, which is worse.

The cache is shared between the reviewer and implementer
dispatchers because both fetch the same endpoint with the same
auth and both benefit equally from the cache.

Errors
------

Every code path is best-effort. Cache read failures fall through
to a full fetch. Delta fetch failures fall through to a full
fetch. Full fetch failures bubble up to the caller as ``(items,
completed=False)`` — same contract as ``_review_fetch._api_get_paginated``.
"""
from __future__ import annotations

import datetime as _dt
import json
import logging
import os
import sys
from pathlib import Path
from typing import Any

_TOOLS_DIR = str(Path(__file__).resolve().parent)
if _TOOLS_DIR not in sys.path:
    sys.path.insert(0, _TOOLS_DIR)
from _loader import (  # noqa: E402  type: ignore[import-not-found]
    load_sibling as _load_sibling,
)

_review_fetch = _load_sibling("_review_fetch", "_review_fetch.py")

_logger = logging.getLogger("pr_comments_cache")

SCHEMA_VERSION = 1

# Default cache directory. Override via env var for tests / archive
# sharing between dispatcher hosts.
_DEFAULT_CACHE_DIR = Path("/tmp/cleveragents-comment-cache")
_CACHE_DIR_ENV = "IMPLEMENTER_DISPATCHER_COMMENT_CACHE_DIR"

# 24 hours — when the cache is older than this we discard it and
# do a full refetch. Bounds the edit/delete staleness window
# without paying the full-fetch cost every cycle.
_DEFAULT_MAX_STALENESS_S = 86400
_STALENESS_ENV = "IMPLEMENTER_DISPATCHER_COMMENT_CACHE_MAX_STALENESS_S"

# Kill switch for tests + operators who want to bypass the cache
# entirely (e.g. while diagnosing a divergence between cached and
# upstream state).
_DISABLE_ENV = "IMPLEMENTER_DISPATCHER_COMMENT_CACHE_DISABLE"

# Exponential backoff for repeated live-delta failures (R1 +
# 2026-05-16). Live test on PR #29 saw the dispatcher hit the
# pagination cap ``max_pages=20`` six times in a single hour
# because every cycle re-attempted the comment fetch against the
# same flaky endpoint, each retry burning ~30s of the worker turn
# AND polluting the prompt with a fresh ``data_complete=False``
# signal. Backoff caps the retry rate: after each consecutive
# failure we defer the next live attempt by an exponentially
# growing delay (capped at ``_BACKOFF_MAX_S``) and serve the
# cached bulk stale in the meantime. Counter resets on the first
# successful delta — a transient outage doesn't permanently throttle
# the PR.
_BACKOFF_BASE_S = 60
_BACKOFF_MAX_S = 1800
_BACKOFF_BASE_ENV = "IMPLEMENTER_DISPATCHER_COMMENT_CACHE_BACKOFF_BASE_S"
_BACKOFF_MAX_ENV = "IMPLEMENTER_DISPATCHER_COMMENT_CACHE_BACKOFF_MAX_S"


def cache_dir() -> Path:
    """Resolve the on-disk cache directory."""
    return Path(os.environ.get(_CACHE_DIR_ENV) or str(_DEFAULT_CACHE_DIR))


def cache_path(pr_number: int) -> Path:
    """Per-PR cache file path."""
    return cache_dir() / f"pr-{int(pr_number)}.json"


def _max_staleness_s() -> int:
    raw = os.environ.get(_STALENESS_ENV)
    if raw:
        try:
            return max(0, int(raw))
        except ValueError:
            pass
    return _DEFAULT_MAX_STALENESS_S


def is_disabled() -> bool:
    raw = os.environ.get(_DISABLE_ENV, "").strip().lower()
    return raw in {"1", "true", "yes", "on"}


def _now() -> str:
    return _dt.datetime.now(_dt.timezone.utc).isoformat()


def _backoff_base_s() -> int:
    raw = os.environ.get(_BACKOFF_BASE_ENV)
    if raw:
        try:
            return max(1, int(raw))
        except ValueError:
            pass
    return _BACKOFF_BASE_S


def _backoff_max_s() -> int:
    raw = os.environ.get(_BACKOFF_MAX_ENV)
    if raw:
        try:
            return max(1, int(raw))
        except ValueError:
            pass
    return _BACKOFF_MAX_S


def _compute_next_attempt_after(
    failures: int, now_dt: _dt.datetime,
) -> str | None:
    """Compute the wall-clock ISO timestamp before which the next
    live-delta attempt MUST be skipped.

    Exponential backoff: ``BASE * 2**(failures-1)``, capped at
    ``BACKOFF_MAX_S``. ``failures == 0`` returns ``None`` (no
    backoff — always allowed to attempt). The cap means we keep
    re-attempting roughly every ``BACKOFF_MAX_S`` once a PR's
    endpoint is genuinely down, rather than escalating to
    intervals so long the recovery is missed.
    """
    if failures <= 0:
        return None
    base = _backoff_base_s()
    max_s = _backoff_max_s()
    # ``2**(failures-1)`` grows fast; cap so we don't compute
    # absurd numbers when failures is large.
    raw_delay_s = base * (2 ** min(failures - 1, 16))
    delay_s = min(raw_delay_s, max_s)
    return (now_dt + _dt.timedelta(seconds=delay_s)).isoformat()


def _backoff_active(cached: dict[str, Any] | None, now_dt: _dt.datetime) -> bool:
    """Return True iff the cache says the next live attempt is in the
    future. Missing / malformed timestamp → not active (attempt now).
    """
    if not cached:
        return False
    raw = cached.get("next_attempt_after")
    if not raw:
        return False
    try:
        deadline = _dt.datetime.fromisoformat(str(raw))
    except (ValueError, TypeError):
        return False
    return now_dt < deadline


def _read_cache(pr_number: int) -> dict[str, Any] | None:
    """Load the cache file for ``pr_number``. Returns ``None`` on any
    error (missing, malformed, schema-mismatch, stale)."""
    target = cache_path(pr_number)
    if not target.exists():
        return None
    try:
        text = target.read_text(encoding="utf-8")
        payload = json.loads(text)
    except (OSError, ValueError) as exc:
        _logger.warning(
            "comment cache read failed for PR #%s at %s: %s",
            pr_number, target, exc,
        )
        return None
    if not isinstance(payload, dict):
        return None
    if payload.get("schema_version") != SCHEMA_VERSION:
        return None
    # Staleness check
    try:
        fetched_at = _dt.datetime.fromisoformat(
            payload.get("fetched_at", "")
        )
        age_s = (
            _dt.datetime.now(_dt.timezone.utc) - fetched_at
        ).total_seconds()
        if age_s > _max_staleness_s():
            _logger.info(
                "comment cache for PR #%s is %.0fs stale (limit %ds); "
                "discarding for full refetch",
                pr_number, age_s, _max_staleness_s(),
            )
            return None
    except (ValueError, TypeError):
        # Bad timestamp → treat as no-cache
        return None
    if not isinstance(payload.get("comments"), list):
        return None
    return payload


def _write_cache(pr_number: int, payload: dict[str, Any]) -> None:
    """Atomic write of the cache file."""
    target = cache_path(pr_number)
    tmp = target.with_suffix(target.suffix + ".tmp")
    try:
        target.parent.mkdir(parents=True, exist_ok=True)
        tmp.write_text(
            json.dumps(payload, indent=2, default=str),
            encoding="utf-8",
        )
        tmp.replace(target)
    except (OSError, TypeError, ValueError) as exc:
        _logger.warning(
            "comment cache write failed for PR #%s at %s: %s",
            pr_number, target, exc,
        )
        try:
            tmp.unlink(missing_ok=True)
        except OSError:
            pass


def _merge_comments(
    cached: list[dict[str, Any]], delta: list[dict[str, Any]],
) -> list[dict[str, Any]]:
    """Append ``delta`` to ``cached``, deduplicated by comment ``id``.

    Forgejo's ``since=`` query is timestamp-based; if a comment was
    created at the boundary timestamp, the server MAY return it in
    both the cached snapshot and the delta. Dedup by id is the
    simplest defence.

    Preserves the order of ``cached`` (oldest first) and appends
    delta items (also oldest-first within delta) at the tail.
    """
    seen_ids: set[Any] = set()
    out: list[dict[str, Any]] = []
    for c in cached:
        cid = c.get("id")
        if cid is not None and cid in seen_ids:
            continue
        if cid is not None:
            seen_ids.add(cid)
        out.append(c)
    for c in delta:
        cid = c.get("id")
        if cid is not None and cid in seen_ids:
            continue
        if cid is not None:
            seen_ids.add(cid)
        out.append(c)
    return out


def _newest_cursor(comments: list[dict[str, Any]]) -> str:
    """Return the ``created_at`` of the newest (last) comment, for use
    as the ``?since=`` delta cursor on the next fetch.

    The comment list is oldest-first, so the last entry is the newest.
    Returns "" when the list is empty or the newest entry has no
    ``created_at`` — the caller falls back to ``fetched_at``.

    Why this and not ``fetched_at`` (wall-clock): a page-capped seed
    holds only the OLDEST N comments, so the newest *cached* comment
    is not the newest on the PR. ``?since=<newest cached created_at>``
    backfills forward from where the cache actually ends;
    ``?since=<wall-clock>`` would skip the un-fetched middle forever.
    """
    if not comments:
        return ""
    last = comments[-1]
    if isinstance(last, dict):
        return str(last.get("created_at") or "")
    return ""


def get_pr_comments(
    cfg: Any, pr_number: int,
    *,
    owner: str | None = None,
    repo: str | None = None,
) -> tuple[list[dict[str, Any]], bool]:
    """Get all comments for the PR, using the on-disk cache for the
    bulk and fetching only the delta.

    Returns ``(comments, completed)`` — same contract as
    :func:`_review_fetch.fetch_pr_comments`. ``completed`` is True
    when both the cache load AND the delta fetch succeeded
    cleanly; False if either was partial.

    ``owner`` / ``repo`` override the values normally read off ``cfg``.
    The shared comment fetch stack only uses ``cfg.owner`` / ``cfg.repo``
    to build the request path; the HTTP transport layer
    (``_claim_runtime.get``) needs just ``token`` / ``request_timeout_s``
    / ``api_retries``. Callers whose ``cfg`` is the narrower
    ``RuntimeContext`` (e.g. ``_claim_runtime``'s sweep, which uses
    module-level ``REPO_OWNER`` / ``REPO_NAME`` constants) pass owner /
    repo explicitly so they can share the cache without growing their
    cfg shape (spec #6).

    When the cache is disabled, missing, or stale, falls back to a
    full fetch (same behaviour as the legacy code path) and seeds
    the cache for next time.

    Seed-on-truncation (R8-5): a full fetch that hits the pagination
    ceiling (``truncated=True``) still seeds the cache — the clipped
    data is valid, just incomplete — marked ``any_partial_fetch=True``.
    The next run's delta fetch (``?since=<since_cursor>``) backfills
    forward from the newest cached comment. Before this fix, a heavy
    PR whose full fetch always truncated NEVER seeded the cache and
    re-paginated from page 1 every single cycle. A genuinely-failed
    fetch (network / 5xx / malformed, ``truncated=False``) still does
    NOT seed — its data is unreliable.
    """
    use_owner = owner or getattr(cfg, "owner", None)
    use_repo = repo or getattr(cfg, "repo", None)
    base_path = (
        f"/repos/{use_owner}/{use_repo}/issues/{int(pr_number)}/comments"
    )
    if is_disabled():
        # Path-based fetch (no cache layer). Uses the same paginator the
        # cache layer uses, so cfg with only RuntimeContext attrs works.
        items, ok = _review_fetch._api_get_paginated(cfg, base_path)
        return items, ok
    cached = _read_cache(pr_number)
    now_dt = _dt.datetime.now(_dt.timezone.utc)
    # Backoff short-circuit. If a previous cycle's live-delta failed
    # and we're inside the deferred window, skip the live attempt
    # entirely and serve whatever cached bulk we have. The caller
    # sees ``completed=False`` so the dispatcher's prompt-time
    # ``data_complete`` aggregate still flips, but we have NOT
    # burned another ~30s on a guaranteed-to-fail pagination walk.
    if cached is not None and _backoff_active(cached, now_dt):
        _logger.info(
            "comment cache for PR #%s in backoff (failures=%s, "
            "next_attempt_after=%s); serving stale cache, skipping live "
            "delta",
            pr_number,
            cached.get("consecutive_failures", 0),
            cached.get("next_attempt_after"),
        )
        return list(cached.get("comments") or []), False
    if cached is None:
        # Cache miss → full fetch. Seed on a clean walk OR a page-cap
        # truncation; skip seeding only on a genuine transient failure
        # (its data is unreliable). ``since_cursor`` is the newest
        # comment we actually have, so a truncated seed's next delta
        # backfills forward instead of from a wall-clock ``?since=``.
        items, completed, truncated = _review_fetch._api_get_paginated(
            cfg, base_path, return_truncation=True,
        )
        if completed or truncated:
            # ``truncated=True`` is also a fault we want backoff to
            # cover — every subsequent cycle that hits the cap is a
            # 30s tax on the prompt build for no new data. Treat it
            # like a delta-fail for failure-count purposes; reset on
            # a clean walk.
            failures = 0 if completed else 1
            _write_cache(pr_number, {
                "schema_version": SCHEMA_VERSION,
                "pr_number": int(pr_number),
                "fetched_at": _now(),
                "since_cursor": _newest_cursor(items),
                "comments": items,
                "any_partial_fetch": not completed,
                "consecutive_failures": failures,
                "next_attempt_after": _compute_next_attempt_after(
                    failures, now_dt,
                ),
            })
        return items, completed

    # Cache hit → delta fetch forward from the newest cached comment.
    # Fall back to ``fetched_at`` for caches written before the
    # ``since_cursor`` field existed.
    since = cached.get("since_cursor") or cached.get("fetched_at", "")
    path = f"{base_path}?since={since}" if since else base_path
    delta_items, delta_ok = _review_fetch._api_get_paginated(cfg, path)

    if not delta_ok:
        # Delta fetch was partial. Bump the failure counter and stamp
        # the deferred-retry deadline so the NEXT cycle short-circuits
        # to the stale-serve path above. We keep the cached bulk
        # intact (a partial-delta merge would risk dropping comments)
        # and let the caller mark ``data_complete=False``.
        prior_failures = int(cached.get("consecutive_failures") or 0)
        next_failures = prior_failures + 1
        merged = _merge_comments(cached["comments"], delta_items)
        _write_cache(pr_number, {
            **cached,
            # Preserve schema_version + comments + since_cursor +
            # any_partial_fetch from the prior payload. Only update
            # the failure-tracking fields.
            "consecutive_failures": next_failures,
            "next_attempt_after": _compute_next_attempt_after(
                next_failures, now_dt,
            ),
        })
        _logger.warning(
            "comment cache live-delta failed for PR #%s "
            "(consecutive_failures=%s); next attempt deferred until %s; "
            "serving %s cached comments to caller",
            pr_number, next_failures,
            _compute_next_attempt_after(next_failures, now_dt),
            len(merged),
        )
        return merged, False

    # Delta succeeded — merge, persist, return. A clean delta walks to
    # the PR's end, so even a previously-truncated seed is now whole:
    # ``any_partial_fetch`` clears AND failure-tracking resets.
    merged = _merge_comments(cached["comments"], delta_items)
    cached_payload = {
        "schema_version": SCHEMA_VERSION,
        "pr_number": int(pr_number),
        "fetched_at": _now(),
        "since_cursor": _newest_cursor(merged),
        "comments": merged,
        "any_partial_fetch": False,
        "consecutive_failures": 0,
        "next_attempt_after": None,
    }
    _write_cache(pr_number, cached_payload)
    return merged, True


def invalidate(pr_number: int) -> None:
    """Force-remove the cache entry for a PR.

    Used by:
    - The dispatcher's post-session cleanup, optionally on hook
      failures or other "this PR's state moved underneath us" cases.
    - Operator manual interventions via a future
      ``tools/clear_comment_cache.py`` script.

    Idempotent — missing entry is a no-op.
    """
    target = cache_path(pr_number)
    try:
        target.unlink(missing_ok=True)
    except OSError as exc:
        _logger.warning(
            "comment cache invalidate failed for PR #%s at %s: %s",
            pr_number, target, exc,
        )


__all__ = (
    "SCHEMA_VERSION",
    "cache_dir",
    "cache_path",
    "get_pr_comments",
    "invalidate",
    "is_disabled",
)