feat(ci): unified CI-log cache — get_ci_logs + implementer path (phase 1)

The implementer's ci_summary carried an empty raw_log_excerpt for every
failed gate (diagnosed via PR 39): prefetch._build_ci_summary passed a
no-op log fetcher, and ci_summarize._gate_to_nox_session never stripped
Forgejo's "(pull_request)" event suffix, so every PR gate fell through
to NoParserAvailable with no log fetched at all.

Phase 1 — the implementer-facing path:

- _ci_logs.get_ci_logs(): unified entry point — every job of a run,
  full untruncated logs, one cache. `partial` marks an in-flight run;
  a terminal + clean bundle is frozen forever. Reuses the existing
  session-cookie login + exponential backoff. Additive —
  fetch_pr_failure_logs and its 9 consumers are untouched.
- ci_summarize: strip the "(pull_request)" event suffix so gates
  resolve to their nox parser; _no_parser_failure now carries the raw
  log instead of hardcoding "".
- prefetch._build_ci_summary + forgejo_http + __main__: wire a real
  get_ci_logs-backed log_fetcher so each failed gate's raw_log_excerpt
  is filled from the cache.
- 9 new tests; full controller suite green (1169 passed).

Deferred to later phases: re-point the freshness gate / ci_status_poll
onto get_ci_logs, RUN_CI_LOCAL into the same cache, and
(head_sha, run_id, attempt) multi-attempt keying.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

tests/auto_agents/controller/test_entry_points.py

@@ -122,6 +122,9 @@ class TestMasterOptInLabelFlag:
             list_pr_comments: callable = lambda o, r, n: []
             get_ci_status: callable = lambda o, r, sha: None
             get_failure_logs: callable = lambda o, r, sha: ""
+            get_ci_logs: callable = lambda o, r, sha: {
+                "jobs": [], "partial": False, "completed": True,
+            }
             trigger_ci_rerun: callable = lambda o, r, branch: _stub_rerun_result()
 
         return _StubCB()

tests/auto_agents/controller/test_master_ci_summarize.py

@@ -174,17 +174,21 @@ class TestErrorHandling:
         assert failure["parser_used"] == "behave"
 
     def test_skip_coverage_session_no_parser(self):
-        # benchmark is in NOX_SESSIONS_SKIP_COVERAGE; should not parse.
+        # benchmark is in NOX_SESSIONS_SKIP_COVERAGE; no structured
+        # parser — but the raw log is still fetched + carried so a
+        # worker is not blind on an unmapped failed gate.
         statuses = [_gate(context="CI / benchmark", state="failure")]
         summary = summarize_ci_status(
             head_sha="abc",
             forgejo_status=_forgejo_status(statuses),
-            log_fetcher=lambda _: pytest.fail("should not fetch log"),
+            log_fetcher=lambda _: "benchmark regressed: 3 cases slower\n",
         )
         # benchmark isn't in NOX_SESSION_TO_PARSER, so gate→session
         # returns None → NoParserAvailable.
         failure = summary["gates"][0]["failure"]
         assert failure["error_class"] == "NoParserAvailable"
+        # ...and the raw log is surfaced rather than left empty.
+        assert "benchmark regressed" in failure["raw_log_excerpt"]
 
 
 # ─── parser_versions aggregate ─────────────────────────────────────
@@ -337,3 +341,18 @@ class TestDatetimeSerializationSafety:
         )
         with pytest.raises(TypeError, match="datetime"):
             json.dumps(summary)
+
+
+def test_pull_request_event_suffix_resolves_to_parser():
+    """Forgejo appends ' (pull_request)' to Actions gate contexts; the
+    parser resolver must strip it so a gate maps to its nox session
+    instead of falling through to NoParserAvailable (the PR 39 bug)."""
+    statuses = [_gate(context="CI / unit_tests (pull_request)", state="failure")]
+    summary = summarize_ci_status(
+        head_sha="abc",
+        forgejo_status=_forgejo_status(statuses),
+        log_fetcher=lambda _: "1 feature failed\n",
+    )
+    failure = summary["gates"][0]["failure"]
+    assert failure["error_class"] != "NoParserAvailable"
+    assert failure["parser_used"] == "behave"

tests/auto_agents/controller/test_master_prefetch.py

@@ -1331,3 +1331,49 @@ class TestToAwareDatetime:
         assert z == offset
         assert not (z > offset)
         assert not (offset > z)
+
+
+def test_build_ci_summary_fills_raw_log_from_get_ci_logs():
+    """_build_ci_summary pulls each failed gate's log from the
+    get_ci_logs callback into raw_log_excerpt."""
+    from tools.controller.master.prefetch import _build_ci_summary
+
+    ctx = "CI / benchmark (pull_request)"  # benchmark: no structured parser
+    forgejo_status = {
+        "state": "failure",
+        "statuses": [{"context": ctx, "state": "failure", "target_url": "u"}],
+    }
+    bundle = {"jobs": [{"context": ctx, "log": "OOM-killed: exit 137\n"}]}
+    cb = PrefetchDataCallbacks(
+        get_pr_details=lambda *a: None,
+        get_pr_diff=lambda *a: None,
+        list_pr_reviews=lambda *a: [],
+        list_pr_comments=lambda *a: [],
+        get_ci_status=lambda o, r, sha: forgejo_status,
+        get_ci_logs=lambda o, r, sha: bundle,
+    )
+    summary = _build_ci_summary(cb, owner="o", repo="r", head_sha="abc")
+    failure = summary["gates"][0]["failure"]
+    assert "OOM-killed: exit 137" in failure["raw_log_excerpt"]
+
+
+def test_build_ci_summary_without_get_ci_logs_leaves_excerpt_empty():
+    """Regression pin for the PR 39 bug: with no get_ci_logs callback
+    the failed gate's raw_log_excerpt stays empty."""
+    from tools.controller.master.prefetch import _build_ci_summary
+
+    ctx = "CI / benchmark (pull_request)"
+    forgejo_status = {
+        "state": "failure",
+        "statuses": [{"context": ctx, "state": "failure", "target_url": "u"}],
+    }
+    cb = PrefetchDataCallbacks(
+        get_pr_details=lambda *a: None,
+        get_pr_diff=lambda *a: None,
+        list_pr_reviews=lambda *a: [],
+        list_pr_comments=lambda *a: [],
+        get_ci_status=lambda o, r, sha: forgejo_status,
+    )
+    summary = _build_ci_summary(cb, owner="o", repo="r", head_sha="abc")
+    failure = summary["gates"][0]["failure"]
+    assert failure["raw_log_excerpt"] == ""

tests/auto_agents/test_ci_logs.py

@@ -672,3 +672,108 @@ def test_cache_path_invalid_sha_falls_back_to_sentinel(ci_mod, cache_dir):
     # NOT a write at the cache_dir root or anywhere outside it.
     p = ci_mod.cache_path("zzz!--__@@")
     assert p.name == "INVALID.json"
+
+
+# ─── get_ci_logs (unified full-log bundle) ──────────────────────────────
+
+
+class TestGetCILogs:
+    def _stub_session(self, monkeypatch, ci_mod, *, text="full\nlog\nbody"):
+        monkeypatch.setattr(
+            ci_mod,
+            "_ui_fetch_with_session",
+            lambda cfg, path: {"status": 200, "body": text.encode("utf-8")},
+        )
+
+    def test_all_jobs_full_untruncated(self, ci_mod, cfg, cache_dir, monkeypatch):
+        # Both passing AND failing jobs are fetched, full log, no tail cut.
+        self._stub_session(monkeypatch, ci_mod, text="X" * 9000)
+        detail = [
+            _failing_status(
+                "CI / unit_tests (pull_request)",
+                "https://h/o/r/actions/runs/5/jobs/1",
+            ),
+            {
+                "context": "CI / lint (pull_request)",
+                "state": "success",
+                "target_url": "https://h/o/r/actions/runs/5/jobs/2",
+            },
+        ]
+        bundle = ci_mod.get_ci_logs(cfg, "deadbeef", ci_detail=detail)
+        assert bundle["partial"] is False
+        assert bundle["completed"] is True
+        assert len(bundle["jobs"]) == 2
+        for job in bundle["jobs"]:
+            assert len(job["log"]) == 9000  # untruncated
+            assert job["fetch_error"] is None
+
+    def test_partial_when_a_check_is_pending(
+        self, ci_mod, cfg, cache_dir, monkeypatch
+    ):
+        self._stub_session(monkeypatch, ci_mod)
+        detail = [
+            _failing_status(
+                "CI / unit_tests", "https://h/o/r/actions/runs/5/jobs/1"
+            ),
+            {"context": "CI / e2e", "state": "pending", "target_url": ""},
+        ]
+        bundle = ci_mod.get_ci_logs(cfg, "deadbeef", ci_detail=detail)
+        assert bundle["partial"] is True
+
+    def test_terminal_clean_bundle_is_frozen(
+        self, ci_mod, cfg, cache_dir, monkeypatch
+    ):
+        self._stub_session(monkeypatch, ci_mod)
+        detail = [
+            _failing_status(
+                "CI / unit_tests", "https://h/o/r/actions/runs/5/jobs/1"
+            )
+        ]
+        first = ci_mod.get_ci_logs(cfg, "deadbeef", ci_detail=detail)
+        assert first["completed"] and not first["partial"]
+        # A frozen bundle must serve from cache — never re-fetch.
+        monkeypatch.setattr(
+            ci_mod,
+            "_ui_fetch_with_session",
+            lambda cfg, path: pytest.fail("frozen bundle must not re-fetch"),
+        )
+        second = ci_mod.get_ci_logs(cfg, "deadbeef", ci_detail=detail)
+        assert second["jobs"] == first["jobs"]
+
+    def test_partial_bundle_refetches(
+        self, ci_mod, cfg, cache_dir, monkeypatch
+    ):
+        self._stub_session(monkeypatch, ci_mod)
+        in_flight = [
+            _failing_status(
+                "CI / unit_tests", "https://h/o/r/actions/runs/5/jobs/1"
+            ),
+            {"context": "CI / e2e", "state": "running", "target_url": ""},
+        ]
+        b1 = ci_mod.get_ci_logs(cfg, "deadbeef", ci_detail=in_flight)
+        assert b1["partial"] is True
+        # CI finished — the next call must NOT serve the stale partial.
+        done = [
+            _failing_status(
+                "CI / unit_tests", "https://h/o/r/actions/runs/5/jobs/1"
+            )
+        ]
+        b2 = ci_mod.get_ci_logs(cfg, "deadbeef", ci_detail=done)
+        assert b2["partial"] is False
+
+    def test_logs_by_context(self, ci_mod, cfg, cache_dir, monkeypatch):
+        self._stub_session(monkeypatch, ci_mod, text="the failure text")
+        detail = [
+            _failing_status(
+                "CI / unit_tests (pull_request)",
+                "https://h/o/r/actions/runs/5/jobs/1",
+            )
+        ]
+        bundle = ci_mod.get_ci_logs(cfg, "deadbeef", ci_detail=detail)
+        mapping = ci_mod.logs_by_context(bundle)
+        assert mapping["CI / unit_tests (pull_request)"] == "the failure text"
+
+    def test_empty_head_sha(self, ci_mod, cfg, cache_dir):
+        bundle = ci_mod.get_ci_logs(cfg, "", ci_detail=[])
+        assert bundle["jobs"] == []
+        assert bundle["completed"] is True

tools/_ci_logs.py

@@ -517,10 +517,14 @@ def _fetch_job_log(
     run_id: int,
     job_id: int,
     *,
-    max_chars: int,
+    max_chars: int | None,
 ) -> tuple[str | None, int | None, bool, str | None]:
     """Fetch the raw log text for a single job. Returns
-    ``(log_tail, bytes_seen, truncated, error)``.
+    ``(log_text, bytes_seen, truncated, error)``.
+
+    ``max_chars=None`` returns the FULL untruncated log — the
+    :func:`get_ci_logs` bundle path. An integer keeps only the last
+    ``max_chars`` characters (the legacy failing-tail path).
 
     Forgejo's REST API does NOT expose action job logs (live-probed
     2026-05-17: every variant of ``/api/v1/.../actions/.../logs``
@@ -560,7 +564,7 @@ def _fetch_job_log(
     else:
         return None, None, False, "log:unexpected-body-shape"
     bytes_seen = len(text.encode("utf-8"))
-    if len(text) > max_chars:
+    if max_chars is not None and len(text) > max_chars:
         # Keep the tail — failing assertions and stack traces live
         # at the end of CI logs. Add a marker so the worker can tell
         # this isn't the full log.
@@ -921,13 +925,306 @@ def invalidate(head_sha: str) -> None:
         )
 
 
+# ─── Unified full-log bundle (get_ci_logs) ───────────────────────────
+#
+# ``get_ci_logs`` is the single entry point every CI-log consumer
+# should use. Unlike ``fetch_pr_failure_logs`` (failing jobs only,
+# 4000-char tails) it returns EVERY job of the run with FULL,
+# untruncated logs, in one cache. ``partial`` is True while the CI run
+# is still in-flight; a terminal run with a clean fetch is frozen
+# forever (per-SHA immutability). Truncation, when a consumer needs it
+# for a prompt budget, is the consumer's job at render time — never in
+# this cache.
+
+_BUNDLE_SCHEMA_VERSION = 1
+
+# Check states that mean the CI run has not reached a terminal verdict.
+_PENDING_STATES = frozenset(
+    {"pending", "running", "in_progress", "in-progress", "queued", "waiting"}
+)
+
+
+def bundle_cache_path(head_sha: str) -> Path:
+    """Per-SHA full-log bundle cache file. A distinct filename from the
+    legacy ``fetch_pr_failure_logs`` cache so the two never collide
+    while consumers are being migrated onto the bundle."""
+    safe = re.sub(r"[^a-fA-F0-9]", "", str(head_sha))[:64] or "INVALID"
+    return cache_dir() / f"{safe}.full.json"
+
+
+def _read_bundle(head_sha: str) -> dict[str, Any] | None:
+    target = bundle_cache_path(head_sha)
+    if not target.exists():
+        return None
+    try:
+        payload = json.loads(target.read_text(encoding="utf-8"))
+    except (OSError, ValueError) as exc:
+        _logger.warning("ci-logs bundle read failed for %s: %s", head_sha, exc)
+        return None
+    if not isinstance(payload, dict):
+        return None
+    if payload.get("schema_version") != _BUNDLE_SCHEMA_VERSION:
+        return None
+    if not isinstance(payload.get("jobs"), list):
+        return None
+    return payload
+
+
+def _write_bundle(head_sha: str, payload: dict[str, Any]) -> None:
+    target = bundle_cache_path(head_sha)
+    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("ci-logs bundle write failed for %s: %s", head_sha, exc)
+        try:
+            tmp.unlink(missing_ok=True)
+        except OSError:
+            pass
+
+
+def _ci_run_partial(ci_detail: list[dict[str, Any]] | None) -> bool:
+    """True iff any check is still pending/running — i.e. the CI run
+    has not reached a terminal verdict for this head_sha."""
+    for s in ci_detail or []:
+        if not isinstance(s, dict):
+            continue
+        state = str(s.get("state") or s.get("status") or "").lower()
+        if state in _PENDING_STATES:
+            return True
+    return False
+
+
+def collect_all_jobs(
+    cfg: Any,
+    ci_detail: list[dict[str, Any]],
+    *,
+    max_jobs: int,
+) -> tuple[list[dict[str, Any]], bool]:
+    """Fetch the FULL (untruncated) log for EVERY job in ``ci_detail`` —
+    passing and failing alike. Returns ``(jobs, all_clean)``;
+    ``all_clean`` is True iff every job's log was fetched cleanly."""
+    statuses = [s for s in ci_detail if isinstance(s, dict)][:max_jobs]
+    out: list[dict[str, Any]] = []
+    all_clean = True
+    for s in statuses:
+        context = s.get("context") or ""
+        target_url = s.get("target_url") or ""
+        run_id, job_id = parse_run_job_ids(target_url)
+        entry: dict[str, Any] = {
+            "context": context,
+            "state": s.get("state") or s.get("status"),
+            "description": s.get("description"),
+            "run_id": run_id,
+            "job_id": job_id,
+            "log_url": target_url,
+            "log": None,
+            "log_bytes": None,
+            "fetch_error": None,
+        }
+        if run_id is None:
+            entry["fetch_error"] = "unsupported-url-shape"
+            all_clean = False
+            out.append(entry)
+            continue
+        if job_id is None:
+            resolved, err = _resolve_job_id_for_context(cfg, run_id, context)
+            if err is not None or resolved is None:
+                entry["fetch_error"] = err or "run-jobs:no-job-id"
+                all_clean = False
+                out.append(entry)
+                continue
+            entry["job_id"] = job_id = resolved
+        log_text, bytes_seen, _trunc, log_err = _fetch_job_log(
+            cfg, int(run_id), int(job_id), max_chars=None,
+        )
+        entry["log"] = log_text
+        entry["log_bytes"] = bytes_seen
+        entry["fetch_error"] = log_err
+        if log_err is not None:
+            all_clean = False
+        out.append(entry)
+    return out, all_clean
+
+
+def _bundle(
+    head_sha: str,
+    *,
+    jobs: list[dict[str, Any]],
+    run_id: int | None,
+    partial: bool,
+    completed: bool,
+    consecutive_failures: int,
+    next_attempt_after: str | None,
+    source: str = "forgejo",
+) -> dict[str, Any]:
+    return {
+        "schema_version": _BUNDLE_SCHEMA_VERSION,
+        "head_sha": head_sha,
+        "fetched_at": _now(),
+        "source": source,
+        "run_id": run_id,
+        "jobs": jobs,
+        "partial": partial,
+        "completed": completed,
+        "consecutive_failures": consecutive_failures,
+        "next_attempt_after": next_attempt_after,
+    }
+
+
+def get_ci_logs(
+    cfg: Any,
+    head_sha: str,
+    *,
+    ci_detail: list[dict[str, Any]] | None = None,
+    max_jobs: int | None = None,
+) -> dict[str, Any]:
+    """Unified CI-log entry point — EVERY job of the run, FULL logs, one
+    cache. The single fetcher all CI-log consumers should call.
+
+    Returns a bundle dict::
+
+        {schema_version, head_sha, fetched_at, source, run_id,
+         partial, completed, consecutive_failures, next_attempt_after,
+         jobs: [{context, state, description, run_id, job_id, log_url,
+                 log, log_bytes, fetch_error}]}
+
+    - ``partial`` — True while the CI run is still in-flight (some check
+      pending/running). A partial bundle is always re-fetched on the
+      next call (CI is expected to change) — no backoff.
+    - ``completed`` — every reachable job log fetched cleanly.
+    - A terminal run (``not partial``) with a clean fetch
+      (``completed``) is frozen forever — served straight from cache.
+    - A terminal run with fetch errors is retried under the shared
+      exponential backoff.
+    """
+    effective_max_jobs = max_jobs if max_jobs is not None else _max_jobs()
+    if not head_sha:
+        return _bundle(
+            "", jobs=[], run_id=None, partial=False, completed=True,
+            consecutive_failures=0, next_attempt_after=None,
+        )
+
+    now_dt = _dt.datetime.now(_dt.timezone.utc)
+    disabled = is_disabled()
+    cached = None if disabled else _read_bundle(head_sha)
+    if cached is not None and not cached.get("partial"):
+        if cached.get("completed"):
+            return cached  # terminal + clean → frozen
+        if _backoff_active(cached, now_dt):
+            return cached  # terminal, fetch errors, inside backoff
+
+    detail = ci_detail
+    if detail is None:
+        try:
+            detail, _ok = _review_fetch.fetch_ci_check_detail(cfg, head_sha)
+        except Exception as exc:  # noqa: BLE001
+            return _record_bundle_failure(
+                head_sha, cached, now_dt,
+                error=f"ci-detail:{type(exc).__name__}",
+                disabled=disabled,
+            )
+
+    partial = _ci_run_partial(detail or [])
+    jobs, ok = collect_all_jobs(cfg, detail or [], max_jobs=effective_max_jobs)
+    run_id = next(
+        (j["run_id"] for j in jobs if j.get("run_id") is not None), None
+    )
+
+    if partial:
+        # In-flight: cache it (consumers still see what exists) but
+        # never freeze and never back off — the next call re-fetches.
+        payload = _bundle(
+            head_sha, jobs=jobs, run_id=run_id, partial=True, completed=ok,
+            consecutive_failures=0, next_attempt_after=None,
+        )
+    elif ok:
+        payload = _bundle(
+            head_sha, jobs=jobs, run_id=run_id, partial=False,
+            completed=True, consecutive_failures=0, next_attempt_after=None,
+        )
+    else:
+        # Terminal run, but >=1 job log unreachable -> backoff retry.
+        nxt = int((cached or {}).get("consecutive_failures") or 0) + 1
+        payload = _bundle(
+            head_sha, jobs=jobs, run_id=run_id, partial=False,
+            completed=False, consecutive_failures=nxt,
+            next_attempt_after=_compute_next_attempt_after(nxt, now_dt),
+        )
+        _logger.warning(
+            "ci-logs bundle partial fetch for %s (%s/%s jobs missing "
+            "logs); consecutive_failures=%s",
+            head_sha,
+            sum(1 for j in jobs if j.get("fetch_error")),
+            len(jobs),
+            nxt,
+        )
+    if not disabled:
+        _write_bundle(head_sha, payload)
+    return payload
+
+
+def _record_bundle_failure(
+    head_sha: str,
+    prior_cache: dict[str, Any] | None,
+    now_dt: _dt.datetime,
+    *,
+    error: str,
+    disabled: bool,
+) -> dict[str, Any]:
+    """"The live attempt couldn't even start" path for the bundle —
+    e.g. the ci-detail fetch raised. Persists a backoff record + any
+    previously-cached jobs so the loop converges."""
+    nxt = int((prior_cache or {}).get("consecutive_failures") or 0) + 1
+    payload = _bundle(
+        head_sha,
+        jobs=list((prior_cache or {}).get("jobs") or []),
+        run_id=(prior_cache or {}).get("run_id"),
+        partial=False,
+        completed=False,
+        consecutive_failures=nxt,
+        next_attempt_after=_compute_next_attempt_after(nxt, now_dt),
+    )
+    payload["last_error"] = error
+    if not disabled:
+        _write_bundle(head_sha, payload)
+    _logger.warning(
+        "ci-logs bundle live fetch failed for %s (%s); "
+        "consecutive_failures=%s",
+        head_sha, error, nxt,
+    )
+    return payload
+
+
+def logs_by_context(bundle: dict[str, Any]) -> dict[str, str]:
+    """Flatten a :func:`get_ci_logs` bundle to ``{gate_context:
+    full_log}`` — the shape a CISummary ``log_fetcher`` consumes. Jobs
+    whose log is missing (fetch error) are omitted."""
+    out: dict[str, str] = {}
+    for j in bundle.get("jobs") or []:
+        if not isinstance(j, dict):
+            continue
+        ctx, log = j.get("context"), j.get("log")
+        if ctx and isinstance(log, str):
+            out[ctx] = log
+    return out
+
+
 __all__ = (
     "SCHEMA_VERSION",
+    "bundle_cache_path",
     "cache_dir",
     "cache_path",
+    "collect_all_jobs",
     "collect_failing_jobs",
     "fetch_pr_failure_logs",
+    "get_ci_logs",
     "invalidate",
     "is_disabled",
+    "logs_by_context",
     "parse_run_job_ids",
 )

tools/controller/master/__main__.py

@@ -327,6 +327,9 @@ def main(argv: list[str] | None = None) -> int:
         # P2: wire the CI-status fetcher so prefetch populates
         # ci_summary / failing_gates instead of leaving them null.
         get_ci_status=callbacks.get_ci_status,
+        # Unified CI-log fetcher so each failed gate's raw_log_excerpt
+        # is filled from the cache (not left empty).
+        get_ci_logs=callbacks.get_ci_logs,
     )
     prefetch_cb = make_prefetch_callback(engine, prefetch_data)
 

tools/controller/master/ci_summarize.py

@@ -28,6 +28,7 @@ implementer at least sees the failure existed.
 from __future__ import annotations
 
 import logging
+import re
 from collections.abc import Callable
 from datetime import datetime, timezone
 from typing import Any
@@ -45,6 +46,10 @@ logger = logging.getLogger(__name__)
 # Forgejo job-log fetcher. None means the log was unreachable.
 LogFetcher = Callable[[str], str | None]
 
+# Cap on the raw-log excerpt carried on a NoParserAvailable gate —
+# matches the V1 ``CIFailure.raw_log_excerpt`` max_length budget.
+_RAW_EXCERPT_MAX_CHARS = 16_384
+
 
 # Forgejo / Gitea / GH-mirror status states → GateResult.status.
 # Covers states observed in the wild across Forgejo, Gitea, and
@@ -182,7 +187,11 @@ def _build_gate(
 
     nox_session = _gate_to_nox_session(context)
     if nox_session is None or nox_session in NOX_SESSIONS_SKIP_COVERAGE:
-        gate["failure"] = _no_parser_failure(context)
+        # No parser for this gate — but still surface the raw log so a
+        # worker is not blind on an unmapped failed gate.
+        gate["failure"] = _no_parser_failure(
+            context, _safe_fetch(log_fetcher, context)
+        )
         return gate, parser_versions
 
     try:
@@ -193,7 +202,9 @@ def _build_gate(
             nox_session,
             context,
         )
-        gate["failure"] = _no_parser_failure(context)
+        gate["failure"] = _no_parser_failure(
+            context, _safe_fetch(log_fetcher, context)
+        )
         return gate, parser_versions
 
     log = _safe_fetch(log_fetcher, context)
@@ -217,6 +228,11 @@ def _gate_to_nox_session(context: str) -> str | None:
     if not context:
         return None
     tail = context.rsplit("/", 1)[-1].strip()
+    # Forgejo appends the triggering event to Actions commit-status
+    # contexts — "CI / unit_tests (pull_request)". Strip that trailing
+    # "(event)" so the job name matches a nox session; without this
+    # every PR gate falls through to NoParserAvailable.
+    tail = re.sub(r"\s*\([^)]*\)\s*$", "", tail).strip()
     if tail in NOX_SESSION_TO_PARSER:
         return tail
     # Also support the parameterized form e.g. "unit_tests-3.13".
@@ -234,7 +250,11 @@ def _safe_fetch(log_fetcher: LogFetcher, context: str) -> str | None:
         return None
 
 
-def _no_parser_failure(context: str) -> dict:
+def _no_parser_failure(context: str, raw_log: str | None = None) -> dict:
+    """CIFailure for a gate with no structured parser. ``raw_log`` —
+    when the caller could fetch it — is carried as the excerpt (tail-
+    capped) so a worker still sees the failure text."""
+    excerpt = (raw_log or "")[-_RAW_EXCERPT_MAX_CHARS:]
     return {
         "parser_used": "(none)",
         "parser_version": "n/a",
@@ -243,8 +263,8 @@ def _no_parser_failure(context: str) -> dict:
         "findings": [],
         "failing_locations": [],
         "failed_assertions": [],
-        "raw_log_excerpt": "",
-        "log_excerpt_lines": 0,
+        "raw_log_excerpt": excerpt,
+        "log_excerpt_lines": (excerpt.count("\n") + 1) if excerpt else 0,
         "composite_findings": [],
     }
 

tools/controller/master/forgejo_http.py

@@ -83,6 +83,10 @@ class ForgejoCallbacks:
     # the infra-vs-real classifier the real log text; Forgejo status
     # descriptions are generic and never carry the error):
     get_failure_logs: "GetFailureLogsCallback"
+    # Unified full-log fetcher (tools/_ci_logs.get_ci_logs) — every job
+    # of the run, full untruncated logs, one cache. Feeds the
+    # implementer/reviewer ci_summary so raw_log_excerpt is populated:
+    get_ci_logs: "pf.GetCILogsCallback"
     # CI-rerun callback (CI-freshness gate — empty-commit push to
     # re-trigger CI; Forgejo 15.0.2 has no Actions rerun API):
     trigger_ci_rerun: "CIRerunCallback"
@@ -131,6 +135,7 @@ def build_callbacks(
         list_pr_comments=_make_list_pr_comments(cfg, runtime),
         get_ci_status=_make_get_ci_status(cfg, runtime),
         get_failure_logs=_make_get_failure_logs(cfg),
+        get_ci_logs=_make_get_ci_logs(cfg),
         trigger_ci_rerun=_make_trigger_ci_rerun(cfg),
     )
 
@@ -641,6 +646,65 @@ def _make_get_failure_logs(cfg: Any) -> "GetFailureLogsCallback":
     return get_failure_logs
 
 
+def _make_get_ci_logs(cfg: Any) -> "pf.GetCILogsCallback":
+    """Build the unified full-log fetcher — every job of a run, full
+    untruncated logs, one cache (``tools/_ci_logs.get_ci_logs``).
+
+    Returns the bundle dict; an empty bundle on any failure so a caller
+    never has to defend. The session-login + on-disk cache live in
+    ``_ci_logs``; this closure only builds the per-call cfg shim."""
+    import sys
+    from pathlib import Path
+
+    tools_dir = Path(__file__).resolve().parents[2]
+    if str(tools_dir) not in sys.path:
+        sys.path.insert(0, str(tools_dir))
+
+    def get_ci_logs(owner: str, repo: str, head_sha: str) -> dict:
+        empty: dict[str, Any] = {
+            "schema_version": 1, "head_sha": head_sha or "",
+            "jobs": [], "partial": False, "completed": True,
+        }
+        if not head_sha:
+            return empty
+        try:
+            from tools import _ci_logs  # type: ignore[import-not-found]
+        except Exception as exc:  # noqa: BLE001
+            logger.warning("get_ci_logs: _ci_logs import failed: %s", exc)
+            return empty
+        import os
+
+        api_base = os.environ.get(
+            "FORGEJO_API_BASE", "https://git.cleverthis.com/api/v1",
+        ).rstrip("/")
+        forgejo_url = (
+            api_base.rsplit("/api/v1", 1)[0]
+            if api_base.endswith("/api/v1")
+            else api_base
+        )
+
+        class _LogCfg:
+            pass
+
+        shim = _LogCfg()
+        shim.owner = owner
+        shim.repo = repo
+        shim.token = getattr(cfg, "token", "") or ""
+        shim.forgejo_url = forgejo_url
+        shim.request_timeout_s = getattr(cfg, "request_timeout_s", 30)
+        shim.api_retries = getattr(cfg, "api_retries", 3)
+        try:
+            return _ci_logs.get_ci_logs(shim, head_sha)
+        except Exception as exc:  # noqa: BLE001 — never abort a tick
+            logger.warning(
+                "get_ci_logs: get_ci_logs raised for %s/%s @%s: %s",
+                owner, repo, head_sha[:12], exc,
+            )
+            return empty
+
+    return get_ci_logs
+
+
 __all__ = [
     "CIRerunCallback",
     "ForgejoCallbacks",

tools/controller/master/prefetch.py

@@ -108,6 +108,9 @@ GetPRDiffCallback = Callable[[str, str, int], str | None]
 ListPRReviewsCallback = Callable[[str, str, int], list[dict]]
 ListPRCommentsCallback = Callable[[str, str, int], list[dict]]
 GetCIStatusCallback = Callable[[str, str, str], dict | None]
+# (owner, repo, head_sha) -> get_ci_logs bundle dict (all jobs, full
+# logs). See tools/_ci_logs.get_ci_logs.
+GetCILogsCallback = Callable[[str, str, str], dict]
 
 
 @dataclass(frozen=True)
@@ -124,6 +127,11 @@ class PrefetchDataCallbacks:
     # care about CI can omit it (defaults to None → ci_summary stays
     # None, the pre-P2 behaviour).
     get_ci_status: GetCIStatusCallback | None = None
+    # Unified CI-log fetcher (tools/_ci_logs.get_ci_logs). When set,
+    # _build_ci_summary fills each failed gate's log text from the
+    # cache instead of leaving raw_log_excerpt empty. Optional — when
+    # omitted the summary still carries gate states, just no log text.
+    get_ci_logs: GetCILogsCallback | None = None
 
 
 # ─── DB-side helpers ─────────────────────────────────────────────────
@@ -539,11 +547,10 @@ def _build_ci_summary(
     - the CI-status fetch fails or raises.
 
     The summarizer never raises; per-gate log fetches degrade to a
-    ``log-fetch-failed`` CIFailure. Prefetch has no per-job log
-    fetcher, so we pass a no-op fetcher — the worker still sees which
-    gates failed + their overall state, which is the load-bearing
-    context (the previous behaviour was ``ci_summary=None``, i.e. the
-    worker saw nothing at all).
+    ``log-fetch-failed`` CIFailure. When ``callbacks.get_ci_logs`` is
+    wired, each failed gate's ``raw_log_excerpt`` is filled from the
+    unified CI-log cache; without it the summary still carries gate
+    states (the worker at least sees which gates failed).
     """
     if callbacks.get_ci_status is None or not head_sha:
         return None
@@ -560,11 +567,34 @@ def _build_ci_summary(
         return None
     if forgejo_status is None:
         return None
+    # Real log fetcher: pull every gate's full log from the unified
+    # CI-log cache so the worker's ci_summary carries actual failure
+    # text, not just gate states. ``dict.get`` IS the LogFetcher
+    # contract (gate_context -> str | None).
+    logs_by_gate: dict[str, str] = {}
+    if callbacks.get_ci_logs is not None:
+        try:
+            bundle = callbacks.get_ci_logs(owner, repo, head_sha)
+            for job in bundle.get("jobs") or []:
+                if (
+                    isinstance(job, dict)
+                    and job.get("context")
+                    and isinstance(job.get("log"), str)
+                ):
+                    logs_by_gate[job["context"]] = job["log"]
+        except Exception as exc:  # noqa: BLE001 — degrade to no logs
+            logger.warning(
+                "prefetch: get_ci_logs failed for %s/%s @%s: %s",
+                owner,
+                repo,
+                head_sha[:12],
+                exc,
+            )
     try:
         return summarize_ci_status(
             head_sha=head_sha,
             forgejo_status=forgejo_status,
-            log_fetcher=lambda _gate: None,
+            log_fetcher=logs_by_gate.get,
         )
     except Exception as exc:  # noqa: BLE001 — never block prefetch
         logger.warning(
@@ -973,6 +1003,7 @@ def make_prefetch_callback(
 
 
 __all__ = [
+    "GetCILogsCallback",
     "GetCIStatusCallback",
     "GetPRDetailsCallback",
     "GetPRDiffCallback",