Merge remote-tracking branch 'gitea/master' into failure-auto-filing

Findings 11/12: the operator startup reconcile thread (rune_operator.py:1986)
IS the only reconcile cadence — no recurring timer exists in the codebase;
recorded as deviation 8 per the deviations contract. The O(N^2)
mark_consumed note is resolved by the finding-2 sidecar design. Full
finding -> commit -> red-evidence table appended to the Completion Record.

docs/failure-semantics.md

@@ -74,6 +74,45 @@ degraded, it is in the ledger even when the turn looked normal.
 4. Check `/health` after forcing a failure; the subsystem must appear in the
    ledger and the event must land in the journal.
 
+## Failure auto-filing (#122)
+
+T0/T1 failures additionally become ONE forge issue per failure signature
+`(subsystem, error_class)` via `src/rune/failure_auto_filer.py` — the
+forge's own watch system pings you; repeats comment with a rising count
+instead of flooding the board. T2 failures (cron noise) NEVER file.
+
+How it works:
+
+- **Signature** — `(subsystem, error_class)` where `error_class` is the
+  exception class name (or first token of the error string). Never the
+  full message: variable payloads would flood.
+- **Dedup** — before filing, a repo-scoped marker search
+  (`ForgeClient.search_issues`, `GET /issues?state=open&type=issues&q=<marker>`
+  — Gitea/Forgejo has no repo-scoped `/issues/search` path) looks for
+  the marker `[auto-failure:<subsystem>:<error_class>]` in titles and
+  bodies. Found → `add_comment` with a rising count; not found →
+  `create_issue`. The list endpoint is paginated, so older failure
+  issues are still seen.
+- **Burst coalescing** — an in-memory per-signature counter batches
+  repeats into one comment per 10 occurrences or 10 minutes (whichever
+  first). A 3am crash-loop writes one comment, not 200.
+- **Spool + reconcile** — any client exception (forge unreachable/auth/
+  5xx) appends a `failure_auto_file_spool` JSON line to the existing
+  event journal. `reconcile()` (startup + recovery) replays them; dedup
+  re-applies, filed lines are marked consumed.
+- **Threading** — the failure path only enqueues; one daemon worker
+  drains the queue and does all network work. Failure-path latency is
+  untouched, and nothing ever raises into the caller's except block.
+- **Tier gate** — `tier in (0, 1)` file/comment; `tier >= 2` journal
+  only. Enforced in ONE place: the filer.
+
+Wiring: the composition root (`rune_operator.get_operator`) creates one
+`FailureAutoFiler` via `create_forge_client()` (unconfigured →
+`NullAdapter` → dormant by construction) and wires it through
+`subsystem_health.set_auto_filer`; `record_failure(..., filer=...)`
+accepts an explicit override. `RUNE_AUTOFILE_OFF=1` disables entirely
+(fail-safe off switch).
+
 ## Scope note
 
 P0-06 converted the ~30 critical handlers on the authoritative path (surveyed

docs/forge-client.md

@@ -10,13 +10,20 @@ prerequisite for the failure auto-filer (#122).
 
 - `ForgeClient` — `@runtime_checkable` Protocol: `create_issue(title, body)`,
   `add_comment(issue_number, body)`, `get_issue_state(issue_number)`,
-  `list_open_issues()`
+  `list_open_issues()`, `search_issues(query)` (#122)
 - `NullAdapter` — every method is a no-op dict; zero errors; the graceful
   dormant path for unconfigured/standalone use
 - `GiteaForgejoAdapter` — real HTTP adapter via Gitea/Forgejo `/api/v1`
   paths (identical on both platforms); `Authorization: token <token>`
 - `create_forge_client(config)` — factory; unconfigured → `NullAdapter`
 
+`search_issues(query)` (#122) is the repo-scoped marker search over the
+paginated LIST endpoint (`GET /issues?state=open&type=issues&q=<query>`
++ client-side marker contains-check on title/body). Gitea/Forgejo has
+NO repo-scoped `/issues/search` path — the failure auto-filer dedups
+through the list endpoint precisely because it is paginated and
+server-filterable.
+
 ## Configuration
 
 The factory checks a config dict first, then environment variables:
@@ -43,6 +50,7 @@ If URL **or** token is missing → `NullAdapter` (never raises).
     POST   {url}/api/v1/repos/{owner}/{repo}/issues/{n}/comments
     GET    {url}/api/v1/repos/{owner}/{repo}/issues/{n}
     GET    {url}/api/v1/repos/{owner}/{repo}/issues?state=open&limit=50
+    GET    {url}/api/v1/repos/{owner}/{repo}/issues?state=open&type=issues&q=<query>&limit=50&page=N
 
 ## Run
 

plans/2026-09-07-122-failure-auto-filing.md

@@ -0,0 +1,137 @@
+# #122 Failure auto-filing: subsystem failures become forge issues (dedup + spool) — Implementation Plan
+
+> **Execution contract:** TDD red→green; controller owns task state; conventional commits referencing (#122); PR + protected checks only; deviations recorded in the plan tail.
+
+## Goal
+
+When a subsystem fails, Rune files ONE forge issue per failure signature, comments on repeats (`+N occurrences`), spools locally when the forge is unreachable, and reconciles the spool on recovery. T0/T1 failures file via ForgeClient; T2 stays journal-only (cron noise must not become board noise). Silent failure was eliminated in #22; this closes the loop: detect → record → track → fix.
+
+## Architecture
+
+- **New module `src/rune/failure_auto_filer.py`:**
+  - `FailureAutoFiler(client: ForgeClient, journal, dedup_window_hours=24)` — hooks `subsystem_health.record_failure` (wraps `registry().record` call sites via a `record_and_maybe_file()` entry point; the registry itself stays untouched — composition, not modification).
+  - Signature = `(subsystem, error_class)` where `error_class` = exception class name or first token of the error string. NOT the full message (variable payloads would flood).
+  - Dedup: before filing, `client.search_issues("[auto-failure:<subsystem>:<error_class>]")` — **server-side title search, NOT client-side list+filter**: `list_open_issues()` is hard-capped at `limit=50` with no pagination (forge_client.py:86-93, verified 2026-09-07), so list+filter silently misses older failure issues and duplicate-files. `search_issues` is a NEW protocol method added to `ForgeClient` (NullAdapter → `[]`, GiteaForgejoAdapter → `/issues/search?type=issues&q=<marker>`). Found → `add_comment(issue_number, "+N occurrences (last: <error excerpt> @ <ts>)")`. Not found → `create_issue` with the marker title, body carrying tier, excerpt, count-1, occurrence history tail.
+  - Spool: on ANY client exception (forge unreachable/auth/5xx), append a JSON line to the existing event journal (`memory/journal/events.jsonl`, already durable) with `kind: "failure_auto_file_spool"`. A `reconcile()` pass (called at operator startup and from maintenance cadence) replays spooled lines: dedup logic re-applies; confirmed-filed lines are marked consumed.
+  - Tier gate: `tier in (0, 1)` → file/comment; `tier >= 2` → journal only. The `in-memory occurrence counter` (per signature, process-lifetime) coalesces bursts: first occurrence files, subsequent occurrences within the dedup window batch into one comment per 10 occurrences or 10 minutes (whichever first) — prevents a 3am crash-loop from writing 200 comments.
+- **Wiring:** one `_auto_filer` instance on the operator composition root; `subsystem_health.record_failure()` gains an optional `filer` callback param (default None — zero behavior change when unwired); the two existing call-path families (context_broker :313, capability_broker :120, and any others found by grep) pass it through via the `registry().record` path unchanged — the filer observes AFTER the journal record, fire-and-forget (never blocks, never raises into the failure path — degrade-not-crash: filer exceptions are swallowed + journaled).
+- **Config:** `RUNE_AUTOFILE_OFF=1` env disables entirely (fail-safe off switch); forge endpoint from existing ForgeClient config (`create_forge_client()` — returns NullAdapter when unconfigured, which makes the filer a no-op by construction).
+
+## Design Decisions (settled)
+
+1. Marker-in-title dedup via NEW `search_issues` protocol method (server-side search — pagination-safe; list_open_issues' 50-cap makes client-side filtering unsound) — survives issue renames, works on Gitea AND any ForgeClient impl.
+2. Spool rides the EXISTING event journal (P0-06 durable store) — no new storage subsystem (anti-complexity rule).
+3. The filer NEVER runs in the exception's hot path synchronously against the network: it enqueues, a single worker drains (thread + queue.Queue, daemon). Failure-path latency stays untouched.
+4. T2 journal-only is enforced in ONE place (the filer), not at call sites.
+
+## Tasks
+
+### Task 1 — AutoFiler core: signature, dedup, file/comment
+Failing tests first (fake ForgeClient + fake journal): T0/T1 file once per signature; repeat → comment with count; T2 → no client calls; client exception → spool line written, no raise. Implement → green → commit `feat(failures): auto-filer core with signature dedup (#122)`.
+
+### Task 2 — Burst coalescing + reconcile
+Failing tests: 50 rapid occurrences → ≤2 client writes; reconcile replays spool (files missing, marks consumed, dedups against live issues); reconcile with forge still down → spool preserved. Implement → green → commit `feat(failures): burst coalescing + spool reconciliation (#122)`.
+
+### Task 3 — Wiring + env off-switch + docs
+`record_failure(filer=...)` param; composition-root wiring; `RUNE_AUTOFILE_OFF`; docs section (failure-handling doc or README ops section). Failing test: unwired call sites behave identically to today; wired sites file. Green → commit `feat(failures): wire auto-filer through composition (#122)`.
+
+## Acceptance (from #122)
+
+- One issue per failure signature, repeat occurrences commented with rising count ✓ Tasks 1–2
+- Forge unreachable → spooled locally, lands on recovery ✓ Task 2
+- T2 journal-only ✓ Task 1
+- You get pinged through the forge's own watch system ✓ (consequence of real issue creation)
+
+## Pipeline requirements checklist (from #122)
+
+- [ ] TDD red→green per task
+- [ ] `ruff check`/`format --check` clean
+- [ ] Spec gate PASS; quality gate Approved
+- [ ] Full suite green (604+new)
+- [ ] Existing canaries green
+- [ ] Closed with formatted comment + PR ref
+
+## Completion Record
+
+(status: COMPLETE — 2026-09-07, branch failure-auto-filing, base 2840f71)
+
+- Commits:
+  - d0df453 feat(failures): auto-filer core with signature dedup (#122)
+  - 082e17d feat(failures): burst coalescing + spool reconciliation (#122)
+  - e5e8727 feat(failures): wire auto-filer through composition (#122)
+- Counts: 663 passed + 84 subtests (baseline 604 + 84 → +59 tests); red
+  evidence per task recorded in session trace (ModuleNotFoundError /
+  ImportError red runs before each implementation).
+- Verification: full pytest green; pyright ratchet PASS (118/120 errors,
+  414/465 warnings — below baseline); ruff format --check + E9,F63,F7,F82
+  clean on all changed files; henxels check --all exit 0 (21 pre-existing
+  warnings, no canary added, ci.yml untouched).
+- Frozen kill tests: T2 → zero client calls (sync, worker, real call site
+  via record_failure path); forge-down → spool + never raises into the
+  failure path (sync + worker). Both in tests/test_failure_auto_filer.py.
+- Deviations (recorded per contract):
+  1. ~~Spool consumption = in-place journal line rewrite (`consumed: true`)
+     with a sidecar `<journal>.consumed` fallback when the rewrite fails~~
+     SUPERSEDED by gate discharge finding 2: consumption is the append-only
+     sidecar ONLY — the canonical journal is never rewritten (racy +
+     crash-poisoning; see Gate Discharge).
+  2. Spool records are plain JSON lines co-located in events.jsonl (NOT
+     canonical envelopes): DurableEventStore.append's dialect guard is
+     enforced only on its own append path, so co-located foreign lines
+     never violate the canonical store contract; verified no production
+     consumer reads offsets.
+  3. Batch comment cadence flushes when pending EXCEEDS batch_size
+     (comment at occurrences 11/21...), not exactly at 10/20 — "one
+     comment per 10 occurrences" semantics; first occurrence always files
+     immediately (plan: first occurrence files).
+  4. Worker additionally coalesces the already-queued backlog into one
+     write per signature (50-burst → ≤2 writes acceptance requires it).
+  5. record_failure forwards the ORIGINAL exception to the filer (plan
+     said "hooks record_failure" without specifying the payload); the
+     filer extracts the class name itself — signature stays authoritative.
+  6. rune_operator._wire_auto_filer uses the repo's established dual-mode
+     import pattern (package + script top-level deploy), matching the
+     existing record_failure import sites.
+  7. NullAdapter search/create never raises, so a spool line replayed
+     against NullAdapter is treated as NOT filed and stays spooled (not
+     falsely consumed) — safer under the "never lose a failure" rule.
+
+## Gate Discharge (2026-09-07 — two gate reviews, branch failure-auto-filing)
+
+All 12 findings discharged red-first (failing test committed/verified before each
+fix); O(N²) mark_consumed (finding 12) is resolved by the finding-2 sidecar
+design (append-only, no rewrite) — recorded here rather than in a separate commit.
+
+| # | Finding | Commit | Red evidence |
+|---|---------|--------|--------------|
+| 1 | search_issues hit nonexistent `/issues/search` (404 → spool forever) | fe9709c | 3 URL tests failed against old client (`assert captured["url"].startswith(..."/issues?")`) |
+| 2 | mark_consumed read-modify-write on shared canonical journal (racy, crash-poisoning) | 38b942c | byte-identity + concurrent-append tests failed (rewrite happened / append destroyed) |
+| 3 | reconcile per-line replay → comment flood after outage | 6765a78 | 10 spooled lines → 10 writes (`assert len(created)+len(commented) == 1` failed with 10) |
+| 4 | no secret redaction into persistent issues | 9105ca8 (+fc103fb test fix) | ImportError: `redact` missing |
+| 5 | reconcile bypassed the tier gate on replay | 5384cfc | crafted tier=2 spool line searched the forge (`searched == []` failed with `['[auto-failure:s:E]']`) |
+| 6 | reconcile thread + worker race → duplicate issues | ba99269 (+10e9788 test correction) | race test produced 2 created issues (`assert len(created) == 1` failed with 2) |
+| 7 | unbounded queue.Queue | 03f606e | `maxsize == 0` assertion failed |
+| 8 | vacuous `forge_created_not_called()` (unconditional True) | 1c16d3d | test rewritten to real fake-state assertions |
+| 9 | unbounded error_class (10KB marker) | 1c16d3d | cap test failed (len 10000 ≠ 64) |
+| 10 | `filer: Any \| None` → protocol type | 8f5c161 (+01b6eb4 operator annotation) | AttributeError: `AutoFiler` missing (test asserts `get_type_hints` equality) |
+| 11 | plan's "maintenance cadence" reconcile call absent | 01b6eb4 (deviation recorded below) | n/a — verification + documentation |
+| 12 | O(N²) mark_consumed | resolved by finding-2 sidecar (no rewrite, O(1) append) | covered by finding-2 red tests |
+
+**Deviation 8 (cadence):** the plan's "maintenance cadence" reconcile pass has
+no dedicated timer. The operator STARTUP thread (rune_operator.py:1986,
+`rune-autofile-reconcile`, launched by `_wire_auto_filer` — called once per
+process from `get_operator`) IS the only cadence: reconcile runs once per
+process boot, so spooled failures replay on restart. Accepted as deliberate
+scope: the filer spools on every failed write, the queue coalesces in-process
+retries, and a per-boot replay bounds the worst staleness to one restart; a
+recurring timer would add a background scheduler to a codebase that has none.
+Recorded per the deviations contract; no code added (the thread-launch comment
+names it the maintenance cadence).
+
+## Research trace
+
+Research round: issue #122 comment thread (posted 2026-09-07, amended after controller challenge — the amendment itself found the pagination defect).
+- subsystem_health.py:36 + forge_client.py:18-96 (INWARD) → composition-over-modification design; filer wraps, registry untouched.
+- ForgeClient.list_open_issues limit=50 no-pagination defect (forge_client.py:86-93, verified) → dedup via NEW search_issues protocol method (server-side, pagination-safe).
+- Event journal exists (memory/journal/events.jsonl) → spool rides existing durable store, no new subsystem.
+Kill test: if a T2 failure ever produces a client call, or a forge-down failure raises into the failure path, the design is wrong — frozen tests assert both.

src/rune/failure_auto_filer.py

@@ -0,0 +1,527 @@
+"""Failure auto-filer (#122): subsystem failures become forge issues.
+
+Detect → record (subsystem_health, untouched) → track (the forge issue) →
+fix. One forge issue per failure signature ``(subsystem, error_class)``;
+repeats comment with a rising count; T2 (ambient/cron) failures stay
+journal-only — cron noise must never become board noise; any forge
+exception spools to the durable event journal and is reconciled on
+recovery. The filer NEVER raises into the failure path.
+
+Threading: the failure path only ``submit()``s (enqueue, no network) and
+a single daemon worker drains the queue, so failure-path latency is
+untouched. The worker coalesces bursts: everything already queued for a
+signature is batched into ONE client write (first occurrence files,
+repeats comment once per ``batch_size`` occurrences or 10 minutes,
+whichever first). ``file()`` is the synchronous single-occurrence
+pipeline step (used by tests and reconcile); the tier gate lives here
+and nowhere else.
+"""
+
+from __future__ import annotations
+
+import json
+import logging
+import queue
+import re
+import threading
+import time
+from pathlib import Path
+from typing import Any, Protocol
+
+from rune.forge_client import ForgeClient
+
+AUTOFILE_OFF_ENV = "RUNE_AUTOFILE_OFF"
+SPOOL_KIND = "failure_auto_file_spool"
+BATCH_INTERVAL_SECONDS = 600  # one comment per 10 occurrences OR 10 minutes
+_EXCERPT_LEN = 200
+_ERROR_CLASS_MAX = 64  # a 10KB single-token error must not make a 10KB marker
+_QUEUE_MAXSIZE = 1000  # bounded: an outage must not balloon failure-path memory
+
+_LOG = logging.getLogger("rune.failure_auto_filer")
+
+
+def error_class_of(error: BaseException | str) -> str:
+    """Exception class name, or the first whitespace token of a string
+    (trailing ``:`` stripped, so ``"RuntimeError: boom"`` → ``RuntimeError``).
+    Capped at 64 chars (#122 gate): an unbounded single-token error would
+    make an unbounded marker."""
+    if isinstance(error, BaseException):
+        return type(error).__name__[:_ERROR_CLASS_MAX]
+    tokens = str(error).split()
+    return (tokens[0].rstrip(":") if tokens else "Unknown")[:_ERROR_CLASS_MAX]
+
+
+def signature_of(subsystem: str, error: BaseException | str) -> tuple[str, str]:
+    """The dedup signature: (subsystem, error_class). Never the full
+    message — variable payloads would flood the board."""
+    return (subsystem, error_class_of(error))
+
+
+def tier_gated(tier: int) -> bool:
+    """THE tier gate — the ONE place (module-level so the live path AND
+    the spool replay share it): T0/T1 file/comment, T2 journal-only
+    (cron noise must never become board noise)."""
+    return int(tier) >= 2
+
+
+def marker_for(subsystem: str, error_class: str) -> str:
+    return f"[auto-failure:{subsystem}:{error_class}]"
+
+
+def excerpt(error: BaseException | str, limit: int = _EXCERPT_LEN) -> str:
+    """The excerpt that lands in PERSISTENT forge issues and spool lines:
+    redacted first (broker exceptions can embed credentialed URLs/DSNs),
+    then truncated. Error class and shape stay visible."""
+    return redact(str(error))[:limit]
+
+
+# Credentials must never persist in forge issues. Patterns (order matters
+# little; all are applied): query-string credential params, URL userinfo
+# (scheme://user:pass@host — covers postgres/redis/mysql DSNs and basic
+# auth), Authorization headers / bearer tokens, and assignment forms
+# (password=..., secret=...). Only the credential SPAN is replaced — the
+# surrounding error text (class, host, port, message shape) survives.
+_QUERY_CREDS = re.compile(
+    r"([?&](?:token|password|passwd|secret|api_key|apikey|access_token|key|sig|signature|credential)=)[^&\s'\"]+",
+    re.IGNORECASE,
+)
+_URL_USERINFO = re.compile(r"\b([a-zA-Z][a-zA-Z0-9+.-]*://)([^\s:/@'\"]+)?:([^\s/@'\"]+)@")
+_BEARER = re.compile(r"(?i)\b((?:authorization\s*:?\s*(?:bearer\s+)?|bearer\s+))[^\s,'\";]+")
+_ASSIGNMENT = re.compile(
+    r"(?i)\b((?:password|passwd|secret|api_key|apikey|access_token|auth)\s*[=:]\s*)['\"]?([^\s'\"]{4,})"
+)
+
+
+def redact(text: str) -> str:
+    """Strip credentials from an error string (no-op on clean text)."""
+    out = _QUERY_CREDS.sub(r"\1<redacted>", text)
+    out = _URL_USERINFO.sub(r"\1<redacted>:<redacted>@", out)
+    out = _BEARER.sub(r"\1<redacted>", out)
+    out = _ASSIGNMENT.sub(lambda m: f"{m.group(1)}<redacted>", out)
+    return out
+
+
+def _error_text(error: BaseException | str) -> str:
+    return str(error) if isinstance(error, str) else f"{type(error).__name__}: {error}"
+
+
+class SpoolJournal(Protocol):
+    """What the filer needs from a durable journal (satisfied by
+    JsonlSpool over the existing event journal file)."""
+
+    def append(self, record: dict) -> None: ...
+    def unconsumed(self) -> list[dict]: ...
+    def mark_consumed(self, spool_id: str) -> None: ...
+
+
+class JsonlSpool:
+    """Spool riding the EXISTING durable event journal (no new storage).
+
+    Spooled records are appended as JSON lines with ``kind:
+    failure_auto_file_spool`` next to the canonical envelopes (which have
+    ``event_id`` keys). The canonical dialect is guarded on
+    DurableEventStore.append only, so co-located foreign lines are safe.
+
+    Consumption NEVER rewrites the canonical journal (#122 gate): the
+    old read-modify-write raced concurrent DurableEventStore.appends
+    (an O_APPEND landing between the read and the write was silently
+    destroyed) and a crash mid-rewrite poisoned the canonical store.
+    Instead the marker is an append-only sidecar ``<journal>.consumed``
+    (one spool id per line, atomic appends only, O(1) per mark — no
+    O(N) rewrite). ``unconsumed()`` = spool lines minus sidecar ids.
+    """
+
+    def __init__(self, path: Path) -> None:
+        self.path = Path(path)
+
+    def append(self, record: dict) -> None:
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        with open(self.path, "a", encoding="utf-8") as fh:
+            fh.write(json.dumps(record, sort_keys=True) + "\n")
+
+    def unconsumed(self) -> list[dict]:
+        if not self.path.exists():
+            return []
+        sidecar: set[str] = set()
+        sidecar_path = Path(f"{self.path}.consumed")
+        if sidecar_path.exists():
+            sidecar = {line.strip() for line in sidecar_path.read_text(encoding="utf-8").splitlines() if line.strip()}
+        rows: list[dict] = []
+        for line in self.path.read_text(encoding="utf-8").splitlines():
+            try:
+                row = json.loads(line)
+            except json.JSONDecodeError:
+                continue
+            if (
+                isinstance(row, dict)
+                and row.get("kind") == SPOOL_KIND
+                and not row.get("consumed")
+                and row.get("spool_id") not in sidecar
+            ):
+                rows.append(row)
+        return rows
+
+    def mark_consumed(self, spool_id: str) -> None:
+        """Record consumption in the append-only sidecar — the canonical
+        journal is NEVER rewritten (append-only, crash-safe, race-free
+        with concurrent DurableEventStore.appends)."""
+        try:
+            self.path.parent.mkdir(parents=True, exist_ok=True)
+            with open(f"{self.path}.consumed", "a", encoding="utf-8") as fh:
+                fh.write(spool_id + "\n")
+        except OSError:
+            pass  # last resort: the line stays unconsumed; dedup via
+        # search_issues still prevents duplicate issues on replay.
+
+
+class _SigState:
+    """In-memory, process-lifetime coalescing state per signature."""
+
+    __slots__ = ("count", "pending", "window_start")
+
+    def __init__(self) -> None:
+        self.count = 0  # total occurrences seen this process
+        self.pending = 0  # occurrences since the last client write
+        self.window_start = 0.0  # monotonic time of the last client write
+
+
+class FailureAutoFiler:
+    """One issue per failure signature; tier gate + spool + worker drain.
+
+    The tier gate (tier in (0, 1) → file/comment, tier >= 2 → journal
+    only) is enforced HERE and nowhere else.
+    """
+
+    def __init__(
+        self,
+        client: ForgeClient,
+        journal: SpoolJournal,
+        batch_size: int = 10,
+        wait: bool = False,
+    ) -> None:
+        self._client = client
+        self._journal = journal
+        self._queue: queue.Queue[tuple[str, str, int, float]] = queue.Queue(maxsize=_QUEUE_MAXSIZE)
+        self._lock = threading.Lock()
+        self._state: dict[tuple[str, str], _SigState] = {}
+        self._batch_size = max(1, batch_size)
+        self._worker: threading.Thread | None = None
+        self._inflight: set[tuple[str, str]] = set()  # single-flight per signature
+        if not wait:
+            self._worker = threading.Thread(target=self._run_worker, name="rune-auto-filer", daemon=True)
+            self._worker.start()
+
+    # ---------------------------------------------------- failure-path API
+
+    def __call__(
+        self,
+        subsystem: str,
+        error: BaseException | str,
+        tier: int,
+        _now: float | None = None,
+    ) -> None:
+        """Callable-filer contract: record_failure invokes the wired
+        filer directly; this is exactly submit()."""
+        self.submit(subsystem, error, tier, _now)
+
+    def submit(
+        self,
+        subsystem: str,
+        error: BaseException | str,
+        tier: int,
+        _now: float | None = None,
+    ) -> None:
+        """The failure-path entry: enqueue and return immediately.
+
+        Never blocks on the network, never raises (swallowed + logged).
+        With ``wait=True`` (synchronous/test mode) nothing is written
+        until ``drain()`` — the whole backlog coalesces into one write
+        per signature.
+        """
+        try:
+            if tier_gated(tier):
+                return  # THE tier gate — T2 journal-only, zero queue traffic
+            now = time.monotonic() if _now is None else _now
+            try:
+                self._queue.put_nowait((subsystem, _error_text(error), int(tier), now))
+            except queue.Full:
+                # Bounded queue (#122 gate): an overflow occurrence is
+                # dropped from the queue but SPOOLED — nothing is lost —
+                # and the drop is logged. Never blocks the failure path.
+                _LOG.warning("auto-filer queue full; spooling dropped occurrence for subsystem=%s", subsystem)
+                self._spool(subsystem, error, tier, Exception("auto-filer queue full"), count=1)
+        except Exception:  # noqa: BLE001 - never raise into the failure path
+            _LOG.exception("auto-filer submit failed for subsystem=%s", subsystem)
+
+    def drain(self, timeout: float | None = None) -> None:
+        """Block until the queue is empty (tests + operator shutdown)."""
+        if self._worker is None:
+            self._drain_once()
+            return
+        deadline = None if timeout is None else time.monotonic() + timeout
+        while self._queue.unfinished_tasks > 0:
+            if deadline is not None and time.monotonic() > deadline:
+                return
+            time.sleep(0.005)
+
+    # -------------------------------------------------- synchronous pipeline
+
+    def file(
+        self,
+        subsystem: str,
+        error: BaseException | str,
+        tier: int,
+        _now: float | None = None,
+    ) -> bool:
+        """Synchronous single-occurrence pipeline step.
+
+        Tier gate → coalescing state → (first occurrence, batch full, or
+        window expired) → search → create-or-comment. ANY client
+        exception is spooled and swallowed; this method never raises.
+        Returns True when a forge write was issued.
+        """
+        try:
+            if tier_gated(tier):
+                return False  # THE tier gate — one place, T2 journal-only
+            now = time.monotonic() if _now is None else _now
+            sig = signature_of(subsystem, error)
+            text = _error_text(error)
+            with self._lock:
+                state = self._state.setdefault(sig, _SigState())
+                state.count += 1
+                state.pending += 1
+                due = (
+                    state.count == 1
+                    or state.pending >= self._batch_size
+                    or (state.pending > 0 and now - state.window_start >= BATCH_INTERVAL_SECONDS)
+                )
+                count_at_flush = state.count
+                if due:
+                    state.pending = 0
+                    state.window_start = now
+            if not due:
+                return False  # coalesced: no client call for this occurrence
+            return self._write(sig, text, int(tier), count_at_flush)
+        except Exception as exc:  # noqa: BLE001 - degrade, never propagate
+            self._spool(subsystem, error, tier, exc)
+            return False
+
+    # ------------------------------------------------------------- reconcile
+
+    def reconcile(self) -> dict[str, int]:
+        """Replay unconsumed spool lines through the real pipeline.
+
+        Lines are GROUPED by signature first (#122 gate): one client
+        write per signature per reconcile pass, with the aggregated
+        offline count (a long outage must not comment-flood the issue).
+        Dedup re-applies via search_issues (a line whose issue already
+        exists becomes one comment, never a duplicate issue). Confirmed-
+        filed lines are marked consumed; failures leave the line spooled.
+        Never raises. Called at operator startup (the maintenance
+        cadence — see the plan-tail deviation, #122).
+        """
+        report = {"replayed": 0, "filed": 0, "failed": 0, "skipped": 0}
+        try:
+            pending = self._journal.unconsumed()
+        except Exception:  # noqa: BLE001
+            _LOG.exception("reconcile could not read the spool journal")
+            report["failed"] += 1
+            return report
+        groups: dict[tuple[str, str], list[dict]] = {}
+        order: list[tuple[str, str]] = []
+        for record in pending:
+            report["replayed"] += 1
+            signature = record.get("signature") or []
+            if len(signature) != 2:
+                report["failed"] += 1  # malformed line: leave it, never crash
+                continue
+            if tier_gated(int(record.get("tier", 1))):
+                # THE tier gate on replay — the same predicate as the live
+                # path (one place). A T2 line resolves journal-only:
+                # consumed, never written.
+                self._journal.mark_consumed(str(record.get("spool_id", "")))
+                report["skipped"] += 1
+                continue
+            sig = (str(signature[0]), str(signature[1]))
+            if sig not in groups:
+                groups[sig] = []
+                order.append(sig)
+            groups[sig].append(record)
+        for sig in order:
+            group = groups[sig]
+            last = group[-1]
+            try:
+                filed = self._write(
+                    sig,
+                    str(last.get("error", "")),
+                    int(last.get("tier", 1)),
+                    sum(max(1, int(r.get("count", 1))) for r in group),
+                    offline=len(group),
+                )
+                if filed:
+                    for record in group:
+                        self._journal.mark_consumed(str(record.get("spool_id", "")))
+                    report["filed"] += len(group)
+                else:
+                    report["failed"] += len(group)  # e.g. NullAdapter: nothing was filed
+            except Exception:  # noqa: BLE001 - forge still down: keep spooled
+                report["failed"] += len(group)
+        return report
+
+    # ------------------------------------------------------------ internals
+
+    def _write(self, sig: tuple[str, str], error_text: str, tier: int, count: int, offline: int = 0) -> bool:
+        """One client write for a signature: comment on the existing issue
+        (dedup via server-side search) or create it. ``offline`` > 0 marks
+        a reconcile replay of ``offline`` spooled lines (aggregated "while
+        offline" count). Raises on client failure — callers spool.
+
+        Single-flight per signature (#122 gate): the reconcile thread and
+        the worker share an in-flight set guarded by the class lock; a
+        racing caller for a signature already being written does NOT
+        write (returns False) — its spool line stays spooled and dedup
+        re-applies on the next replay. Cross-PROCESS dedup remains
+        best-effort TOCTOU between search and create (documented
+        limitation: two rune PROCESSES can still double-file).
+        """
+        with self._lock:
+            if sig in self._inflight:
+                return False  # single-flight: another thread owns this write
+            self._inflight.add(sig)
+        try:
+            return self._write_once(sig, error_text, tier, count, offline)
+        finally:
+            with self._lock:
+                self._inflight.discard(sig)
+
+    def _write_once(self, sig: tuple[str, str], error_text: str, tier: int, count: int, offline: int = 0) -> bool:
+        subsystem, error_class = sig
+        marker = marker_for(subsystem, error_class)
+        existing = self._client.search_issues(marker)
+        match = next(
+            (i for i in existing if marker in str(i.get("title", ""))),
+            None,
+        )
+        ts = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
+        if match is not None:
+            if offline > 0:
+                body = f"+{count} occurrences while offline (last: {excerpt(error_text)} @ {ts})"
+            else:
+                body = f"+{max(1, count - 1)} occurrences (last: {excerpt(error_text)} @ {ts})"
+            self._client.add_comment(int(match["number"]), body)
+            return True
+        title = f"{marker} {subsystem}"
+        body = (
+            f"Auto-filed failure signature `{subsystem}:{error_class}`\n\n"
+            f"- tier={tier}\n"
+            f"- occurrences={count}\n"
+            f"- last: {excerpt(error_text)}\n\n"
+            f"(detected by the failure auto-filer, #122)"
+        )
+        result = self._client.create_issue(title, body)
+        if isinstance(result, dict) and result.get("created") is False:
+            return False  # e.g. NullAdapter: nothing was filed
+        return True
+
+    def _spool(
+        self,
+        subsystem: str,
+        error: BaseException | str,
+        tier: int,
+        exc: Exception,
+        count: int = 1,
+    ) -> None:
+        """Persist the failure for reconcile(). Never raises."""
+        try:
+            subsystem_name, error_class = signature_of(subsystem, error)
+            record = {
+                "kind": SPOOL_KIND,
+                "spool_id": f"spool-{time.time_ns()}",
+                "timestamp": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+                "signature": [subsystem_name, error_class],
+                "tier": int(tier),
+                "count": int(count),
+                "error": excerpt(error),
+                "client_error": excerpt(exc),
+            }
+            self._journal.append(record)
+        except Exception:  # noqa: BLE001 - even the spool must not raise
+            _LOG.exception("failed to spool auto-file record for subsystem=%s", subsystem)
+
+    # ------------------------------------------------------------- worker
+
+    def _run_worker(self) -> None:
+        while True:
+            item = self._queue.get()
+            if item is None:  # shutdown sentinel
+                self._queue.task_done()
+                return
+            batch = [item]
+            while True:  # coalesce: everything already queued rides one write
+                try:
+                    batch.append(self._queue.get_nowait())
+                except queue.Empty:
+                    break
+            try:
+                self._process_batch(batch)
+            except Exception:  # noqa: BLE001 - worker must never die
+                _LOG.exception("auto-filer worker batch failed")
+            finally:
+                for _ in batch:
+                    self._queue.task_done()
+
+    def _process_batch(self, batch: list[tuple[str, str, int, float]]) -> None:
+        groups: dict[tuple[str, str], list[tuple[str, str, int, float]]] = {}
+        order: list[tuple[str, str]] = []
+        for subsystem, text, tier, now in batch:
+            if tier_gated(tier):
+                continue  # THE tier gate (defense in depth; submit gates too)
+            sig = signature_of(subsystem, text)
+            if sig not in groups:
+                groups[sig] = []
+                order.append(sig)
+            groups[sig].append((subsystem, text, tier, now))
+        for sig in order:
+            group = groups[sig]
+            try:
+                self._flush_group(sig, group)
+            except Exception as exc:  # noqa: BLE001 - spool, never propagate
+                for subsystem, text, tier, _now in group:
+                    self._spool(subsystem, text, tier, exc, count=len(group))
+
+    def _flush_group(self, sig: tuple[str, str], group: list[tuple[str, str, int, float]]) -> None:
+        now = group[-1][3]
+        text = group[-1][1]  # last error of the batch
+        tier = group[-1][2]
+        with self._lock:
+            state = self._state.setdefault(sig, _SigState())
+            state.count += len(group)
+            state.pending += len(group)
+            due = (
+                state.count == len(group)
+                or state.pending >= self._batch_size
+                or (state.pending > 0 and now - state.window_start >= BATCH_INTERVAL_SECONDS)
+            )
+            count_at_flush = state.count
+            if due:
+                state.pending = 0
+                state.window_start = now
+        if due:
+            self._write(sig, text, tier, count_at_flush)
+
+    def _drain_once(self) -> None:
+        """Inline drain for synchronous (wait=True) mode."""
+        batch: list[tuple[str, str, int, float]] = []
+        while True:
+            try:
+                batch.append(self._queue.get_nowait())
+            except queue.Empty:
+                break
+        if batch:
+            try:
+                self._process_batch(batch)
+            except Exception:  # noqa: BLE001 - must never escape
+                _LOG.exception("auto-filer inline drain failed")
+            finally:
+                for _ in batch:
+                    self._queue.task_done()

src/rune/forge_client.py

@@ -20,6 +20,7 @@ class ForgeClient(Protocol):
     def add_comment(self, issue_number: int, body: str) -> dict: ...
     def get_issue_state(self, issue_number: int) -> dict: ...
     def list_open_issues(self) -> list[dict]: ...
+    def search_issues(self, query: str) -> list[dict]: ...
 
 
 class NullAdapter:
@@ -37,6 +38,9 @@ class NullAdapter:
     def list_open_issues(self) -> list[dict]:
         return []
 
+    def search_issues(self, query: str) -> list[dict]:
+        return []
+
 
 class GiteaForgejoAdapter:
     """Gitea/Forgejo adapter — /api/v1 paths (identical on both platforms)."""
@@ -92,6 +96,37 @@ class GiteaForgejoAdapter:
         r.raise_for_status()
         return r.json()
 
+    def search_issues(self, query: str) -> list[dict]:
+        """Repo-scoped marker search over the paginated list endpoint.
+
+        #122 gate fix: Gitea/Forgejo has NO repo-scoped ``/issues/search``
+        path (verified against live swagger) — that URL 404s on a real
+        forge, so every dedup search would raise and everything would
+        spool forever. The repo-scoped LIST endpoint
+        (``GET /repos/{owner}/{repo}/issues``) supports ``state``, ``type``
+        and ``q`` server-side and is paginated: ``q`` matches titles (and
+        keywords), so the full marker is sent as ``q`` and a client-side
+        contains-check keeps exactly the issues whose title OR body
+        contains the marker (dedup must see body mentions too, but a
+        title hit is what the filer matches on).
+        """
+        results: list[dict] = []
+        page = 1
+        while True:
+            from urllib.parse import urlencode
+
+            r = httpx.get(
+                f"{self._base_url}/issues?{urlencode({'state': 'open', 'type': 'issues', 'q': query, 'limit': 50, 'page': page})}",
+                headers=self._headers(),
+                timeout=30,
+            )
+            r.raise_for_status()
+            batch = r.json()
+            results.extend(batch)
+            if len(batch) < 50:
+                return [i for i in results if query in str(i.get("title", "")) + str(i.get("body", ""))]
+            page += 1
+
 
 def create_forge_client(config: dict | None = None) -> ForgeClient:
     """Create a forge client from config or environment.

src/rune/rune_operator.py

@@ -24,7 +24,11 @@ import urllib.request
 from datetime import datetime, timezone
 from dataclasses import asdict, dataclass, field
 from pathlib import Path
-from typing import Dict, List, Optional, Any
+from typing import TYPE_CHECKING, Dict, List, Optional, Any
+
+if TYPE_CHECKING:
+    from rune.failure_auto_filer import FailureAutoFiler as _FailureAutoFiler  # noqa: F401
+
 
 # Safety override — adaptive escalation on refusal
 _SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
@@ -81,6 +85,7 @@ def _build_model_router():
         record_failure("model_routing", exc, tier=1, context="build_model_router")
         return None
 
+
 try:
     try:
         from .local_tool_runtime import LocalToolRuntime, OllamaChatTransport
@@ -131,6 +136,7 @@ try:
         TaskStateRepository,
         migrate_operator_tasks,
     )
+
     try:
         from .orchestration.event_store import DurableEventStore
     except ImportError:  # TUI top-level deploy
@@ -208,11 +214,7 @@ def _memory_query_from_trace(trace: Dict[str, Any], task_id: str) -> MemoryQuery
     Admitted blocks become the {block_name: tokens} map; dropped blocks ride
     through with their recorded reasons. Observable selection state only.
     """
-    admitted = {
-        block["name"]: block["tokens"]
-        for block in trace.get("blocks", [])
-        if block.get("admitted")
-    }
+    admitted = {block["name"]: block["tokens"] for block in trace.get("blocks", []) if block.get("admitted")}
     dropped = [
         {"name": block.get("name", ""), "tokens": block.get("tokens", 0)}
         for block in trace.get("blocks", [])
@@ -261,9 +263,7 @@ def _observations_from_degraded(degraded: list[str], task_id: str) -> list:
     ]
 
 
-def _verification_and_outcome(
-    response: OperatorResponse, task_id: str, degraded: list[str]
-) -> tuple:
+def _verification_and_outcome(response: OperatorResponse, task_id: str, degraded: list[str]) -> tuple:
     """VerificationResult + linked Outcome for the final response (#30).
 
     Method contract (frozen): 'exit_code' when a run_command tool event
@@ -290,9 +290,7 @@ def _verification_and_outcome(
             evidence = _bounded(result.get("result") or "run_tests passed")
             break
 
-    has_tool_failure = any(
-        event.kind in {"error", "failed"} for event in response.tool_events
-    )
+    has_tool_failure = any(event.kind in {"error", "failed"} for event in response.tool_events)
     passed = bool(response.success) and not has_tool_failure
     if method == "none":
         # No objective evidence ran — never self-report (plan: the
@@ -320,6 +318,7 @@ def _verification_and_outcome(
     )
     return verification, outcome
 
+
 #: A message that looks like a confirmation/follow-up of the active task
 #: gets amended onto it instead of minting a new task (P0-02 #18).
 _FOLLOW_UP_RE = re.compile(
@@ -439,9 +438,7 @@ class RuneOperator:
         self._events_store = DurableEventStore(root / "events.jsonl")
         self._hot_store = AtomicStore(root / "hot")
         self._task_state = TaskStateRepository(self._events_store, self._hot_store)
-        self._conversation = ConversationStateRepository(
-            self._events_store, self._hot_store, tasks=self._task_state
-        )
+        self._conversation = ConversationStateRepository(self._events_store, self._hot_store, tasks=self._task_state)
         try:
             self._migration_report = migrate_operator_tasks(
                 Path(self.storage_path), self._task_state, self._conversation
@@ -449,9 +446,7 @@ class RuneOperator:
         except Exception as exc:  # P0-06 contract: degrade loudly, never silently
             import logging
 
-            logging.getLogger(__name__).warning(
-                "Legacy operator.json migration failed: %s", exc
-            )
+            logging.getLogger(__name__).warning("Legacy operator.json migration failed: %s", exc)
             self._migration_report = None
 
     def _load(self):
@@ -466,13 +461,7 @@ class RuneOperator:
                     current_task=data.get("current_task"),
                     active_model=data.get("active_model"),
                     tasks=[
-                        RuneTask(
-                            **{
-                                k: v
-                                for k, v in t.items()
-                                if k in RuneTask.__dataclass_fields__
-                            }
-                        )
+                        RuneTask(**{k: v for k, v in t.items() if k in RuneTask.__dataclass_fields__})
                         for t in data.get("tasks", [])
                     ],
                     pending_task_id=data.get("pending_task_id"),
@@ -482,9 +471,7 @@ class RuneOperator:
             except Exception as e:
                 import logging
 
-                logging.getLogger(__name__).warning(
-                    "Failed to load operator state: %s", e
-                )
+                logging.getLogger(__name__).warning("Failed to load operator state: %s", e)
 
     def _save(self):
         """Save Rune's state to storage."""
@@ -499,15 +486,9 @@ class RuneOperator:
             "role": self.state.role,
             "current_task": self.state.current_task,
             "active_model": self.state.active_model,
-            "tasks_pending": len(
-                [t for t in self.state.tasks if t.status == "pending"]
-            ),
-            "tasks_running": len(
-                [t for t in self.state.tasks if t.status == "running"]
-            ),
-            "tasks_completed": len(
-                [t for t in self.state.tasks if t.status == "completed"]
-            ),
+            "tasks_pending": len([t for t in self.state.tasks if t.status == "pending"]),
+            "tasks_running": len([t for t in self.state.tasks if t.status == "running"]),
+            "tasks_completed": len([t for t in self.state.tasks if t.status == "completed"]),
         }
 
     def send_task(self, description: str, model: str | None = None, channel: str = OPERATOR_CHANNEL) -> Dict[str, Any]:
@@ -519,9 +500,7 @@ class RuneOperator:
         a fresh task records its parent for lineage.
         """
         active = self._task_state.get_active(channel)
-        if active is not None and (
-            active.status == "running" or _is_follow_up(description)
-        ):
+        if active is not None and (active.status == "running" or _is_follow_up(description)):
             state_task = self._task_state.amend(active.task_id, description)
         else:
             latest = self._task_state.latest(channel)
@@ -577,9 +556,7 @@ class RuneOperator:
             if decision is not None:
                 model = decision.model
                 routing_reason = (
-                    decision.reason
-                    if decision.reason.startswith("router:")
-                    else f"router: {decision.reason}"
+                    decision.reason if decision.reason.startswith("router:") else f"router: {decision.reason}"
                 )
                 routing_mode = "cloud" if decision.provider not in ("ollama", "local") else "local"
             else:
@@ -602,11 +579,7 @@ class RuneOperator:
         )
 
         failed_event = next(
-            (
-                event
-                for event in result.tool_events
-                if event.kind in {"error", "failed"}
-            ),
+            (event for event in result.tool_events if event.kind in {"error", "failed"}),
             None,
         )
         if failed_event is not None:
@@ -645,9 +618,7 @@ class RuneOperator:
             task.result = result.user_text
             task.completed_at = datetime.now(timezone.utc).isoformat()
             self.state.current_task = None
-            self._task_state.transition(
-                task.task_id, "completed", outcome_ref="response:" + task.task_id
-            )
+            self._task_state.transition(task.task_id, "completed", outcome_ref="response:" + task.task_id)
             self._conversation.append(
                 ConversationMessageRecord(
                     role="assistant",
@@ -680,9 +651,7 @@ class RuneOperator:
             if "status" not in obj or obj.get("task_id") != task.task_id:
                 continue
             obj["status"] = task.status
-            verification = next(
-                (item for item in result.runtime_objects if "method" in item), None
-            )
+            verification = next((item for item in result.runtime_objects if "method" in item), None)
             obj["verification_id"] = (
                 verification["verification_id"]
                 if verification is not None and verification.get("passed") and task.status == "completed"
@@ -786,9 +755,7 @@ class RuneOperator:
             self._last_capability_selection = selection
             return selection
 
-    def _conversation_messages(
-        self, current_task_id: str, limit: int = 8
-    ) -> list[dict]:
+    def _conversation_messages(self, current_task_id: str, limit: int = 8) -> list[dict]:
         """Return recent completed user/assistant pairs for the next turn."""
         completed = [
             item
@@ -807,36 +774,22 @@ class RuneOperator:
     def resolve_pending_approval(self, approved: bool) -> Dict[str, Any]:
         """Resolve and resume the single persisted operator approval."""
         task = next(
-            (
-                item
-                for item in self.state.tasks
-                if item.task_id == self.state.pending_task_id
-            ),
+            (item for item in self.state.tasks if item.task_id == self.state.pending_task_id),
             None,
         )
         request = self.state.pending_request
         pending_state = self.state.pending_runtime_state
         if task is None or request is None or pending_state is None:
-            return self._response_envelope(
-                OperatorResponse(success=False, error="no pending approval"), task
-            )
+            return self._response_envelope(OperatorResponse(success=False, error="no pending approval"), task)
 
         model = task.assigned_model or self.state.active_model or ""
         runtime = self._runtime_for_model(model, task.description)
-        result = (
-            runtime.approve(pending_state)
-            if approved
-            else runtime.reject(pending_state)
-        )
+        result = runtime.approve(pending_state) if approved else runtime.reject(pending_state)
         result = self._finalize_runtime_response(
             result, model, task.description, result.override_tier, result.recall_count
         )
         failed_event = next(
-            (
-                event
-                for event in result.tool_events
-                if event.kind in {"error", "failed"}
-            ),
+            (event for event in result.tool_events if event.kind in {"error", "failed"}),
             None,
         )
         if failed_event is not None:
@@ -872,9 +825,7 @@ class RuneOperator:
         """
         try:
             runtime_objects: list = []
-            for observation in _observations_from_degraded(
-                getattr(response, "degraded", None) or [], task_id
-            ):
+            for observation in _observations_from_degraded(getattr(response, "degraded", None) or [], task_id):
                 runtime_objects.append(observation.to_dict())
             if response.pending_approval is None:
                 verification, outcome = _verification_and_outcome(
@@ -962,26 +913,18 @@ class RuneOperator:
         return [m.to_dict() for m in self._conversation.window(channel, n=n)]
 
     @staticmethod
-    def _response_envelope(
-        result: OperatorResponse, task: Optional[RuneTask]
-    ) -> Dict[str, Any]:
+    def _response_envelope(result: OperatorResponse, task: Optional[RuneTask]) -> Dict[str, Any]:
         envelope = asdict(result)
         envelope["task"] = task.to_dict() if task is not None else None
         envelope["response"] = result.user_text
         if result.pending_approval is not None:
             pending_event = next(
-                (
-                    event
-                    for event in reversed(result.tool_events)
-                    if event.kind == "pending_approval"
-                ),
+                (event for event in reversed(result.tool_events) if event.kind == "pending_approval"),
                 None,
             )
             envelope["needs_confirm"] = True
             envelope["tool_call"] = asdict(result.pending_approval)
-            envelope["tool_result"] = (
-                dict(pending_event.result or {}) if pending_event else {}
-            )
+            envelope["tool_result"] = dict(pending_event.result or {}) if pending_event else {}
         return envelope
 
     def _fallback_model(self, description: str) -> str:
@@ -994,22 +937,15 @@ class RuneOperator:
         desc_lower = description.lower()
 
         # Quick tasks — laptop model
-        if any(
-            word in desc_lower for word in ["hello", "hi", "thanks", "ok", "status"]
-        ):
+        if any(word in desc_lower for word in ["hello", "hi", "thanks", "ok", "status"]):
             return "qwen2.5:7b"
 
         # Coding tasks — local coding model
-        if any(
-            word in desc_lower for word in ["fix", "code", "debug", "test", "refactor"]
-        ):
+        if any(word in desc_lower for word in ["fix", "code", "debug", "test", "refactor"]):
             return "qwen3:14b"
 
         # Complex tasks — strong local model
-        if any(
-            word in desc_lower
-            for word in ["design", "architecture", "research", "analyze"]
-        ):
+        if any(word in desc_lower for word in ["design", "architecture", "research", "analyze"]):
             return "hf.co/DavidAU/Qwen3.6-27B-Fable-Fusion-711-Uncensored-Heretic-NM-DAU-NEO-MAX-MTP-GGUF:IQ2_M"
 
         # Default to laptop model
@@ -1553,20 +1489,24 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
             if not hasattr(broker, "assemble"):
                 from types import SimpleNamespace as _NS
 
-                broker = _NS(assemble=lambda *a, **k: (
-                    _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
-                    {},
-                ))
+                broker = _NS(
+                    assemble=lambda *a, **k: (
+                        _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
+                        {},
+                    )
+                )
             assemble = getattr(broker, "assemble", None)
             if assemble is None:
                 # Degrade-not-crash: a stand-in broker without assemble gets an
                 # empty capsule + trace (scripted tests, Mock runtimes).
                 from types import SimpleNamespace as _NS
 
-                broker = _NS(assemble=lambda *a, **k: (
-                    _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
-                    {},
-                ))
+                broker = _NS(
+                    assemble=lambda *a, **k: (
+                        _NS(token_count=0, blocks=[], active_objective="", to_messages=lambda: []),
+                        {},
+                    )
+                )
             capsule, trace = broker.assemble(task, session, model_id=model)  # type: ignore[attr-defined]
             self._last_trace = trace
             for subsystem in trace.get("degraded", []):
@@ -1637,9 +1577,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                     "output": getattr(transport, "usage_output_tokens", 0),
                     "duration_ns": getattr(transport, "usage_total_duration_ns", 0),
                 }
-                prev = _prev_usage.setdefault(
-                    id(transport), {"calls": 0, "prompt": 0, "output": 0, "duration_ns": 0}
-                )
+                prev = _prev_usage.setdefault(id(transport), {"calls": 0, "prompt": 0, "output": 0, "duration_ns": 0})
                 for key, value in cur.items():
                     if isinstance(value, int) and not isinstance(value, bool):
                         run_usage[key] += max(0, value - prev[key])
@@ -1667,9 +1605,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                         for trace_block in trace["blocks"]:
                             if trace_block.get("name") == "user" and trace_block.get("admitted"):
                                 trace_block["tokens"] = block.tokens
-                        capsule.token_count = sum(
-                            block.tokens for block in capsule.blocks if hasattr(block, "tokens")
-                        )
+                        capsule.token_count = sum(block.tokens for block in capsule.blocks if hasattr(block, "tokens"))
                 messages = capsule.to_messages()
                 # P1-02 (#25) acceptance: the log must capture the EXACT
                 # payload the runtime receives — capture BEFORE run, assert
@@ -1727,14 +1663,10 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                         "active_objective": capsule.active_objective,
                         "blocks": [block.to_dict() for block in capsule.blocks],
                     }
-                    response.pending_state["exposed_tools"] = _sorted_tool_names(
-                        getattr(runtime, "tool_names", ())
-                    )
+                    response.pending_state["exposed_tools"] = _sorted_tool_names(getattr(runtime, "tool_names", ()))
 
                 terminal_refusal = (
-                    response.success
-                    and response.pending_approval is None
-                    and override.is_refusal(response.user_text)
+                    response.success and response.pending_approval is None and override.is_refusal(response.user_text)
                 )
                 if terminal_refusal and current_tier < max_tier:
                     current_tier += 1
@@ -1775,16 +1707,12 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                     runtime_objects: list = []
                     runtime_objects.append(_memory_query_from_trace(trace, task_id).to_dict())
                     for compaction_event in trace.get("compaction", []):
-                        runtime_objects.append(
-                            _observation_from_compaction(compaction_event, task_id).to_dict()
-                        )
+                        runtime_objects.append(_observation_from_compaction(compaction_event, task_id).to_dict())
                     response.runtime_objects = runtime_objects
                     # Shared emission (verification/outcome/degraded) — same
                     # helper the approval-resume path uses (quality I-2).
                     self._attach_runtime_objects(response, task_id)
-                    response.runtime_objects = (
-                        runtime_objects + list(response.runtime_objects)
-                    )
+                    response.runtime_objects = runtime_objects + list(response.runtime_objects)
                 except Exception as exc:
                     try:
                         from .subsystem_health import record_failure
@@ -1890,9 +1818,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                         # P1-02 (#25) schema v4: capsule fidelity fields —
                         # the logged messages EXACTLY equal the runtime.run
                         # payload (captured pre-run on the mutated capsule).
-                        capsule_messages=[
-                            dict(m) for m in _sent if isinstance(m, dict)
-                        ],
+                        capsule_messages=[dict(m) for m in _sent if isinstance(m, dict)],
                         capsule_content_hashes=capsule_hashes,
                         capsule_token_count=(
                             int(capsule.token_count)
@@ -1911,9 +1837,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
                     except ImportError:  # TUI top-level deploy (#168)
                         from rune.model_routing import update_profiles_from_traces
 
-                    merged = update_profiles_from_traces(
-                        self._routing_profiles_path(), [asdict(trace_record)]
-                    )
+                    merged = update_profiles_from_traces(self._routing_profiles_path(), [asdict(trace_record)])
                     from .model_routing import save_profiles as _save_profiles
 
                     _save_profiles(self._routing_profiles_path(), merged)
@@ -1940,7 +1864,9 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
 
         # Unreachable in practice (the tier loop always returns or raises),
         # but the declared contract requires a value on all code paths.
-        return OperatorResponse(success=False, model=model, error="task execution ended without a result", degraded=sorted(set(degraded)))
+        return OperatorResponse(
+            success=False, model=model, error="task execution ended without a result", degraded=sorted(set(degraded))
+        )
 
     @staticmethod
     def _finalize_runtime_response(
@@ -1956,9 +1882,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
         response.model = response.model or model
         response.override_tier = override_tier
         response.recall_count = recall_count
-        has_tool_failure = any(
-            event.kind in {"error", "failed"} for event in response.tool_events
-        )
+        has_tool_failure = any(event.kind in {"error", "failed"} for event in response.tool_events)
         valid_final = (
             response.success
             and response.pending_approval is None
@@ -1980,10 +1904,7 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
 
         # Rotate if file exceeds 10MB
         try:
-            if (
-                os.path.exists(log_path)
-                and os.path.getsize(log_path) > 10 * 1024 * 1024
-            ):
+            if os.path.exists(log_path) and os.path.getsize(log_path) > 10 * 1024 * 1024:
                 rotated = log_path + ".old"
                 os.replace(log_path, rotated)
         except OSError:
@@ -2026,11 +1947,72 @@ I have tools to read, write, edit, search, lint, test, format, and commit code.
 _operator = None
 
 
+def _wire_auto_filer() -> None:
+    """Composition-root wiring for the failure auto-filer (#122).
+
+    Creates exactly ONE FailureAutoFiler on the process, backed by
+    create_forge_client() (NullAdapter when unconfigured → dormant by
+    construction) and the canonical event journal as the spool store,
+    then runs one startup reconcile() pass so spooled failures land on
+    recovery. Never raises — a broken wiring is logged and skipped, the
+    operator still boots (degrade-not-crash). RUNE_AUTOFILE_OFF=1
+    disables the whole thing via set_auto_filer's fail-safe gate.
+    """
+    import logging
+    import threading
+
+    try:
+        from .event_bus import _store_path
+        from .failure_auto_filer import AUTOFILE_OFF_ENV, FailureAutoFiler, JsonlSpool
+        from .forge_client import create_forge_client
+    except ImportError:  # TUI top-level deploy (script mode)
+        from event_bus import _store_path  # type: ignore[no-redef]
+        from failure_auto_filer import AUTOFILE_OFF_ENV, FailureAutoFiler, JsonlSpool  # type: ignore[no-redef]
+        from forge_client import create_forge_client  # type: ignore[no-redef]
+
+    log = logging.getLogger(__name__)
+    try:
+        import os as _os
+
+        if _os.environ.get(AUTOFILE_OFF_ENV) == "1":
+            log.info("failure auto-filer disabled (%s=1)", AUTOFILE_OFF_ENV)
+            return
+        client = create_forge_client()
+        spool = JsonlSpool(_store_path())
+        filer = FailureAutoFiler(client, spool)
+        try:
+            from . import subsystem_health
+        except ImportError:  # TUI top-level deploy (script mode)
+            import subsystem_health  # type: ignore[no-redef]
+        subsystem_health.set_auto_filer(filer)
+        # Startup reconcile (the maintenance cadence, #122): replay anything
+        # spooled while the forge was down (own thread — startup latency
+        # stays untouched).
+        threading.Thread(target=_reconcile_guarded, args=(filer,), daemon=True, name="rune-autofile-reconcile").start()
+
+    except Exception:  # noqa: BLE001 - wiring must never block operator boot
+        log.exception("failure auto-filer wiring failed; continuing unwired")
+
+
+def _reconcile_guarded(filer: "_FailureAutoFiler") -> None:
+    import logging
+
+    try:
+        try:
+            from .failure_auto_filer import _LOG
+        except ImportError:  # TUI top-level deploy (script mode)
+            from failure_auto_filer import _LOG  # type: ignore[no-redef]
+        filer.reconcile()
+    except Exception:  # noqa: BLE001 - belt and braces; reconcile never raises
+        _LOG.exception("startup reconcile failed")
+
+
 def get_operator() -> RuneOperator:
     """Get the global Rune operator."""
     global _operator
     if _operator is None:
         _operator = RuneOperator()
+        _wire_auto_filer()
     return _operator
 
 

src/rune/subsystem_health.py

@@ -17,9 +17,11 @@ from __future__ import annotations
 import contextlib
 import json
 import logging
+import os
 import threading
 from logging.handlers import RotatingFileHandler
 from pathlib import Path
+from typing import Protocol
 
 # Subsystem name -> canonical event_type (the event_bus vocabulary).
 _TIER_EVENTS = {0: "task.failed", 1: "capability.failed", 2: "storage.degraded"}
@@ -35,9 +37,7 @@ class SubsystemHealthRegistry:
 
     def record(self, subsystem: str, error: str, tier: int) -> bool:
         with self._lock:
-            entry = self._entries.setdefault(
-                subsystem, {"failures": 0, "last_error": "", "tier": tier}
-            )
+            entry = self._entries.setdefault(subsystem, {"failures": 0, "last_error": "", "tier": tier})
             entry["failures"] += 1
             entry["last_error"] = error
             entry["tier"] = tier
@@ -59,17 +59,50 @@ def registry() -> SubsystemHealthRegistry:
     return _REGISTRY
 
 
+class AutoFiler(Protocol):
+    """The failure auto-filer callback contract (#122): callable with
+    ``(subsystem, error, tier)``, fire-and-forget, never raises."""
+
+    def __call__(self, subsystem: str, error: BaseException | str, tier: int) -> None: ...
+
+
+# Composition-root wiring (#122): the process-default auto-filer. Stays
+# None unless the composition root wires one — unwired, record_failure
+# behaves exactly as before (zero behavior change).
+_AUTO_FILER: AutoFiler | None = None
+
+
+def set_auto_filer(filer: AutoFiler | None) -> None:
+    """Wire (or unwire, None) the process-default failure auto-filer.
+
+    Called once by the composition root (rune_operator.get_operator).
+    Honors RUNE_AUTOFILE_OFF=1: the fail-safe off switch disables the
+    wiring entirely.
+    """
+    global _AUTO_FILER
+    if filer is not None and os.environ.get("RUNE_AUTOFILE_OFF") == "1":
+        return  # fail-safe off switch (#122)
+    _AUTO_FILER = filer
+
+
 def record_failure(
     subsystem: str,
     error: BaseException,
     tier: int = 1,
     registry: SubsystemHealthRegistry | None = None,
     context: str = "",
+    filer: AutoFiler | None = None,
 ) -> bool:
     """Record a subsystem failure: canonical event + ledger + log.
 
     Returns True when the caller should ALSO surface this to the user
     (T0/T1 first occurrence). T2 never surfaces.
+
+    ``filer`` (#122): optional auto-filer callback invoked AFTER the
+    journal record (fire-and-forget — the filer enqueues and returns; it
+    must never block or raise into this failure path). Defaults to the
+    composition-root-wired process default, which is None → zero
+    behavior change. The tier gate lives inside the filer, not here.
     """
     # The observability layer must NEVER raise inside a caller's except
     # block — a raising record_failure masks the very failure it reports
@@ -102,18 +135,28 @@ def record_failure(
                 "record_failure itself failed for subsystem=%s", subsystem
             )
         return False  # last resort: never mask the caller's original error
+    finally:
+        # #122: the auto-filer observes AFTER the journal record. It
+        # ENQUEUES (never blocks on the network) and must never raise into
+        # the caller's except block — degrade-not-crash, same rule as the
+        # observability layer itself.
+        try:
+            target = filer if filer is not None else _AUTO_FILER
+            if target is not None:
+                target(subsystem, error, tier)
+        except Exception:
+            with contextlib.suppress(Exception):
+                logging.getLogger("rune.subsystem_health").exception(
+                    "auto-filer callback failed for subsystem=%s", subsystem
+                )
 
 
 def configure_logging(log_dir: Path | None = None):
     """Give the rune loggers a real destination (they had none — P0-06)."""
     target = Path(log_dir) if log_dir is not None else Path.home() / ".rune" / "logs"
     target.mkdir(parents=True, exist_ok=True)
-    handler = RotatingFileHandler(
-        target / "rune.log", maxBytes=5_000_000, backupCount=3, encoding="utf-8"
-    )
-    handler.setFormatter(
-        logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s")
-    )
+    handler = RotatingFileHandler(target / "rune.log", maxBytes=5_000_000, backupCount=3, encoding="utf-8")
+    handler.setFormatter(logging.Formatter("%(asctime)s %(name)s %(levelname)s %(message)s"))
     root = logging.getLogger("rune")
     root.setLevel(logging.INFO)
     root.addHandler(handler)

tests/test_failure_auto_filer.py

@@ -0,0 +1,408 @@
+"""Failure auto-filer (#122): subsystem failures become forge issues.
+
+Core tests: signature extraction, dedup via NEW server-side search_issues,
+file-once + comment-on-repeat, the ONE tier gate (T0/T1 file, T2 journal
+only), and the spool contract (any client exception lands in the journal
+as a failure_auto_file_spool line, nothing ever raises into the failure
+path, T2 failures produce ZERO client calls).
+
+``FailureAutoFiler.file(...)`` is the synchronous pipeline step (used by
+the worker thread, reconcile, and these tests); ``submit(...)`` is the
+failure-path entry that only enqueues.
+"""
+
+import json
+
+import pytest
+
+from rune import event_bus
+from rune.failure_auto_filer import (
+    FailureAutoFiler,
+    JsonlSpool,
+    error_class_of,
+    excerpt,
+    marker_for,
+    redact,
+    signature_of,
+)
+from rune.forge_client import ForgeClient, NullAdapter
+
+MARKER = "[auto-failure:memory_selector:RuntimeError]"
+
+
+class FakeJournal:
+    """In-memory stand-in for the spool-journal protocol."""
+
+    def __init__(self):
+        self.lines: list[dict] = []
+        self.consumed: set[str] = set()
+
+    def append(self, record: dict) -> None:
+        self.lines.append(record)
+
+    def unconsumed(self) -> list[dict]:
+        return [
+            r
+            for r in self.lines
+            if r.get("kind") == "failure_auto_file_spool" and r.get("spool_id") not in self.consumed
+        ]
+
+    def mark_consumed(self, spool_id: str) -> None:
+        self.consumed.add(spool_id)
+
+
+class FakeForge:
+    """Test double recording every client call the filer makes.
+
+    search_issues mirrors the REAL repo-scoped list endpoint (fix #122
+    gate): server q= filters by marker-in-TITLE only (like real Gitea q),
+    NOT by mirroring the client's filter — the old fake filtered the same
+    way the filer did, making the filer's own check untestable.
+    """
+
+    def __init__(self, search_results=None, fail_with: Exception | None = None):
+        self.search_results = search_results if search_results is not None else []
+        self.fail_with = fail_with
+        self._next_number = 7
+        self.created: list[dict] = []
+        self.commented: list[dict] = []
+        self.searched: list[str] = []
+
+    def search_issues(self, query: str) -> list[dict]:
+        self.searched.append(query)
+        if self.fail_with is not None:
+            raise self.fail_with
+        return [i for i in self.search_results if query in str(i.get("title", ""))]
+
+    def create_issue(self, title: str, body: str) -> dict:
+        if self.fail_with is not None:
+            raise self.fail_with
+        self.created.append({"title": title, "body": body})
+        issue = {"number": self._next_number, "title": title, "state": "open"}
+        self._next_number += 1
+        self.search_results.append(dict(issue))  # findable by later searches
+        return issue
+
+    def add_comment(self, issue_number: int, body: str) -> dict:
+        if self.fail_with is not None:
+            raise self.fail_with
+        self.commented.append({"issue_number": issue_number, "body": body})
+        return {"created": True}
+
+    def get_issue_state(self, issue_number: int) -> dict:
+        return {"state": "open"}
+
+    def list_open_issues(self) -> list[dict]:
+        return []
+
+
+@pytest.fixture(autouse=True)
+def _real_journal(tmp_path, monkeypatch):
+    """Point the canonical journal at tmp so emit() never touches real data."""
+    monkeypatch.setattr("rune.event_bus._store_path", lambda: tmp_path / "events.jsonl")
+    return None
+
+
+# ---------------------------------------------------------------- signature
+
+
+def test_signature_is_subsystem_and_error_class():
+    assert signature_of("memory_selector", "RuntimeError: boom") == ("memory_selector", "RuntimeError")
+
+
+def test_signature_from_plain_error_string_takes_first_token():
+    assert signature_of("cron", "Connection refused to forge") == ("cron", "Connection")
+
+
+def test_signature_accepts_exception_instance():
+    try:
+        raise ValueError("bad payload")
+    except ValueError as exc:
+        assert signature_of("persona", exc) == ("persona", "ValueError")
+
+
+def test_error_class_of_exception_and_string():
+    assert error_class_of(RuntimeError("x")) == "RuntimeError"
+    assert error_class_of("TimeoutError: read timed out") == "TimeoutError"
+    assert error_class_of("connection reset by peer") == "connection"
+
+
+def test_error_class_of_caps_single_token_at_64_chars():
+    # #122 gate: a 10KB single-token error must not make a 10KB marker.
+    huge = "E" * 10000
+    capped = error_class_of(f"{huge}: boom")
+    assert len(capped) == 64
+    assert len(marker_for("subsystem", capped)) < 100  # marker stays bounded
+    assert error_class_of("RuntimeError: ok") == "RuntimeError"  # short: untouched
+
+
+def test_marker_format():
+    assert marker_for("memory_selector", "RuntimeError") == MARKER
+
+
+# ---------------------------------------------------------------- file path
+
+
+def test_first_occurrence_creates_issue_with_marker_title():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=1)
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    assert len(forge.created) == 1
+    assert forge.created[0]["title"].startswith(MARKER)
+    assert forge.commented == []
+
+
+def test_repeat_occurrence_comments_on_existing_issue():
+    forge = FakeForge(search_results=[{"number": 7, "title": MARKER + " memory_selector"}])
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=1)
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    assert forge.created == []  # dedup: never a second issue per signature
+    assert len(forge.commented) == 1
+    assert "+1 occurrences" in forge.commented[0]["body"]
+    assert "RuntimeError" in forge.commented[0]["body"]
+
+
+def test_each_repeat_comments_never_creates():
+    forge = FakeForge(search_results=[{"number": 7, "title": MARKER + " memory_selector"}])
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=1)
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    filer.file("memory_selector", "RuntimeError: another payload", tier=1)
+    assert forge.created == []
+    assert len(forge.commented) == 2
+    assert "another payload" in forge.commented[1]["body"]
+
+
+def test_search_results_without_marker_in_title_are_ignored():
+    # Server-side q= is fuzzy (title OR body); only a title match dedups.
+    forge = FakeForge(search_results=[{"number": 9, "title": "unrelated discussion of the marker"}])
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=1)
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    assert len(forge.created) == 1
+    assert forge.commented == []
+
+
+def test_tier_gate_t2_produces_zero_client_calls():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal())
+    filer.file("cron", "Connection refused", tier=2)
+    assert forge.searched == []
+    assert forge.created == []
+    assert forge.commented == []
+
+
+def test_tier_gate_t0_and_t1_file():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=1)
+    filer.file("memory_selector", "RuntimeError: a", tier=0)
+    filer.file("warm_router", "RuntimeError: b", tier=1)
+    assert len(forge.created) == 2
+
+
+def test_create_issue_body_carries_tier_and_count():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal())
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    body = forge.created[0]["body"]
+    assert "tier=1" in body
+    assert "occurrences=1" in body
+
+
+def test_error_excerpt_is_truncated():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal())
+    filer.file("memory_selector", "RuntimeError: " + ("x" * 500), tier=1)
+    body = forge.created[0]["body"]
+    assert ("x" * 500) not in body
+    assert ("x" * 100) in body
+
+
+# ---------------------------------------------------------------- redaction
+
+
+def test_redact_strips_query_string_credentials():
+    assert (
+        redact("GET https://forge.example.com/api?token=supersecret&page=2 failed")
+        == "GET https://forge.example.com/api?token=<redacted>&page=2 failed"
+    )
+    assert "password=hunter2" not in redact("postgres://host/db?password=hunter2&x=1")
+    assert "api_key=abcd" not in redact("https://api.example.com/v1?api_key=abcd&y=2")
+    assert "key=zzz" not in redact("https://x.example/?key=zzz")
+
+
+def test_redact_strips_url_basic_auth_and_dsn_credentials():
+    out = redact("connect failed for postgres://alice:hunter2@db.example.com:5432/rune")
+    assert "hunter2" not in out
+    assert "alice" not in out
+    assert "postgres://<redacted>:<redacted>@db.example.com:5432/rune" in out
+    out2 = redact("redis://:s3cr3t@cache.example.com:6379/0")
+    assert "s3cr3t" not in out2
+
+
+def test_redact_strips_bearer_and_assignment_credentials():
+    out = redact("Authorization: Bearer eyJhbGciOi.eyJzdWIi.SflKxwRJSMeK")
+    assert "eyJhbGciOi" not in out
+    assert "<redacted>" in out
+    out2 = redact("login failed with password='hunter2' and secret=abcdef1234")
+    assert "hunter2" not in out2
+    assert "abcdef1234" not in out2
+
+
+def test_redact_keeps_error_class_and_shape_visible():
+    out = redact("RuntimeError: connection to https://user:pass@forge.example.com refused (port 443)")
+    assert out.startswith("RuntimeError: ")
+    assert "refused (port 443)" in out  # non-credential detail survives
+    # Redaction is deterministic on the credential span only:
+    body = "connection to https://user:pass@forge.example.com refused (port 443)"
+    assert out == "RuntimeError: " + redact(body)
+    assert redact("plain error, nothing sensitive") == "plain error, nothing sensitive"
+
+
+def test_excerpt_redacts_credentials():
+    # excerpt() is what lands in persistent forge issues and spool lines:
+    # it must redact before returning.
+    assert "hunter2" not in excerpt("postgres://bob:hunter2@db/rune down")
+    assert "<redacted>" in excerpt("GET https://forge.example.com/api?token=topsecret")
+
+
+def test_credentialed_error_never_lands_raw_in_issue_or_spool():
+    forge = FakeForge(fail_with=ConnectionError("connect: https://user:hunter2@forge.example.com/api?token=t0psecret"))
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1)
+    filer.file("memory_selector", "RuntimeError: dsn postgres://alice:s3cr3t@db.example.com/rune", tier=1)
+    spooled = journal.unconsumed()[0]
+    for blob in (json.dumps(forge.created), json.dumps(spooled)):
+        for secret in ("hunter2", "s3cr3t", "t0psecret"):
+            assert secret not in blob
+
+
+# ---------------------------------------------------------------- spool path
+
+
+def test_client_exception_spools_and_never_raises():
+    forge = FakeForge(fail_with=ConnectionError("forge down"))
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal)
+    # Must NOT raise.
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    pending = journal.unconsumed()
+    assert len(pending) == 1
+    line = pending[0]
+    assert line["kind"] == "failure_auto_file_spool"
+    assert line["signature"] == ["memory_selector", "RuntimeError"]
+    assert line["tier"] == 1
+    assert "boom" in line["error"]
+    assert "timestamp" in line
+    assert forge.created == []
+
+
+def test_spool_write_failure_still_never_raises():
+    forge = FakeForge()
+    journal = FakeJournal()
+
+    def explode(record):
+        raise OSError("disk full")
+
+    journal.append = explode
+    filer = FailureAutoFiler(forge, journal)
+    # Neither forge failure nor journal failure may escape.
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+
+
+def test_jsonl_spool_line_is_durable_and_json(tmp_path):
+    path = tmp_path / "events.jsonl"
+    forge = FakeForge(fail_with=ConnectionError("forge down"))
+    spool = JsonlSpool(path)
+    filer = FailureAutoFiler(forge, spool)
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    rows = [json.loads(line) for line in path.read_text().splitlines()]
+    assert rows and rows[-1]["kind"] == "failure_auto_file_spool"
+    assert spool.unconsumed()[0]["spool_id"] == rows[-1]["spool_id"]
+
+
+# ---------------------------------------------------------------- protocol
+
+
+def test_null_adapter_search_returns_empty():
+    assert NullAdapter().search_issues("anything") == []
+
+
+def test_filer_accepts_protocol_clients():
+    # Structural check: the fake satisfies ForgeClient incl. search_issues.
+    assert issubclass(FakeForge, ForgeClient)
+    assert issubclass(NullAdapter, ForgeClient)
+
+
+# ------------------------------------------------- queue / worker / kill tests
+
+
+def test_submit_enqueues_and_drain_delivers_writes():
+    forge = FakeForge(search_results=[{"number": 7, "title": MARKER + " memory_selector"}])
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=1)
+    filer.submit("cron", "Connection refused", tier=2)  # gated: zero client calls
+    filer.submit("memory_selector", "RuntimeError: a", tier=1)
+    filer.submit("memory_selector", "RuntimeError: b", tier=1)
+    filer.drain(timeout=5)
+    assert forge.created == []  # existing issue deduped
+    # Worker may batch both queued occurrences into one comment, or handle
+    # them separately — either way: comments only, never a second issue.
+    assert 1 <= len(forge.commented) <= 2
+
+
+def test_queue_is_bounded_and_overflow_drops_spools_and_logs(caplog):
+    # #122 gate: the queue must be bounded (a forge outage must not balloon
+    # memory) and an overflow occurrence is DROPPED from the queue but
+    # SPOOLED to the journal (nothing lost) + logged — never blocked.
+    import logging
+
+    forge = FakeForge()
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, wait=True)
+    assert filer._queue.maxsize == 1000
+    for i in range(1000):
+        filer.submit("memory_selector", f"RuntimeError: fill {i}", tier=1)
+    assert filer._queue.full()
+    with caplog.at_level(logging.WARNING, logger="rune.failure_auto_filer"):
+        filer.submit("memory_selector", "RuntimeError: dropped-occurrence", tier=1)
+    assert filer._queue.full()  # queue did not grow past maxsize
+    pending = journal.unconsumed()
+    assert any(r["error"].endswith("dropped-occurrence") for r in pending)  # spooled
+    assert any("queue full" in rec.getMessage() for rec in caplog.records)  # logged
+
+
+def test_worker_exception_never_propagates():
+    forge = FakeForge(fail_with=ConnectionError("forge down"))
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal)
+    filer.submit("memory_selector", "RuntimeError: boom", tier=1)
+    filer.drain(timeout=5)  # worker hit the exception; must not raise here
+    assert len(journal.unconsumed()) == 1
+
+
+def test_t2_kill_test_zero_client_calls_through_worker():
+    # Frozen kill test: a T2 failure must produce ZERO client calls even
+    # when the worker thread drains the queue.
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal())
+    filer.submit("cron", "Connection refused", tier=2)
+    filer.submit("cron", "JSONDecodeError while parsing", tier=2)
+    filer.drain(timeout=5)
+    assert forge.searched == []
+    assert forge.created == []
+    assert forge.commented == []
+
+
+def test_forge_down_kill_test_never_raises_into_failure_path():
+    forge = FakeForge(fail_with=ConnectionError("forge down"))
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal)
+    try:
+        filer.submit("memory_selector", "RuntimeError: boom", tier=1)
+        filer.drain(timeout=5)
+    except Exception as exc:  # pragma: no cover - the assertion below bites
+        pytest.fail(f"filer raised into the failure path: {exc!r}")
+    assert len(journal.unconsumed()) == 1
+
+
+def test_event_bus_emit_unchanged_after_filer_import():
+    # Spool rides the canonical journal; regular emit keeps working.
+    event_bus.emit("user.message", source="t", payload={"k": "v"})

tests/test_failure_auto_filer_task2.py

@@ -0,0 +1,416 @@
+"""#122 Task 2: burst coalescing + spool reconciliation.
+
+Burst coalescing: after the first occurrence files, subsequent occurrences
+within the dedup window batch into ONE comment per batch_size occurrences
+or 10 minutes (whichever first) — a 3am crash-loop must not write 200
+comments. The worker additionally coalesces everything already queued
+into one write per signature. Reconcile: replay unconsumed spool lines
+through the same pipeline (dedup re-applies via search_issues); filed
+lines are marked consumed; forge still down → spool preserved, never
+raises.
+"""
+
+import json
+import time
+from pathlib import Path
+
+from rune.failure_auto_filer import (
+    BATCH_INTERVAL_SECONDS,
+    FailureAutoFiler,
+    JsonlSpool,
+)
+from rune.forge_client import NullAdapter
+
+from tests.test_failure_auto_filer import FakeForge, FakeJournal, MARKER
+
+
+# ------------------------------------------------------------ burst coalescing
+
+
+def test_burst_of_50_produces_at_most_2_client_writes():
+    # First occurrence files; everything already queued coalesces into one
+    # comment; the tail (< batch_size) stays pending: 1 create + ≤1 comment.
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=True)
+    for i in range(50):
+        filer.submit("memory_selector", f"RuntimeError: boom {i}", tier=1)
+    filer.drain(timeout=5)  # the whole backlog coalesces into ONE write
+    assert len(forge.created) == 1
+    assert len(forge.created) + len(forge.commented) <= 2
+
+
+def test_batch_flushes_every_batch_size_occurrences():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
+    for i in range(25):
+        filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
+    assert len(forge.created) == 1
+    # Comments when pending EXCEEDS the batch size (occurrences 11 and 21):
+    # one comment per 10 occurrences, tail pending.
+    assert len(forge.commented) == 2
+    assert "+10 occurrences" in forge.commented[0]["body"]
+    assert "+20 occurrences" in forge.commented[1]["body"]
+
+
+def test_batch_flushes_after_ten_minutes():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
+    start = time.monotonic()
+    filer.file("memory_selector", "RuntimeError: boom 0", tier=1, _now=start)  # files
+    later = start + BATCH_INTERVAL_SECONDS + 1  # 10 minutes pass
+    filer.file("memory_selector", "RuntimeError: boom 1", tier=1, _now=later)
+    assert len(forge.created) == 1
+    assert len(forge.commented) == 1  # window expired -> flush on next occurrence
+    assert "+1 occurrences" in forge.commented[0]["body"]
+
+
+def test_burst_counter_is_per_signature():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
+    for i in range(12):
+        filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
+    for i in range(3):
+        filer.file("warm_router", f"TimeoutError: slow {i}", tier=1)
+    assert len(forge.created) == 2  # one per signature
+    assert len(forge.commented) == 1  # memory_selector hit 10; warm_router (3) not yet
+
+
+def test_burst_tail_below_batch_size_stays_pending_until_window():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
+    for i in range(7):
+        filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
+    filer.drain(timeout=5)
+    assert len(forge.created) == 1
+    assert len(forge.commented) == 0  # 7 < batch_size, window not expired
+
+
+def test_burst_batched_comment_carries_last_error_excerpt():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=True)
+    for i in range(11):  # 11th occurrence crosses the batch threshold
+        filer.file("memory_selector", f"RuntimeError: boom {i}", tier=1)
+    filer.drain(timeout=5)
+    assert len(forge.commented) == 1
+    assert "+10 occurrences" in forge.commented[0]["body"]
+    assert "boom 10" in forge.commented[0]["body"]  # last error of the batch
+
+
+def test_window_resets_after_flush():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), batch_size=10, wait=False)
+    base = time.monotonic()
+    filer.file("memory_selector", "RuntimeError: a0", tier=1, _now=base)  # files
+    for i in range(1, 12):
+        filer.file("memory_selector", f"RuntimeError: a{i}", tier=1, _now=base + i * 0.1)
+    assert len(forge.commented) == 1  # flush at pending=11 (> batch_size)
+    # The window restarted at that flush: no second comment for 3 more.
+    for i in range(3):
+        filer.file("memory_selector", f"RuntimeError: b{i}", tier=1, _now=base + 2 + i * 0.1)
+    assert len(forge.commented) == 1
+
+
+# ---------------------------------------------------------------- reconcile
+
+
+def test_reconcile_groups_spool_lines_by_signature_into_one_write():
+    # #122 gate: 10 spooled lines of ONE signature (a long outage) must
+    # produce exactly ONE client write on reconcile — not a comment flood.
+    forge = FakeForge()
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    for i in range(10):
+        journal.append(
+            {
+                "kind": "failure_auto_file_spool",
+                "spool_id": f"spool-{i}",
+                "timestamp": "t",
+                "signature": ["memory_selector", "RuntimeError"],
+                "tier": 1,
+                "count": 1,
+                "error": f"RuntimeError: boom {i}",
+            }
+        )
+    report = filer.reconcile()
+    assert len(forge.created) + len(forge.commented) == 1  # ONE write
+    assert len(forge.created) == 1  # dedup still applies: issue created once
+    assert "occurrences=10" in forge.created[0]["body"]  # aggregated count
+    assert "boom 9" in forge.created[0]["body"]  # last error of the group
+    assert report["filed"] == 10
+    assert journal.unconsumed() == []
+
+
+def test_reconcile_group_comments_once_with_offline_count():
+    # Same grouping when the issue already exists: ONE aggregated comment.
+    forge = FakeForge(search_results=[{"number": 7, "title": MARKER + " memory_selector"}])
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    for i in range(4):
+        journal.append(
+            {
+                "kind": "failure_auto_file_spool",
+                "spool_id": f"spool-{i}",
+                "timestamp": "t",
+                "signature": ["memory_selector", "RuntimeError"],
+                "tier": 1,
+                "count": 1,
+                "error": f"RuntimeError: boom {i}",
+            }
+        )
+    report = filer.reconcile()
+    assert forge.created == []
+    assert len(forge.commented) == 1
+    assert "+4 occurrences while offline" in forge.commented[0]["body"]
+    assert "boom 3" in forge.commented[0]["body"]
+    assert report["filed"] == 4
+    assert journal.unconsumed() == []
+
+
+def test_reconcile_replays_spool_and_marks_consumed():
+    forge = FakeForge()
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    forge.fail_with = ConnectionError("forge down")
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    assert len(journal.unconsumed()) == 1
+    forge.fail_with = None
+    filer.reconcile()
+    assert len(forge.created) == 1
+    assert forge.created[0]["title"].startswith(MARKER)
+    assert journal.unconsumed() == []  # marked consumed
+
+
+def test_reconcile_dedups_against_live_issues():
+    forge = FakeForge(search_results=[{"number": 7, "title": MARKER + " memory_selector"}])
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    forge.fail_with = ConnectionError("forge down")
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    forge.fail_with = None
+    filer.reconcile()
+    assert forge.created == []  # dedup re-applied: comment, not a second issue
+    assert len(forge.commented) == 1
+    assert journal.unconsumed() == []
+
+
+def test_reconcile_with_forge_still_down_preserves_spool():
+    forge = FakeForge(fail_with=ConnectionError("forge still down"))
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    pending_before = [dict(r) for r in journal.unconsumed()]
+    filer.reconcile()  # must not raise
+    assert journal.unconsumed() == pending_before  # preserved, still pending
+    assert forge.created == []
+
+
+def test_reconcile_on_jsonl_spool_marks_consumed_in_sidecar(tmp_path):
+    path = tmp_path / "events.jsonl"
+    forge = FakeForge()
+    spool = JsonlSpool(path)
+    filer = FailureAutoFiler(forge, spool, batch_size=1, wait=False)
+    forge.fail_with = ConnectionError("down")
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    assert len(spool.unconsumed()) == 1
+    canonical_before = path.read_bytes()
+    forge.fail_with = None
+    filer.reconcile()
+    assert spool.unconsumed() == []
+    # Canonical journal untouched — consumption lives in the sidecar only.
+    assert path.read_bytes() == canonical_before
+    assert (tmp_path / "events.jsonl.consumed").exists()
+
+
+def _spool_record(spool_id: str) -> dict:
+    return {
+        "kind": "failure_auto_file_spool",
+        "spool_id": spool_id,
+        "timestamp": "t",
+        "signature": ["s", "E"],
+        "tier": 1,
+        "count": 1,
+        "error": "E: x",
+    }
+
+
+def test_mark_consumed_never_rewrites_the_canonical_journal(tmp_path):
+    # #122 gate: mark_consumed must be append-only. The canonical journal
+    # is shared with DurableEventStore — a read-modify-write there is
+    # racy and a crash mid-rewrite poisons the canonical store.
+    path = tmp_path / "events.jsonl"
+    spool = JsonlSpool(path)
+    spool.append(_spool_record("spool-x"))
+    before = path.read_bytes()
+    spool.mark_consumed("spool-x")
+    assert path.read_bytes() == before  # byte-identical: no rewrite, ever
+    assert spool.unconsumed() == []  # consumption recorded in the sidecar
+    assert (tmp_path / "events.jsonl.consumed").read_text().splitlines() == ["spool-x"]
+
+
+def test_concurrent_append_between_read_and_mark_is_not_lost(tmp_path, monkeypatch):
+    # A DurableEventStore.append landing while mark_consumed runs must
+    # survive. The old implementation read the journal, then rewrote it —
+    # the append landed in between and was silently DESTROYED.
+    import builtins
+
+    path = tmp_path / "events.jsonl"
+    spool = JsonlSpool(path)
+    spool.append(_spool_record("spool-x"))
+    real_open = builtins.open
+    state = {"fired": False}
+
+    def racing_open(file, mode="r", *a, **k):
+        # The concurrent O_APPEND lands while mark_consumed is mid-flight
+        # (fires on the first write/append-mode open inside the mark).
+        if not state["fired"] and any(m in mode for m in ("a", "w", "x")):
+            state["fired"] = True
+            with real_open(path, "a", encoding="utf-8") as fh:
+                fh.write(json.dumps({"event_id": "e-race", "event_type": "task.failed"}) + "\n")
+        return real_open(file, mode, *a, **k)
+
+    monkeypatch.setattr(builtins, "open", racing_open)
+    spool.mark_consumed("spool-x")
+    rows = [json.loads(line) for line in path.read_text().splitlines()]
+    assert any(r.get("event_id") == "e-race" for r in rows)  # NOT destroyed
+    assert spool.unconsumed() == []
+
+
+def test_reconcile_routes_tier2_line_through_the_tier_gate():
+    # #122 gate: a crafted tier=2 spool line must stay journal-only on
+    # replay — the SAME one tier gate as the live path (no _write bypass).
+    forge = FakeForge()
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, wait=False)
+    journal.append(_spool_record("spool-t2") | {"tier": 2})
+    report = filer.reconcile()
+    assert forge.searched == []  # zero client calls: journal-only on replay
+    assert forge.created == []
+    assert forge.commented == []
+    assert report["skipped"] == 1
+    assert journal.unconsumed() == []  # resolved: consumed, never written
+
+
+def test_reconcile_mixed_group_tier2_lines_dropped_tier1_filed():
+    # A group mixing tiers keeps filing the T0/T1 lines; only the T2
+    # lines are gate-dropped (journal-only).
+    forge = FakeForge()
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    journal.append(_spool_record("spool-t1") | {"tier": 1})
+    journal.append(_spool_record("spool-t2") | {"tier": 2})
+    report = filer.reconcile()
+    assert len(forge.created) == 1
+    assert report["filed"] == 1
+    assert report["skipped"] == 1
+    assert journal.unconsumed() == []
+
+
+def test_reconcile_and_worker_single_flight_per_signature():
+    # #122 gate: the reconcile thread and the worker racing on the SAME
+    # signature must not duplicate the issue — single-flight lets exactly
+    # one write land per signature at a time.
+    import threading
+
+    forge = FakeForge()
+    entries = {"n": 0}
+    release = threading.Event()
+
+    def slow_search(query):
+        entries["n"] += 1
+        if entries["n"] == 1:
+            release.wait(timeout=5)  # hold the first writer inside the window
+        # Like a real forge: issues created earlier ARE found by later
+        # searches, and search results carry the issue number.
+        return [
+            {"number": n, "title": i["title"]} for n, i in enumerate(forge.created, start=100) if query in i["title"]
+        ]
+
+    forge.search_issues = slow_search  # type: ignore[method-assign]
+    journal = FakeJournal()
+    journal.append(_spool_record("spool-race"))
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    filer.submit("s", "E: x", tier=1)  # WORKER path (submit), same signature
+    deadline = time.monotonic() + 5
+    while entries["n"] < 1 and time.monotonic() < deadline:
+        time.sleep(0.01)  # worker is inside the write window
+    reconcile_report = filer.reconcile()  # races the worker on the same sig
+    release.set()
+    filer.drain(timeout=5)
+    assert len(forge.created) == 1  # NEVER a duplicate issue
+    assert forge.commented == []
+    assert len(forge.created) + len(forge.commented) == 1
+    # The racing replay stayed spooled (not falsely consumed): the
+    # in-flight writer owned the signature; a later replay dedups.
+    assert reconcile_report["filed"] == 0
+    filer.reconcile()
+    assert len(forge.created) == 1  # dedup re-applies: comment, not issue 2
+    assert len(forge.commented) == 1
+    assert journal.unconsumed() == []
+
+
+def test_reconcile_empty_spool_is_a_noop():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), wait=False)
+    filer.reconcile()
+    assert forge.searched == []
+    assert forge.created == []
+
+
+def test_reconcile_files_and_consumes_multiple_spooled_signatures():
+    forge = FakeForge()
+    journal = FakeJournal()
+    filer = FailureAutoFiler(forge, journal, batch_size=1, wait=False)
+    forge.fail_with = ConnectionError("down")
+    filer.file("memory_selector", "RuntimeError: boom", tier=1)
+    filer.file("warm_router", "TimeoutError: slow", tier=1)
+    assert len(journal.unconsumed()) == 2
+    forge.fail_with = None
+    filer.reconcile()
+    assert len(forge.created) == 2
+    assert journal.unconsumed() == []
+
+
+def test_null_adapter_reconcile_consumes_spool():
+    # NullAdapter never raises; a search returns [] and create_issue is a
+    # no-op dict — the filer treats that as "nothing was filed" and keeps
+    # the line spooled rather than falsely consuming it.
+    journal = FakeJournal()
+    filer = FailureAutoFiler(NullAdapter(), journal, batch_size=1, wait=False)
+    journal.append(
+        {
+            "kind": "failure_auto_file_spool",
+            "spool_id": "spool-1",
+            "timestamp": "t",
+            "signature": ["memory_selector", "RuntimeError"],
+            "tier": 1,
+            "count": 1,
+            "error": "RuntimeError: boom",
+        }
+    )
+    filer.reconcile()
+    assert journal.unconsumed() != []  # not filed → stays spooled
+
+
+def test_reconcile_never_raises_when_journal_read_fails():
+    class ExplodingJournal(FakeJournal):
+        def unconsumed(self):
+            raise OSError("disk gone")
+
+    filer = FailureAutoFiler(FakeForge(), ExplodingJournal(), wait=False)
+    filer.reconcile()  # must not raise
+
+
+def test_reconcile_malformed_line_is_skipped_without_crash():
+    # #122 gate: assert against the REAL fake state (the old version ended
+    # in forge_created_not_called(), which returned True unconditionally).
+    journal = FakeJournal()
+    journal.lines.append({"kind": "failure_auto_file_spool", "spool_id": "bad", "tier": 1})
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, journal, wait=False)
+    report = filer.reconcile()
+    assert forge.created == []  # nothing was written to the forge
+    assert forge.commented == []
+    assert forge.searched == []
+    assert report["replayed"] == 1
+    assert report["failed"] == 1
+    assert journal.unconsumed() == [{"kind": "failure_auto_file_spool", "spool_id": "bad", "tier": 1}]

tests/test_failure_auto_filer_wiring.py

@@ -0,0 +1,290 @@
+"""#122 Task 3: wiring — record_failure filer param, composition root,
+RUNE_AUTOFILE_OFF, and the frozen call-site compatibility contract.
+
+The two existing call-site families (capability_broker.py:122,
+context_broker.py:315) must keep working UNCHANGED when no filer is
+wired; when the composition root wires one _auto_filer, they file.
+"""
+
+import time
+
+import pytest
+
+from rune import rune_operator
+from rune import event_bus, subsystem_health
+from rune.capability_broker import select_for_job
+from rune.failure_auto_filer import (
+    AUTOFILE_OFF_ENV,
+    FailureAutoFiler,
+    JsonlSpool,
+)
+from rune.subsystem_health import SubsystemHealthRegistry, record_failure, set_auto_filer
+
+from tests.test_failure_auto_filer import FakeForge, FakeJournal
+
+
+class RecordingFiler:
+    """Callable filer stand-in matching the callback contract."""
+
+    def __init__(self):
+        self.calls: list[tuple[str, str, int]] = []
+        self.fail_with: Exception | None = None
+
+    def __call__(self, subsystem: str, error, tier: int) -> None:
+        if self.fail_with is not None:
+            raise self.fail_with
+        self.calls.append((subsystem, str(error), int(tier)))
+
+
+@pytest.fixture(autouse=True)
+def _isolated(tmp_path, monkeypatch):
+    """Isolate journal + process default filer for every test."""
+    monkeypatch.setattr("rune.event_bus._store_path", lambda: tmp_path / "events.jsonl")
+    monkeypatch.setattr(subsystem_health, "_AUTO_FILER", None)
+    monkeypatch.setattr(rune_operator, "_operator", None)
+    monkeypatch.delenv("FORGE_URL", raising=False)
+    monkeypatch.delenv("FORGE_TOKEN", raising=False)
+    monkeypatch.delenv(AUTOFILE_OFF_ENV, raising=False)
+    yield
+    subsystem_health.set_auto_filer(None)
+
+
+# ------------------------------------------------- record_failure filer param
+
+
+def test_filer_param_is_typed_not_bare_any():
+    # #122 gate: the filer callback carries a real protocol type, not Any.
+    import typing
+
+    hints = typing.get_type_hints(subsystem_health.record_failure)
+    assert hints["filer"] == subsystem_health.AutoFiler | None
+    hints_set = typing.get_type_hints(subsystem_health.set_auto_filer)
+    assert hints_set["filer"] == subsystem_health.AutoFiler | None
+
+
+def test_unwired_record_failure_is_unchanged():
+    # No filer param, no process default: registry entry recorded, return
+    # contract intact, nothing else happens.
+    registry = SubsystemHealthRegistry()
+    assert record_failure("memory_selector", RuntimeError("boom"), tier=1, registry=registry)
+    assert registry.snapshot()["memory_selector"]["failures"] == 1
+
+
+def test_record_failure_explicit_filer_param_files():
+    filer = RecordingFiler()
+    registry = SubsystemHealthRegistry()
+    record_failure("memory_selector", RuntimeError("boom"), tier=1, registry=registry, filer=filer)
+    assert len(filer.calls) == 1
+    subsystem_name, error, tier = filer.calls[0]
+    assert subsystem_name == "memory_selector" and tier == 1
+    assert "boom" in str(error)
+    assert registry.snapshot()["memory_selector"]["failures"] == 1
+
+
+def test_record_failure_return_contract_unchanged_with_filer():
+    filer = RecordingFiler()
+    registry = SubsystemHealthRegistry()
+    first = record_failure("memory_selector", RuntimeError("a"), tier=1, registry=registry, filer=filer)
+    second = record_failure("memory_selector", RuntimeError("b"), tier=1, registry=registry, filer=filer)
+    assert first and not second  # anti-spam surfacing untouched by wiring
+    assert not record_failure("cron", RuntimeError("x"), tier=2, registry=registry, filer=filer)
+
+
+def test_filer_exception_never_propagates_from_record_failure():
+    filer = RecordingFiler()
+    filer.fail_with = RuntimeError("filer exploded")
+    registry = SubsystemHealthRegistry()
+    # Must not raise; the registry record still lands.
+    assert record_failure("memory_selector", RuntimeError("boom"), tier=1, registry=registry, filer=filer)
+    assert registry.snapshot()["memory_selector"]["failures"] == 1
+
+
+def test_process_default_filer_wired_by_composition():
+    filer = RecordingFiler()
+    subsystem_health.set_auto_filer(filer)
+    registry = SubsystemHealthRegistry()
+    record_failure("memory_selector", RuntimeError("boom"), tier=1, registry=registry)
+    assert len(filer.calls) == 1
+    subsystem_name, error, tier = filer.calls[0]
+    assert subsystem_name == "memory_selector" and tier == 1
+    assert "boom" in str(error)
+
+
+def test_explicit_filer_param_overrides_process_default():
+    default_filer = RecordingFiler()
+    explicit_filer = RecordingFiler()
+    subsystem_health.set_auto_filer(default_filer)
+    registry = SubsystemHealthRegistry()
+    record_failure("memory_selector", RuntimeError("boom"), tier=1, registry=registry, filer=explicit_filer)
+    assert explicit_filer.calls and default_filer.calls == []
+
+
+# --------------------------------------------------------------- tier gate
+
+
+def test_t2_reaches_filer_but_filer_gates_it():
+    # The gate lives in ONE place (the filer): record_failure forwards all
+    # tiers; the filer's submit drops T2 before any queue traffic.
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), wait=True)
+    subsystem_health.set_auto_filer(filer)
+    registry = SubsystemHealthRegistry()
+    record_failure("cron", "Connection refused", tier=2, registry=registry)
+    filer.drain(timeout=5)
+    assert forge.searched == []
+    assert forge.created == []
+    assert forge.commented == []
+
+
+def test_t1_files_through_process_default_filer():
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), wait=True)
+    subsystem_health.set_auto_filer(filer)
+    registry = SubsystemHealthRegistry()
+    record_failure("memory_selector", "RuntimeError: boom", tier=1, registry=registry)
+    filer.drain(timeout=5)
+    assert len(forge.created) == 1
+    assert forge.created[0]["title"].startswith("[auto-failure:memory_selector:RuntimeError]")
+
+
+# -------------------------------------------------------- call-site contract
+
+
+def test_capability_broker_call_site_unchanged_when_unwired(monkeypatch):
+    def broken_select(description, need_tools):
+        raise RuntimeError("selector exploded")
+
+    monkeypatch.setattr("rune.capability_broker._select_for_job", broken_select)
+    result = select_for_job("write tests", need_tools=True)
+    assert result["reason"].startswith("fallback full toolset")  # degrade-not-crash
+
+
+def test_capability_broker_call_site_files_when_wired(monkeypatch):
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), wait=True)
+    subsystem_health.set_auto_filer(filer)
+
+    def broken_select(description, need_tools):
+        raise RuntimeError("selector exploded")
+
+    monkeypatch.setattr("rune.capability_broker._select_for_job", broken_select)
+    result = select_for_job("write tests", need_tools=True)
+    assert result["reason"].startswith("fallback full toolset")
+    filer.drain(timeout=5)
+    assert len(forge.created) == 1
+    assert forge.created[0]["title"].startswith("[auto-failure:capability_selection:RuntimeError]")
+
+
+def test_context_broker_t2_call_site_stays_journal_only_when_wired(monkeypatch):
+    # context_broker.py:315 records history_archive failures at tier 2 —
+    # the frozen kill test through a REAL call site: zero client calls.
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, FakeJournal(), wait=True)
+    subsystem_health.set_auto_filer(filer)
+    registry = SubsystemHealthRegistry()
+    record_failure("history_archive", OSError("disk on fire"), tier=2, registry=registry)
+    filer.drain(timeout=5)
+    assert forge.searched == []
+    assert forge.created == []
+    assert forge.commented == []
+
+
+# ---------------------------------------------------------- composition root
+
+
+def _reset_and_get_operator(monkeypatch, forge):
+    monkeypatch.setattr("rune.forge_client.create_forge_client", lambda config=None: forge)
+    return rune_operator.get_operator()
+
+
+def test_composition_root_wires_one_auto_filer(monkeypatch):
+    forge = FakeForge()
+    monkeypatch.setenv("FORGE_URL", "https://forge.example.com")
+    monkeypatch.setenv("FORGE_TOKEN", "tok")
+    _reset_and_get_operator(monkeypatch, forge)
+    assert isinstance(subsystem_health._AUTO_FILER, FailureAutoFiler)
+    # A wired failure files through the composition-root client.
+    record_failure("memory_selector", "RuntimeError: boom", tier=1)
+    subsystem_health._AUTO_FILER.drain(timeout=5)
+    assert len(forge.created) == 1
+
+
+def test_composition_root_reconciles_spool_at_startup(monkeypatch, tmp_path):
+    # A spooled line from a previous run lands once the operator boots.
+    forge = FakeForge()
+    monkeypatch.setenv("FORGE_URL", "https://forge.example.com")
+    monkeypatch.setenv("FORGE_TOKEN", "tok")
+    spool = JsonlSpool(tmp_path / "events.jsonl")
+    spool.append(
+        {
+            "kind": "failure_auto_file_spool",
+            "spool_id": "spool-1",
+            "timestamp": "t",
+            "signature": ["memory_selector", "RuntimeError"],
+            "tier": 1,
+            "count": 1,
+            "error": "RuntimeError: boom",
+        }
+    )
+    monkeypatch.setattr("rune.event_bus._store_path", lambda: tmp_path / "events.jsonl")
+    _reset_and_get_operator(monkeypatch, forge)
+    deadline = time.monotonic() + 5
+    while not forge.created and time.monotonic() < deadline:
+        time.sleep(0.02)  # startup reconcile runs on its own daemon thread
+    assert len(forge.created) == 1
+    assert spool.unconsumed() == []
+
+
+def test_composition_root_wiring_never_raises(monkeypatch):
+    def explode(config=None):
+        raise RuntimeError("forge factory broken")
+
+    monkeypatch.setenv("FORGE_URL", "https://forge.example.com")
+    monkeypatch.setenv("FORGE_TOKEN", "tok")
+    monkeypatch.setattr("rune.forge_client.create_forge_client", explode)
+    op = rune_operator.get_operator()  # must not raise
+    assert op is not None
+
+
+def test_rune_autofile_off_disables_entirely(monkeypatch):
+    forge = FakeForge()
+    monkeypatch.setenv("FORGE_URL", "https://forge.example.com")
+    monkeypatch.setenv("FORGE_TOKEN", "tok")
+    monkeypatch.setenv(AUTOFILE_OFF_ENV, "1")
+    _reset_and_get_operator(monkeypatch, forge)
+    assert subsystem_health._AUTO_FILER is None
+    record_failure("memory_selector", "RuntimeError: boom", tier=1)
+    assert forge.searched == []
+    assert forge.created == []
+
+
+def test_autofile_off_blocks_set_auto_filer(monkeypatch):
+    monkeypatch.setenv(AUTOFILE_OFF_ENV, "1")
+    subsystem_health.set_auto_filer(RecordingFiler())
+    assert subsystem_health._AUTO_FILER is None
+
+
+def test_unconfigured_forge_wiring_is_noop_by_construction(monkeypatch):
+    # No FORGE_URL/TOKEN: create_forge_client() returns NullAdapter; the
+    # filer is wired but can never touch the network by construction.
+    from rune.forge_client import NullAdapter
+
+    _reset_and_get_operator(monkeypatch, NullAdapter())
+    assert isinstance(subsystem_health._AUTO_FILER, FailureAutoFiler)
+    record_failure("memory_selector", "RuntimeError: boom", tier=1)
+    subsystem_health._AUTO_FILER.drain(timeout=5)  # no network, no raise
+
+
+def test_event_journal_still_receives_canonical_events_after_wiring(tmp_path, monkeypatch):
+    monkeypatch.setattr("rune.event_bus._store_path", lambda: tmp_path / "events.jsonl")
+    forge = FakeForge()
+    filer = FailureAutoFiler(forge, JsonlSpool(tmp_path / "events.jsonl"), wait=True)
+    subsystem_health.set_auto_filer(filer)
+    registry = SubsystemHealthRegistry()
+    record_failure("memory_selector", RuntimeError("boom"), tier=1, registry=registry)
+    filer.drain(timeout=5)
+    # The canonical tier event still lands in the journal.
+    from rune.orchestration.event_store import DurableEventStore
+
+    store = DurableEventStore(tmp_path / "events.jsonl")
+    assert any(r["event_type"] == "capability.failed" for r in store.replay())

tests/test_forge_client.py

@@ -26,6 +26,7 @@ def test_null_adapter_is_noop():
     assert r["created"] is False
     assert adapter.get_issue_state(1) == {"state": "unknown"}
     assert adapter.list_open_issues() == []
+    assert adapter.search_issues("query") == []
 
 
 def test_create_forge_client_unconfigured_returns_null():
@@ -65,3 +66,93 @@ def test_gitea_adapter_create_issue_sends_post(monkeypatch):
     assert "issues" in captured["url"]
     assert captured["json_data"]["title"] == "Test"
     assert captured["headers"]["Authorization"] == "token tok"
+
+
+def test_gitea_adapter_search_issues_uses_repo_scoped_list_endpoint():
+    # #122 gate discharge: Gitea/Forgejo has NO repo-scoped
+    # /issues/search path (verified against live swagger) — the old URL
+    # 404'd on every real forge, so every search raised, everything
+    # spooled, and nothing ever filed. The real request must be the
+    # repo-scoped LIST endpoint, which supports state/type/q server-side
+    # and is paginated.
+    adapter = GiteaForgejoAdapter("https://forge.example.com", "tok", owner="o", repo="r")
+    captured = {}
+
+    class FakeResponse:
+        def json(self):
+            return [{"number": 3, "title": "[auto-failure:cron:Connection]"}]
+
+        def raise_for_status(self):
+            pass
+
+    def fake_get(url, **kwargs):
+        captured["url"] = url
+        return FakeResponse()
+
+    with patch("rune.forge_client.httpx.get", side_effect=fake_get):
+        results = adapter.search_issues("[auto-failure:cron:Connection]")
+
+    assert captured["url"].startswith("https://forge.example.com/api/v1/repos/o/r/issues?")
+    assert "/issues/search" not in captured["url"]
+    assert "state=open" in captured["url"]
+    assert "type=issues" in captured["url"]
+    assert "q=%5Bauto-failure%3Acron%3AConnection%5D" in captured["url"]
+    assert results == [{"number": 3, "title": "[auto-failure:cron:Connection]"}]
+
+
+def test_gitea_adapter_search_issues_paginates_until_short_page():
+    # The list endpoint is paginated: a 50-item page must be followed by
+    # a page=2 request (a single 50-cap window would silently miss older
+    # failure issues — the same defect list_open_issues has).
+    adapter = GiteaForgejoAdapter("https://forge.example.com", "tok", owner="o", repo="r")
+    pages = {
+        1: [{"number": n, "title": f"issue {n}"} for n in range(50)],
+        2: [{"number": 100, "title": "[auto-failure:cron:Connection] cron"}],
+    }
+    urls = []
+
+    class FakeResponse:
+        def __init__(self, payload):
+            self._payload = payload
+
+        def json(self):
+            return self._payload
+
+        def raise_for_status(self):
+            pass
+
+    def fake_get(url, **kwargs):
+        urls.append(url)
+        return FakeResponse(pages[1 if "page=1" in url else 2])
+
+    with patch("rune.forge_client.httpx.get", side_effect=fake_get):
+        results = adapter.search_issues("[auto-failure:cron:Connection]")
+
+    assert len(urls) == 2
+    assert any("page=2" in u for u in urls)
+    assert results == [{"number": 100, "title": "[auto-failure:cron:Connection] cron"}]
+
+
+def test_gitea_adapter_search_filters_to_marker_hits_client_side():
+    # Server q= matches the title (and fuzzy keywords); the client-side
+    # contains-check keeps title/body marker hits and drops fuzzy noise
+    # so a quoted marker in an unrelated issue's body never dedups falsely.
+    adapter = GiteaForgejoAdapter("https://forge.example.com", "tok", owner="o", repo="r")
+    marker = "[auto-failure:cron:Connection]"
+    payload = [
+        {"number": 1, "title": f"{marker} cron", "body": "x"},
+        {"number": 2, "title": "plain title", "body": f"quoted {marker} in body"},
+        {"number": 3, "title": "mention of connection only", "body": "unrelated"},
+    ]
+
+    class FakeResponse:
+        def json(self):
+            return payload
+
+        def raise_for_status(self):
+            pass
+
+    with patch("rune.forge_client.httpx.get", return_value=FakeResponse()):
+        results = adapter.search_issues(marker)
+
+    assert [r["number"] for r in results] == [1, 2]