feat(failure_auto_filer): close PARTIAL gaps — dichotomy wave

scripts/gen_failure_auto_filer_truth.py

@@ -0,0 +1,377 @@
+#!/usr/bin/env python3
+"""Freeze oracle truth for failure_auto_filer (gap-closing wave).
+
+Two artifacts, both consumed VERBATIM by the Go battery:
+  1. src/go/failure-filer/frozen_failure_auto_filer_truth.json — pure
+     module-level functions via the shared freeze_truth harness (every
+     public function and JSON-expressible branch).
+  2. src/go/failure-filer/frozen_filer_scenarios.json — deterministic
+     FailureAutoFiler behavior traces (wait=True, fake client + fake
+     journal, pinned ``_now``); wall-clock fields normalized so both
+     sides agree. Exception-only branches (error_class_of on an
+     exception instance) are covered by Go unit tests — exceptions are
+     not JSON-expressible, so they cannot ride the frozen truth.
+"""
+from __future__ import annotations
+
+import json
+import re
+import sys
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "src"))
+sys.path.insert(0, str(REPO / "scripts"))
+
+from freeze_truth import freeze, write  # noqa: E402
+
+from rune import failure_auto_filer as faf  # noqa: E402
+
+TRUTH_PATH = REPO / "src/go/failure-filer/frozen_failure_auto_filer_truth.json"
+SCENARIOS_PATH = REPO / "src/go/failure-filer/frozen_filer_scenarios.json"
+
+LONG = "A" * 70  # > _ERROR_CLASS_MAX token
+CRED_URL = "postgres://admin:hunter2@db.example.com:5432/rune?password=sekret&token=abc123&host=x"
+
+
+# ── deterministic fakes (mirrored 1:1 by the Go battery) ────────────────────
+
+_TS = re.compile(r"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ")
+
+
+def _norm(text: str) -> str:
+    return _TS.sub("<TS>", text)
+
+
+class FakeClient:
+    def __init__(self, issues=None, fail=False):
+        self.calls = []
+        self.issues = [dict(i) for i in (issues or [])]
+        self.fail = fail
+
+    def search_issues(self, marker):
+        self.calls.append(["search_issues", marker])
+        if self.fail:
+            raise Exception("forge down")
+        return [dict(i) for i in self.issues]
+
+    def add_comment(self, number, body):
+        self.calls.append(["add_comment", number, _norm(body)])
+        if self.fail:
+            raise Exception("forge down")
+
+    def create_issue(self, title, body):
+        self.calls.append(["create_issue", title, _norm(body)])
+        if self.fail:
+            raise Exception("forge down")
+        self.calls[-1] = ["create_issue", title, _norm(body)]
+        return {"number": len(self.issues) + 1, "created": True}
+
+
+class FakeJournal:
+    def __init__(self, fail_unconsumed=False, fail_append=False):
+        self.records = []
+        self.consumed = []
+        self.fail_unconsumed = fail_unconsumed
+        self.fail_append = fail_append
+
+    def append(self, record):
+        if self.fail_append:
+            raise Exception("journal append failed")
+        record = dict(record)
+        record["spool_id"] = f"spool-det-{len(self.records)}"
+        record["timestamp"] = "<TS>"
+        self.records.append(record)
+
+    def unconsumed(self):
+        if self.fail_unconsumed:
+            raise Exception("journal read failed")
+        return [dict(r) for r in self.records if r["spool_id"] not in self.consumed]
+
+    def mark_consumed(self, spool_id):
+        self.consumed.append(spool_id)
+
+
+# ── scenarios: each returns a deterministic trace dict ──────────────────────
+
+def sc_submit_t2_gate():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f.submit("sub", "ValueError: boom", 2, 100.0)
+    f.drain()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": []}
+
+
+def sc_submit_coalesce_create():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    for i in range(3):
+        f.submit("api", "TimeoutError: t", 1, 100.0 + i)
+    f.drain()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": []}
+
+
+def sc_file_first_only():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    returns = [
+        f.file("s", "E: a", 1, 100.0),
+        f.file("s", "E: b", 1, 101.0),
+        f.file("s", "E: c", 1, 102.0),
+    ]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_file_batch_full():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=3, wait=True)
+    returns = [
+        f.file("s", "E: a", 1, 100.0),
+        f.file("s", "E: b", 1, 101.0),
+        f.file("s", "E: c", 1, 102.0),
+    ]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_file_window_expiry():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    returns = [f.file("s", "E: a", 1, 100.0), f.file("s", "E: b", 1, 100.0 + 600)]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_file_comment_existing():
+    client = FakeClient(issues=[{"number": 7, "title": "[auto-failure:s:E] s"}])
+    journal = FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    returns = [f.file("s", "E: latest text", 1, 100.0)]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_file_t2():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    returns = [f.file("s", "E: cron noise", 2, 100.0)]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_file_client_failure_spools():
+    client, journal = FakeClient(fail=True), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    returns = [f.file("api", "TimeoutError: down", 1, 100.0)]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_file_single_flight():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f._inflight.add(("s", "E"))
+    returns = [f.file("s", "E: x", 1, 100.0)]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_file_spool_journal_failure_never_raises():
+    client, journal = FakeClient(fail=True), FakeJournal(fail_append=True)
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    returns = [f.file("api", "TimeoutError: down", 1, 100.0)]
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": returns}
+
+
+def sc_call_dunder_is_submit():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f("api", "ValueError: boom", 1, 100.0)
+    f.drain()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": []}
+
+
+def sc_reconcile_groups_and_marks_consumed():
+    client, journal = FakeClient(fail=True), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f.file("api", "TimeoutError: t1", 1, 100.0)
+    f.file("api", "TimeoutError: t2", 1, 101.0)
+    client.fail = False
+    report = f.reconcile()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "report": report}
+
+
+def sc_reconcile_comment_existing():
+    client, journal = FakeClient(fail=True), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f.file("api", "TimeoutError: t1", 1, 100.0)
+    client.fail = False
+    client.issues = [{"number": 9, "title": "[auto-failure:api:TimeoutError] api"}]
+    report = f.reconcile()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "report": report}
+
+
+def sc_reconcile_t2_skipped():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    journal.append({"kind": faf.SPOOL_KIND, "spool_id": "spool-cron", "timestamp": "<TS>",
+                    "signature": ["cron", "Noise"], "tier": 2, "count": 1,
+                    "error": "noise", "client_error": "-"})
+    report = f.reconcile()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "report": report}
+
+
+def sc_reconcile_malformed_line():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    journal.append({"kind": faf.SPOOL_KIND, "spool_id": "spool-bad", "timestamp": "<TS>",
+                    "signature": ["only-one"], "tier": 1, "count": 1,
+                    "error": "x", "client_error": "-"})
+    report = f.reconcile()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "report": report}
+
+
+def sc_reconcile_failing_client_keeps_spooled():
+    client, journal = FakeClient(fail=True), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f.file("api", "TimeoutError: t1", 1, 100.0)
+    report = f.reconcile()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "report": report}
+
+
+def sc_reconcile_journal_read_failure():
+    client, journal = FakeClient(), FakeJournal(fail_unconsumed=True)
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    report = f.reconcile()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "report": report}
+
+
+def sc_reconcile_null_adapter_result():
+    """create_issue returning created=False must count as failed, not filed."""
+    client, journal = FakeClient(fail=True), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f.file("api", "TimeoutError: t1", 1, 100.0)
+    # swap in a NullAdapter-shaped client: create_issue reports created=False
+    client2 = FakeClient()
+
+    def null_create(title, body):
+        client2.calls.append(["create_issue", title, _norm(body)])
+        return {"created": False}
+
+    client2.create_issue = null_create  # type: ignore[method-assign]
+    f._client = client2  # type: ignore[attr-defined]
+    report = f.reconcile()
+    return {"calls": client2.calls, "records": journal.records, "consumed": journal.consumed,
+            "report": report}
+
+
+def sc_queue_full_spools_overflow():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f._queue = __import__("queue").Queue(maxsize=2)  # type: ignore[assignment]
+    for i in range(4):
+        f.submit("api", "TimeoutError: t", 1, 100.0)
+    f.drain()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": []}
+
+
+def sc_drain_twice_idempotent():
+    client, journal = FakeClient(), FakeJournal()
+    f = faf.FailureAutoFiler(client, journal, batch_size=10, wait=True)
+    f.submit("api", "TimeoutError: t", 1, 100.0)
+    f.drain()
+    f.drain()
+    return {"calls": client.calls, "records": journal.records, "consumed": journal.consumed,
+            "returns": []}
+
+
+SCENARIOS = [
+    sc_submit_t2_gate, sc_submit_coalesce_create, sc_file_first_only,
+    sc_file_batch_full, sc_file_window_expiry, sc_file_comment_existing,
+    sc_file_t2, sc_file_client_failure_spools, sc_file_single_flight,
+    sc_file_spool_journal_failure_never_raises, sc_call_dunder_is_submit,
+    sc_reconcile_groups_and_marks_consumed, sc_reconcile_comment_existing,
+    sc_reconcile_t2_skipped, sc_reconcile_malformed_line,
+    sc_reconcile_failing_client_keeps_spooled, sc_reconcile_journal_read_failure,
+    sc_reconcile_null_adapter_result, sc_queue_full_spools_overflow,
+    sc_drain_twice_idempotent,
+]
+
+
+# ── pure-function case matrix (every JSON-expressible branch) ───────────────
+
+def pure_cases():
+    cases = []
+    add = cases.append
+    # error_class_of — string branches
+    add(("error_class_of", ("RuntimeError: boom",), {}))
+    add(("error_class_of", ("plain token error",), {}))
+    add(("error_class_of", ("",), {}))                      # empty → Unknown
+    add(("error_class_of", ("   ",), {}))                   # whitespace-only → Unknown
+    add(("error_class_of", ("ValueError:",), {}))           # trailing colon stripped
+    add(("error_class_of", ("ValueError::",), {}))          # rstrip removes ALL colons
+    add(("error_class_of", (LONG + ": tail",), {}))         # capped at 64
+    add(("error_class_of", (LONG,), {}))                    # capped at 64, no tail
+    add(("error_class_of", ("::weird::",), {}))             # colons both sides
+    add(("error_class_of", ("BrokerError: postgres://u:p@h",), {}))
+    # signature_of
+    add(("signature_of", ("api", "TimeoutError: down"), {}))
+    add(("signature_of", ("api", ""), {}))
+    # tier_gated — the gate boundaries
+    for t in (0, 1, 2, 3):
+        add(("tier_gated", (t,), {}))
+    # marker_for
+    add(("marker_for", ("api", "TimeoutError"), {}))
+    add(("marker_for", ("", ""), {}))
+    # excerpt — default limit and custom
+    add(("excerpt", (CRED_URL,), {}))
+    add(("excerpt", (CRED_URL,), {"limit": 40}))
+    add(("excerpt", ("short",), {"limit": 200}))
+    add(("excerpt", ("x" * 300,), {}))                       # truncation
+    add(("excerpt", ("x" * 300,), {"limit": 5}))
+    # redact — each pattern family + clean + combined
+    add(("redact", ("RuntimeError: boom",), {}))             # clean: no-op
+    add(("redact", ("GET https://h/x?token=tok123&ok=1 done",), {}))
+    add(("redact", ("GET https://h/x?API_KEY=ABC&sig=s1&access_token=at9 done",), {}))
+    add(("redact", (CRED_URL,), {}))
+    add(("redact", ("redis://:s3cret@cache:6379",), {}))     # empty user
+    add(("redact", ("https://user:pass@host/path",), {}))
+    add(("redact", ("Authorization: Bearer abc.def.ghi expired",), {}))
+    add(("redact", ("bearer tok-here trailing",), {}))
+    add(("redact", ("password=hunter2 and secret: 'quoted'",), {}))
+    add(("redact", ("api_key='abcd1234' auth: short",), {}))
+    add(("redact", ("password=short",), {}))                 # <4 chars: NOT redacted
+    add(("redact", ("auth=x1 auth=abcd1234",), {}))
+    add(("redact", ("connect postgres://u:p@h then ?token=t&x=2",), {}))  # combined
+    # _error_text — string branch (exception branch covered by Go unit tests)
+    add(("_error_text", ("plain string",), {}))
+    return cases
+
+
+def main() -> int:
+    truth = freeze("failure_auto_filer", pure_cases(), oracle_module="rune.failure_auto_filer")
+    write(truth, TRUTH_PATH)
+    print(f"frozen truth: {len(truth)} cases -> {TRUTH_PATH.relative_to(REPO)}")
+
+    scenarios = [{"name": fn.__name__.removeprefix("sc_"), "trace": fn()} for fn in SCENARIOS]
+    SCENARIOS_PATH.write_text(
+        json.dumps(scenarios, indent=2, ensure_ascii=False, sort_keys=False) + "\n"
+    )
+    print(f"frozen scenarios: {len(scenarios)} -> {SCENARIOS_PATH.relative_to(REPO)}")
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/walk_contract_failure_auto_filer.py

@@ -0,0 +1,39 @@
+#!/usr/bin/env python3
+"""Open + walk the #2479 gap-closing construction contract for
+failure_auto_filer to verifying. Full lifecycle in ONE process."""
+import json
+import sys
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from rune.construction_interface import ConstructionInterface
+
+iface = ConstructionInterface(builder="factory-gap-faf")
+
+r = iface.submit_need(
+    "Close the failure_auto_filer dichotomy gap (#2479 tranche): the Go "
+    "port is PARTIAL (11 fns) — port the missing worker/pipeline half "
+    "(_error_text, JsonlSpool append/unconsumed/mark_consumed, "
+    "FailureAutoFiler __call__/submit/drain/file/reconcile/_write/"
+    "_write_once/_spool/_run_worker/_process_batch/_flush_group/"
+    "_drain_once) name-preserving with logic 1:1, forge IO + journal IO "
+    "+ clock behind injectable seams, no goroutine spawns in test paths "
+    "(wait=True synchronous mode mirrors the oracle). Freeze oracle "
+    "truth via the shared scripts/freeze_truth.py harness (every public "
+    "function and branch) and embed it VERBATIM in the Go battery; live "
+    "parity 100% on the new fns; all existing tests stay green.",
+    origin="builder",
+)
+print("submit_need:", json.dumps(r)[:300])
+if not r.get("success"):
+    print("NEED FAILED — aborting")
+    sys.exit(1)
+cid = r["ids"]["contract_id"]
+print("contract_id:", cid)
+for phase in ("approved", "building", "verifying"):
+    t = iface.transition(cid, phase)
+    print(f"transition->{phase}:", json.dumps(t)[:150])
+    if not t.get("success"):
+        sys.exit(1)
+print("CONTRACT", cid, "verifying")

src/go/failure-filer/battery_test.go

@@ -0,0 +1,88 @@
+package main
+
+// battery_test.go — frozen-oracle battery for the pure pipeline fns
+// (frozen_failure_auto_filer_truth.json, generated by
+// scripts/gen_failure_auto_filer_truth.py via the shared
+// scripts/freeze_truth.py harness; VERIFIED byte-identical against the
+// live Python oracle rune.failure_auto_filer with
+//   PYTHONPATH=src python3 scripts/freeze_truth.py --verify \
+//     src/go/failure-filer/frozen_failure_auto_filer_truth.json \
+//     --module rune.failure_auto_filer
+// ). Replayed here VERBATIM — no hand-transcribed expectations.
+
+import (
+	"encoding/json"
+	"os"
+	"testing"
+)
+
+func TestFrozenTruthBattery(t *testing.T) {
+	raw, err := os.ReadFile("frozen_failure_auto_filer_truth.json")
+	if err != nil {
+		t.Fatalf("frozen truth missing: %v", err)
+	}
+	var cases []struct {
+		Fn       string         `json:"fn"`
+		Args     []any          `json:"args"`
+		Kwargs   map[string]any `json:"kwargs"`
+		Expected any            `json:"expected"`
+	}
+	if err := json.Unmarshal(raw, &cases); err != nil {
+		t.Fatalf("unparseable: %v", err)
+	}
+	for i, c := range cases {
+		got := replayPure(t, c.Fn, c.Args, c.Kwargs)
+		want, _ := json.Marshal(c.Expected)
+		gotJSON, _ := json.Marshal(got)
+		if string(gotJSON) != string(want) {
+			t.Errorf("case %d %s%v: got %s want %s", i, c.Fn, c.Args, gotJSON, want)
+		}
+	}
+}
+
+// replayPure dispatches a frozen case to the Go implementation. Name map
+// is the one documented in filer.go.
+func replayPure(t *testing.T, fn string, args []any, kwargs map[string]any) any {
+	t.Helper()
+	arg := func(i int) string { return args[i].(string) }
+	switch fn {
+	case "error_class_of":
+		return errorClassOf(arg(0))
+	case "signature_of":
+		sub, cls := signatureOf(arg(0), arg(1))
+		return []string{sub, cls}
+	case "tier_gated":
+		return tierGated(int(args[0].(float64)))
+	case "marker_for":
+		return markerFor(arg(0), arg(1))
+	case "redact":
+		return redact(arg(0))
+	case "excerpt":
+		limit := 200 // oracle _EXCERPT_LEN default
+		if v, ok := kwargs["limit"]; ok {
+			limit = int(v.(float64))
+		}
+		return excerpt(arg(0), limit)
+	case "_error_text":
+		return errorText(arg(0))
+	default:
+		t.Fatalf("unknown frozen fn: %s", fn)
+		return nil
+	}
+}
+
+// errorText on a Go error value (the oracle's BaseException branch —
+// exceptions are not JSON-expressible so they cannot ride the frozen
+// truth; covered here instead).
+type batteryErr struct{}
+
+func (batteryErr) Error() string { return "boom" }
+
+func TestErrorTextExceptionBranch(t *testing.T) {
+	if got, want := errorText(batteryErr{}), "batteryErr: boom"; got != want {
+		t.Errorf("errorText(error): got %q want %q", got, want)
+	}
+	if got, want := errorText("plain"), "plain"; got != want {
+		t.Errorf("errorText(string): got %q want %q", got, want)
+	}
+}

src/go/failure-filer/filer.go

@@ -0,0 +1,614 @@
+package main
+
+// filer.go — the FailureAutoFiler pipeline (oracle failure_auto_filer.py,
+// FailureAutoFiler class, 1:1).
+//
+// Failure path only Submit()s (enqueue, no network); a single worker
+// drains the queue (worker goroutine spawned ONLY when wait=false —
+// production mode; tests use wait=true + Drain, so no goroutine spawns
+// in test paths). Bursts coalesce: everything already queued for a
+// signature rides ONE client write. File() is the synchronous
+// single-occurrence pipeline step; the tier gate lives in tierGated and
+// nowhere else. The filer NEVER raises into the failure path.
+//
+// Injectable seams (oracle passes a ForgeClient + SpoolJournal and uses
+// the wall clock; here all three are interfaces / vars so tests pin them):
+//   - ForgeClient interface (search_issues/add_comment/create_issue)
+//   - SpoolJournal interface (append/unconsumed/mark_consumed)
+//   - monotonic / nowUTC / timeNS vars (clock)
+//
+// Name map (oracle → here, existing camelCase style of main.go):
+//   _error_text→errorText, FailureAutoFiler.__call__→Call,
+//   submit→Submit, drain→Drain, file→File, reconcile→Reconcile,
+//   _write→write, _write_once→writeOnce, _spool→spool,
+//   _run_worker→runWorker, _process_batch→processBatch,
+//   _flush_group→flushGroup, _drain_once→drainOnce,
+//   _SigState→sigState, _inflight→inflight, _batch_size→batchSize.
+
+import (
+	"errors"
+	"fmt"
+	"log"
+	"math"
+	"strings"
+	"sync"
+	"time"
+)
+
+const (
+	batchIntervalSeconds = 600  // one comment per 10 occurrences OR 10 minutes
+	excerptLen           = 200  // oracle _EXCERPT_LEN
+	queueMaxsize         = 1000 // bounded: an outage must not balloon failure-path memory
+)
+
+// ForgeClient is the forge-write seam (the oracle's ForgeClient contract
+// as used by writeOnce). Mirrors ForgeClient.search_issues /
+// add_comment / create_issue.
+type ForgeClient interface {
+	SearchIssues(marker string) ([]map[string]any, error)
+	AddComment(number int, body string) error
+	CreateIssue(title, body string) (map[string]any, error)
+}
+
+// SpoolJournal is the durable-journal seam (oracle SpoolJournal protocol,
+// satisfied by JsonlSpool).
+type SpoolJournal interface {
+	Append(record map[string]any) error
+	Unconsumed() ([]map[string]any, error)
+	MarkConsumed(spoolID string)
+}
+
+// errorText is oracle _error_text: strings pass through; an error (the
+// Go analogue of a BaseException) renders "Type: message".
+func errorText(err any) string {
+	switch t := err.(type) {
+	case string:
+		return t
+	case error:
+		return fmt.Sprintf("%s: %v", errClassName(t), t)
+	default:
+		return fmt.Sprintf("%v", t)
+	}
+}
+
+func errClassName(err error) string {
+	name := fmt.Sprintf("%T", err)
+	if i := strings.LastIndex(name, "."); i >= 0 {
+		name = name[i+1:]
+	}
+	return strings.TrimPrefix(name, "*")
+}
+
+// sigKey is the dedup signature (subsystem, error_class) — the oracle's
+// (str, str) tuple key.
+type sigKey struct {
+	subsystem string
+	class     string
+}
+
+func signatureKey(subsystem, errMsg string) sigKey {
+	_, class := signatureOf(subsystem, errMsg)
+	return sigKey{subsystem: subsystem, class: class}
+}
+
+// sigState is oracle _SigState: in-memory, process-lifetime coalescing
+// state per signature.
+type sigState struct {
+	count       int     // total occurrences seen this process
+	pending     int     // occurrences since the last client write
+	windowStart float64 // monotonic time of the last client write
+}
+
+// filerItem is one queued occurrence (oracle queue tuple
+// (subsystem, error_text, tier, now)).
+type filerItem struct {
+	subsystem string
+	text      string
+	tier      int
+	now       float64
+}
+
+// filerQueue mirrors queue.Queue(maxsize): put_nowait drops on full,
+// get blocks, unfinished_tasks/task_done gate Drain.
+type filerQueue struct {
+	mu         sync.Mutex
+	notEmpty   sync.Cond
+	items      []filerItem
+	maxsize    int
+	unfinished int
+	sentinel   bool
+}
+
+func newFilerQueue(maxsize int) *filerQueue {
+	q := &filerQueue{maxsize: maxsize}
+	q.notEmpty.L = &q.mu
+	return q
+}
+
+func (q *filerQueue) putNowait(it filerItem) bool {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if len(q.items) >= q.maxsize {
+		return false // oracle queue.Full
+	}
+	q.items = append(q.items, it)
+	q.unfinished++
+	q.notEmpty.Signal()
+	return true
+}
+
+// get blocks until an item or the shutdown sentinel; ok=false → sentinel.
+func (q *filerQueue) get() (filerItem, bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	for len(q.items) == 0 && !q.sentinel {
+		q.notEmpty.Wait()
+	}
+	if len(q.items) == 0 {
+		return filerItem{}, false
+	}
+	it := q.items[0]
+	q.items = q.items[1:]
+	return it, true
+}
+
+func (q *filerQueue) getNowait() (filerItem, bool) {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	if len(q.items) == 0 {
+		return filerItem{}, false
+	}
+	it := q.items[0]
+	q.items = q.items[1:]
+	return it, true
+}
+
+func (q *filerQueue) putSentinel() {
+	q.mu.Lock()
+	q.sentinel = true
+	q.mu.Unlock()
+	q.notEmpty.Broadcast()
+}
+
+func (q *filerQueue) taskDone() {
+	q.mu.Lock()
+	if q.unfinished > 0 {
+		q.unfinished--
+	}
+	q.mu.Unlock()
+}
+
+func (q *filerQueue) unfinishedTasks() int {
+	q.mu.Lock()
+	defer q.mu.Unlock()
+	return q.unfinished
+}
+
+// Injectable clock seams — production wires the wall clock; tests pin
+// them. NaN `now` sentinels mean "ask the clock" (the oracle's _now=None).
+var (
+	monotonic = func() float64 { return float64(time.Now().UnixNano()) / 1e9 }
+	nowUTC    = func() string { return time.Now().UTC().Format("2006-01-02T15:04:05Z") }
+	timeNS    = func() int64 { return time.Now().UnixNano() }
+)
+
+func clockNow(now float64) float64 {
+	if math.IsNaN(now) {
+		return monotonic()
+	}
+	return now
+}
+
+// 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 in tierGated and nowhere else.
+type FailureAutoFiler struct {
+	client    ForgeClient
+	journal   SpoolJournal
+	queue     *filerQueue
+	mu        sync.Mutex
+	state     map[sigKey]*sigState
+	batchSize int
+	inflight  map[sigKey]bool // single-flight per signature
+	waitMode  bool            // oracle: worker thread is None
+}
+
+func NewFailureAutoFiler(client ForgeClient, journal SpoolJournal, batchSize int, wait bool) *FailureAutoFiler {
+	if batchSize < 1 {
+		batchSize = 1
+	}
+	f := &FailureAutoFiler{
+		client:    client,
+		journal:   journal,
+		queue:     newFilerQueue(queueMaxsize),
+		state:     map[sigKey]*sigState{},
+		batchSize: batchSize,
+		inflight:  map[sigKey]bool{},
+		waitMode:  wait,
+	}
+	if !wait {
+		go f.runWorker()
+	}
+	return f
+}
+
+// ------------------------------------------------------ failure-path API
+
+// Call is oracle __call__ (the callable-filer contract): exactly Submit.
+func (f *FailureAutoFiler) Call(subsystem, errMsg string, tier int, now float64) {
+	f.Submit(subsystem, errMsg, tier, now)
+}
+
+// Submit is the failure-path entry: enqueue and return immediately.
+// Never blocks on the network, never raises (swallowed + logged). With
+// wait=true nothing is written until Drain — the whole backlog
+// coalesces into one write per signature.
+func (f *FailureAutoFiler) Submit(subsystem, errMsg string, tier int, now float64) {
+	defer func() {
+		if r := recover(); r != nil { // never raise into the failure path
+			log.Printf("[failure_auto_filer] submit failed for subsystem=%s: %v", subsystem, r)
+		}
+	}()
+	if tierGated(tier) {
+		return // THE tier gate — T2 journal-only, zero queue traffic
+	}
+	item := filerItem{subsystem: subsystem, text: errMsg, tier: tier, now: clockNow(now)}
+	if !f.queue.putNowait(item) {
+		// 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.Printf("[failure_auto_filer] auto-filer queue full; spooling dropped occurrence for subsystem=%s", subsystem)
+		f.spool(subsystem, errMsg, tier, errors.New("auto-filer queue full"), 1)
+	}
+}
+
+// Drain blocks until the queue is empty (tests + operator shutdown).
+// timeout < 0 waits forever (the oracle's timeout=None).
+func (f *FailureAutoFiler) Drain(timeout float64) {
+	if f.waitMode {
+		f.drainOnce()
+		return
+	}
+	hasDeadline := timeout >= 0
+	deadline := monotonic() + timeout
+	for f.queue.unfinishedTasks() > 0 {
+		if hasDeadline && monotonic() > deadline {
+			return
+		}
+		time.Sleep(5 * time.Millisecond)
+	}
+}
+
+// -------------------------------------------------- synchronous pipeline
+
+// File is the 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.
+func (f *FailureAutoFiler) File(subsystem, errMsg string, tier int, now float64) (ret bool) {
+	defer func() {
+		if r := recover(); r != nil { // degrade, never propagate
+			f.spool(subsystem, errMsg, tier, fmt.Errorf("%v", r), 1)
+			ret = false
+		}
+	}()
+	if tierGated(tier) {
+		return false // THE tier gate — one place, T2 journal-only
+	}
+	now = clockNow(now)
+	sig := signatureKey(subsystem, errMsg)
+	text := errMsg
+	f.mu.Lock()
+	st := f.state[sig]
+	if st == nil {
+		st = &sigState{}
+		f.state[sig] = st
+	}
+	st.count++
+	st.pending++
+	due := st.count == 1 ||
+		st.pending >= f.batchSize ||
+		(st.pending > 0 && now-st.windowStart >= batchIntervalSeconds)
+	countAtFlush := st.count
+	if due {
+		st.pending = 0
+		st.windowStart = now
+	}
+	f.mu.Unlock()
+	if !due {
+		return false // coalesced: no client call for this occurrence
+	}
+	ok, err := f.write(sig, text, tier, countAtFlush, 0)
+	if err != nil {
+		f.spool(subsystem, errMsg, tier, err, 1)
+		return false
+	}
+	return ok
+}
+
+// ------------------------------------------------------------- reconcile
+
+// Reconcile replays 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. Dedup
+// re-applies via SearchIssues (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.
+func (f *FailureAutoFiler) Reconcile() map[string]int {
+	report := map[string]int{"replayed": 0, "filed": 0, "failed": 0, "skipped": 0}
+	defer func() {
+		if r := recover(); r != nil {
+			report["failed"]++
+			log.Printf("[failure_auto_filer] reconcile failed: %v", r)
+		}
+	}()
+	pending, err := f.journal.Unconsumed()
+	if err != nil {
+		log.Printf("[failure_auto_filer] reconcile could not read the spool journal: %v", err)
+		report["failed"]++
+		return report
+	}
+	groups := map[sigKey][]map[string]any{}
+	order := []sigKey{}
+	for _, record := range pending {
+		report["replayed"]++
+		sigList, _ := record["signature"].([]any)
+		if len(sigList) != 2 {
+			report["failed"]++ // malformed line: leave it, never crash
+			continue
+		}
+		tier := int(asNumber(record["tier"], 1))
+		if tierGated(tier) {
+			// THE tier gate on replay — the same predicate as the live
+			// path (one place). A T2 line resolves journal-only:
+			// consumed, never written.
+			f.journal.MarkConsumed(asStr(record["spool_id"]))
+			report["skipped"]++
+			continue
+		}
+		sig := sigKey{subsystem: asStr(sigList[0]), class: asStr(sigList[1])}
+		if _, ok := groups[sig]; !ok {
+			order = append(order, sig)
+		}
+		groups[sig] = append(groups[sig], record)
+	}
+	for _, sig := range order {
+		group := groups[sig]
+		last := group[len(group)-1]
+		total := 0
+		for _, r := range group {
+			c := int(asNumber(r["count"], 1))
+			if c < 1 {
+				c = 1
+			}
+			total += c
+		}
+		filed, err := f.write(
+			sig,
+			asStr(last["error"]),
+			int(asNumber(last["tier"], 1)),
+			total,
+			len(group),
+		)
+		if err == nil && filed {
+			for _, record := range group {
+				f.journal.MarkConsumed(asStr(record["spool_id"]))
+			}
+			report["filed"] += len(group)
+		} else {
+			report["failed"] += len(group) // e.g. NullAdapter: nothing was filed
+		}
+	}
+	return report
+}
+
+// ------------------------------------------------------------ internals
+
+// write is oracle _write: one client write for a signature. Single-flight
+// per signature (#122 gate): the reconcile path 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, nil) —
+// its spool line stays spooled and dedup re-applies on the next replay.
+func (f *FailureAutoFiler) write(sig sigKey, text string, tier, count, offline int) (bool, error) {
+	f.mu.Lock()
+	if f.inflight[sig] {
+		f.mu.Unlock()
+		return false, nil // single-flight: another goroutine owns this write
+	}
+	f.inflight[sig] = true
+	f.mu.Unlock()
+	defer func() {
+		f.mu.Lock()
+		delete(f.inflight, sig)
+		f.mu.Unlock()
+	}()
+	return f.writeOnce(sig, text, tier, count, offline)
+}
+
+// writeOnce is oracle _write_once. Raises on client failure — callers
+// spool. (Cross-PROCESS dedup remains best-effort TOCTOU between search
+// and create; documented oracle limitation.)
+func (f *FailureAutoFiler) writeOnce(sig sigKey, text string, tier, count, offline int) (bool, error) {
+	marker := markerFor(sig.subsystem, sig.class)
+	existing, err := f.client.SearchIssues(marker)
+	if err != nil {
+		return false, err
+	}
+	var match map[string]any
+	for _, issue := range existing {
+		if strings.Contains(asStr(issue["title"]), marker) {
+			match = issue
+			break
+		}
+	}
+	ts := nowUTC()
+	if match != nil {
+		var body string
+		if offline > 0 {
+			body = fmt.Sprintf("+%d occurrences while offline (last: %s @ %s)",
+				count, excerpt(text, excerptLen), ts)
+		} else {
+			body = fmt.Sprintf("+%d occurrences (last: %s @ %s)",
+				max(1, count-1), excerpt(text, excerptLen), ts)
+		}
+		number := int(asNumber(match["number"], 0))
+		if err := f.client.AddComment(number, body); err != nil {
+			return false, err
+		}
+		return true, nil
+	}
+	title := fmt.Sprintf("%s %s", marker, sig.subsystem)
+	body := fmt.Sprintf(
+		"Auto-filed failure signature `%s:%s`\n\n- tier=%d\n- occurrences=%d\n- last: %s\n\n(detected by the failure auto-filer, #122)",
+		sig.subsystem, sig.class, tier, count, excerpt(text, excerptLen))
+	result, err := f.client.CreateIssue(title, body)
+	if err != nil {
+		return false, err
+	}
+	if result != nil && result["created"] == false {
+		return false, nil // e.g. NullAdapter: nothing was filed
+	}
+	return true, nil
+}
+
+// spool is oracle _spool: persist the failure for Reconcile. Never raises.
+func (f *FailureAutoFiler) spool(subsystem, errMsg string, tier int, exc error, count int) {
+	defer func() {
+		if r := recover(); r != nil { // even the spool must not raise
+			log.Printf("[failure_auto_filer] failed to spool auto-file record for subsystem=%s: %v", subsystem, r)
+		}
+	}()
+	sub, class := signatureOf(subsystem, errMsg)
+	record := map[string]any{
+		"kind":         SPOOL_KIND,
+		"spool_id":     fmt.Sprintf("spool-%d", timeNS()),
+		"timestamp":    nowUTC(),
+		"signature":    []any{sub, class}, // list dialect (JSON round-trips as []any)
+		"tier":         tier,
+		"count":        count,
+		"error":        excerpt(errMsg, excerptLen),
+		"client_error": excerpt(exc.Error(), excerptLen),
+	}
+	_ = f.journal.Append(record)
+}
+
+// ------------------------------------------------------------- worker
+
+// runWorker is oracle _run_worker: drain the queue forever, coalescing
+// everything already queued into one batch. Never exits except on the
+// shutdown sentinel.
+func (f *FailureAutoFiler) runWorker() {
+	for {
+		item, ok := f.queue.get()
+		if !ok { // shutdown sentinel
+			f.queue.taskDone()
+			return
+		}
+		batch := []filerItem{item}
+		for { // coalesce: everything already queued rides one write
+			it, ok := f.queue.getNowait()
+			if !ok {
+				break
+			}
+			batch = append(batch, it)
+		}
+		f.runBatchSafely(batch)
+	}
+}
+
+func (f *FailureAutoFiler) runBatchSafely(batch []filerItem) {
+	defer func() {
+		if r := recover(); r != nil { // worker must never die
+			log.Printf("[failure_auto_filer] auto-filer worker batch failed: %v", r)
+		}
+		for range batch {
+			f.queue.taskDone()
+		}
+	}()
+	f.processBatch(batch)
+}
+
+// processBatch is oracle _process_batch: group by signature, then
+// flushGroup each; a flush failure spools the whole group.
+func (f *FailureAutoFiler) processBatch(batch []filerItem) {
+	groups := map[sigKey][]filerItem{}
+	order := []sigKey{}
+	for _, it := range batch {
+		if tierGated(it.tier) {
+			continue // THE tier gate (defense in depth; Submit gates too)
+		}
+		sig := signatureKey(it.subsystem, it.text)
+		if _, ok := groups[sig]; !ok {
+			order = append(order, sig)
+		}
+		groups[sig] = append(groups[sig], it)
+	}
+	for _, sig := range order {
+		group := groups[sig]
+		err := f.flushGroup(sig, group)
+		if err != nil {
+			for _, it := range group {
+				f.spool(it.subsystem, it.text, it.tier, err, len(group))
+			}
+		}
+	}
+}
+
+// flushGroup is oracle _flush_group: coalescing state for a whole batch
+// slice, then one write when due. Errors propagate (the oracle raises;
+// processBatch spools).
+func (f *FailureAutoFiler) flushGroup(sig sigKey, group []filerItem) error {
+	last := group[len(group)-1]
+	now := last.now
+	text := last.text // last error of the batch
+	tier := last.tier
+	f.mu.Lock()
+	st := f.state[sig]
+	if st == nil {
+		st = &sigState{}
+		f.state[sig] = st
+	}
+	st.count += len(group)
+	st.pending += len(group)
+	due := st.count == len(group) ||
+		st.pending >= f.batchSize ||
+		(st.pending > 0 && now-st.windowStart >= batchIntervalSeconds)
+	countAtFlush := st.count
+	if due {
+		st.pending = 0
+		st.windowStart = now
+	}
+	f.mu.Unlock()
+	if due {
+		if _, err := f.write(sig, text, tier, countAtFlush, 0); err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+// drainOnce is oracle _drain_once: inline drain for synchronous
+// (wait=true) mode.
+func (f *FailureAutoFiler) drainOnce() {
+	batch := []filerItem{}
+	for {
+		it, ok := f.queue.getNowait()
+		if !ok {
+			break
+		}
+		batch = append(batch, it)
+	}
+	if len(batch) == 0 {
+		return
+	}
+	defer func() {
+		if r := recover(); r != nil { // must never escape
+			log.Printf("[failure_auto_filer] auto-filer inline drain failed: %v", r)
+		}
+		for range batch {
+			f.queue.taskDone()
+		}
+	}()
+	f.processBatch(batch)
+}

src/go/failure-filer/frozen_failure_auto_filer_truth.json

@@ -0,0 +1,314 @@
+[
+  {
+    "fn": "error_class_of",
+    "args": [
+      "RuntimeError: boom"
+    ],
+    "kwargs": {},
+    "expected": "RuntimeError"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "plain token error"
+    ],
+    "kwargs": {},
+    "expected": "plain"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      ""
+    ],
+    "kwargs": {},
+    "expected": "Unknown"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "   "
+    ],
+    "kwargs": {},
+    "expected": "Unknown"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "ValueError:"
+    ],
+    "kwargs": {},
+    "expected": "ValueError"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "ValueError::"
+    ],
+    "kwargs": {},
+    "expected": "ValueError"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA: tail"
+    ],
+    "kwargs": {},
+    "expected": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+    ],
+    "kwargs": {},
+    "expected": "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "::weird::"
+    ],
+    "kwargs": {},
+    "expected": "::weird"
+  },
+  {
+    "fn": "error_class_of",
+    "args": [
+      "BrokerError: postgres://u:p@h"
+    ],
+    "kwargs": {},
+    "expected": "BrokerError"
+  },
+  {
+    "fn": "signature_of",
+    "args": [
+      "api",
+      "TimeoutError: down"
+    ],
+    "kwargs": {},
+    "expected": [
+      "api",
+      "TimeoutError"
+    ]
+  },
+  {
+    "fn": "signature_of",
+    "args": [
+      "api",
+      ""
+    ],
+    "kwargs": {},
+    "expected": [
+      "api",
+      "Unknown"
+    ]
+  },
+  {
+    "fn": "tier_gated",
+    "args": [
+      0
+    ],
+    "kwargs": {},
+    "expected": false
+  },
+  {
+    "fn": "tier_gated",
+    "args": [
+      1
+    ],
+    "kwargs": {},
+    "expected": false
+  },
+  {
+    "fn": "tier_gated",
+    "args": [
+      2
+    ],
+    "kwargs": {},
+    "expected": true
+  },
+  {
+    "fn": "tier_gated",
+    "args": [
+      3
+    ],
+    "kwargs": {},
+    "expected": true
+  },
+  {
+    "fn": "marker_for",
+    "args": [
+      "api",
+      "TimeoutError"
+    ],
+    "kwargs": {},
+    "expected": "[auto-failure:api:TimeoutError]"
+  },
+  {
+    "fn": "marker_for",
+    "args": [
+      "",
+      ""
+    ],
+    "kwargs": {},
+    "expected": "[auto-failure::]"
+  },
+  {
+    "fn": "excerpt",
+    "args": [
+      "postgres://admin:hunter2@db.example.com:5432/rune?password=sekret&token=abc123&host=x"
+    ],
+    "kwargs": {},
+    "expected": "postgres://<redacted>:<redacted>@db.example.com:5432/rune?password=<redacted>"
+  },
+  {
+    "fn": "excerpt",
+    "args": [
+      "postgres://admin:hunter2@db.example.com:5432/rune?password=sekret&token=abc123&host=x"
+    ],
+    "kwargs": {
+      "limit": 40
+    },
+    "expected": "postgres://<redacted>:<redacted>@db.exam"
+  },
+  {
+    "fn": "excerpt",
+    "args": [
+      "short"
+    ],
+    "kwargs": {
+      "limit": 200
+    },
+    "expected": "short"
+  },
+  {
+    "fn": "excerpt",
+    "args": [
+      "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+    ],
+    "kwargs": {},
+    "expected": "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+  },
+  {
+    "fn": "excerpt",
+    "args": [
+      "xxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxxx"
+    ],
+    "kwargs": {
+      "limit": 5
+    },
+    "expected": "xxxxx"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "RuntimeError: boom"
+    ],
+    "kwargs": {},
+    "expected": "RuntimeError: boom"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "GET https://h/x?token=tok123&ok=1 done"
+    ],
+    "kwargs": {},
+    "expected": "GET https://h/x?token=<redacted>&ok=1 done"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "GET https://h/x?API_KEY=ABC&sig=s1&access_token=at9 done"
+    ],
+    "kwargs": {},
+    "expected": "GET https://h/x?API_KEY=<redacted> done"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "postgres://admin:hunter2@db.example.com:5432/rune?password=sekret&token=abc123&host=x"
+    ],
+    "kwargs": {},
+    "expected": "postgres://<redacted>:<redacted>@db.example.com:5432/rune?password=<redacted>"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "redis://:s3cret@cache:6379"
+    ],
+    "kwargs": {},
+    "expected": "redis://<redacted>:<redacted>@cache:6379"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "https://user:pass@host/path"
+    ],
+    "kwargs": {},
+    "expected": "https://<redacted>:<redacted>@host/path"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "Authorization: Bearer abc.def.ghi expired"
+    ],
+    "kwargs": {},
+    "expected": "Authorization: Bearer <redacted> expired"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "bearer tok-here trailing"
+    ],
+    "kwargs": {},
+    "expected": "bearer <redacted> trailing"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "password=hunter2 and secret: 'quoted'"
+    ],
+    "kwargs": {},
+    "expected": "password=<redacted> and secret: <redacted>'"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "api_key='abcd1234' auth: short"
+    ],
+    "kwargs": {},
+    "expected": "api_key=<redacted>' auth: <redacted>"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "password=short"
+    ],
+    "kwargs": {},
+    "expected": "password=<redacted>"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "auth=x1 auth=abcd1234"
+    ],
+    "kwargs": {},
+    "expected": "auth=x1 auth=<redacted>"
+  },
+  {
+    "fn": "redact",
+    "args": [
+      "connect postgres://u:p@h then ?token=t&x=2"
+    ],
+    "kwargs": {},
+    "expected": "connect postgres://<redacted>:<redacted>@h then ?token=<redacted>&x=2"
+  },
+  {
+    "fn": "_error_text",
+    "args": [
+      "plain string"
+    ],
+    "kwargs": {},
+    "expected": "plain string"
+  }
+]

src/go/failure-filer/frozen_filer_scenarios.json

@@ -0,0 +1,514 @@
+[
+  {
+    "name": "submit_t2_gate",
+    "trace": {
+      "calls": [],
+      "records": [],
+      "consumed": [],
+      "returns": []
+    }
+  },
+  {
+    "name": "submit_coalesce_create",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:api:TimeoutError] api",
+          "Auto-filed failure signature `api:TimeoutError`\n\n- tier=1\n- occurrences=3\n- last: TimeoutError: t\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": []
+    }
+  },
+  {
+    "name": "file_first_only",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:s:E]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:s:E] s",
+          "Auto-filed failure signature `s:E`\n\n- tier=1\n- occurrences=1\n- last: E: a\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": [
+        true,
+        false,
+        false
+      ]
+    }
+  },
+  {
+    "name": "file_batch_full",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:s:E]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:s:E] s",
+          "Auto-filed failure signature `s:E`\n\n- tier=1\n- occurrences=1\n- last: E: a\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": [
+        true,
+        false,
+        false
+      ]
+    }
+  },
+  {
+    "name": "file_window_expiry",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:s:E]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:s:E] s",
+          "Auto-filed failure signature `s:E`\n\n- tier=1\n- occurrences=1\n- last: E: a\n\n(detected by the failure auto-filer, #122)"
+        ],
+        [
+          "search_issues",
+          "[auto-failure:s:E]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:s:E] s",
+          "Auto-filed failure signature `s:E`\n\n- tier=1\n- occurrences=2\n- last: E: b\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": [
+        true,
+        true
+      ]
+    }
+  },
+  {
+    "name": "file_comment_existing",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:s:E]"
+        ],
+        [
+          "add_comment",
+          7,
+          "+1 occurrences (last: E: latest text @ <TS>)"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": [
+        true
+      ]
+    }
+  },
+  {
+    "name": "file_t2",
+    "trace": {
+      "calls": [],
+      "records": [],
+      "consumed": [],
+      "returns": [
+        false
+      ]
+    }
+  },
+  {
+    "name": "file_client_failure_spools",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ]
+      ],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "api",
+            "TimeoutError"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "TimeoutError: down",
+          "client_error": "forge down"
+        }
+      ],
+      "consumed": [],
+      "returns": [
+        false
+      ]
+    }
+  },
+  {
+    "name": "file_single_flight",
+    "trace": {
+      "calls": [],
+      "records": [],
+      "consumed": [],
+      "returns": [
+        false
+      ]
+    }
+  },
+  {
+    "name": "file_spool_journal_failure_never_raises",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": [
+        false
+      ]
+    }
+  },
+  {
+    "name": "call_dunder_is_submit",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:ValueError]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:api:ValueError] api",
+          "Auto-filed failure signature `api:ValueError`\n\n- tier=1\n- occurrences=1\n- last: ValueError: boom\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": []
+    }
+  },
+  {
+    "name": "reconcile_groups_and_marks_consumed",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:api:TimeoutError] api",
+          "Auto-filed failure signature `api:TimeoutError`\n\n- tier=1\n- occurrences=1\n- last: TimeoutError: t1\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "api",
+            "TimeoutError"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "TimeoutError: t1",
+          "client_error": "forge down"
+        }
+      ],
+      "consumed": [
+        "spool-det-0"
+      ],
+      "report": {
+        "replayed": 1,
+        "filed": 1,
+        "failed": 0,
+        "skipped": 0
+      }
+    }
+  },
+  {
+    "name": "reconcile_comment_existing",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "add_comment",
+          9,
+          "+1 occurrences while offline (last: TimeoutError: t1 @ <TS>)"
+        ]
+      ],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "api",
+            "TimeoutError"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "TimeoutError: t1",
+          "client_error": "forge down"
+        }
+      ],
+      "consumed": [
+        "spool-det-0"
+      ],
+      "report": {
+        "replayed": 1,
+        "filed": 1,
+        "failed": 0,
+        "skipped": 0
+      }
+    }
+  },
+  {
+    "name": "reconcile_t2_skipped",
+    "trace": {
+      "calls": [],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "cron",
+            "Noise"
+          ],
+          "tier": 2,
+          "count": 1,
+          "error": "noise",
+          "client_error": "-"
+        }
+      ],
+      "consumed": [
+        "spool-det-0"
+      ],
+      "report": {
+        "replayed": 1,
+        "filed": 0,
+        "failed": 0,
+        "skipped": 1
+      }
+    }
+  },
+  {
+    "name": "reconcile_malformed_line",
+    "trace": {
+      "calls": [],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "only-one"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "x",
+          "client_error": "-"
+        }
+      ],
+      "consumed": [],
+      "report": {
+        "replayed": 1,
+        "filed": 0,
+        "failed": 1,
+        "skipped": 0
+      }
+    }
+  },
+  {
+    "name": "reconcile_failing_client_keeps_spooled",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ]
+      ],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "api",
+            "TimeoutError"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "TimeoutError: t1",
+          "client_error": "forge down"
+        }
+      ],
+      "consumed": [],
+      "report": {
+        "replayed": 1,
+        "filed": 0,
+        "failed": 1,
+        "skipped": 0
+      }
+    }
+  },
+  {
+    "name": "reconcile_journal_read_failure",
+    "trace": {
+      "calls": [],
+      "records": [],
+      "consumed": [],
+      "report": {
+        "replayed": 0,
+        "filed": 0,
+        "failed": 1,
+        "skipped": 0
+      }
+    }
+  },
+  {
+    "name": "reconcile_null_adapter_result",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:api:TimeoutError] api",
+          "Auto-filed failure signature `api:TimeoutError`\n\n- tier=1\n- occurrences=1\n- last: TimeoutError: t1\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "api",
+            "TimeoutError"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "TimeoutError: t1",
+          "client_error": "forge down"
+        }
+      ],
+      "consumed": [],
+      "report": {
+        "replayed": 1,
+        "filed": 0,
+        "failed": 1,
+        "skipped": 0
+      }
+    }
+  },
+  {
+    "name": "queue_full_spools_overflow",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:api:TimeoutError] api",
+          "Auto-filed failure signature `api:TimeoutError`\n\n- tier=1\n- occurrences=2\n- last: TimeoutError: t\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-0",
+          "timestamp": "<TS>",
+          "signature": [
+            "api",
+            "TimeoutError"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "TimeoutError: t",
+          "client_error": "auto-filer queue full"
+        },
+        {
+          "kind": "failure_auto_file_spool",
+          "spool_id": "spool-det-1",
+          "timestamp": "<TS>",
+          "signature": [
+            "api",
+            "TimeoutError"
+          ],
+          "tier": 1,
+          "count": 1,
+          "error": "TimeoutError: t",
+          "client_error": "auto-filer queue full"
+        }
+      ],
+      "consumed": [],
+      "returns": []
+    }
+  },
+  {
+    "name": "drain_twice_idempotent",
+    "trace": {
+      "calls": [
+        [
+          "search_issues",
+          "[auto-failure:api:TimeoutError]"
+        ],
+        [
+          "create_issue",
+          "[auto-failure:api:TimeoutError] api",
+          "Auto-filed failure signature `api:TimeoutError`\n\n- tier=1\n- occurrences=1\n- last: TimeoutError: t\n\n(detected by the failure auto-filer, #122)"
+        ]
+      ],
+      "records": [],
+      "consumed": [],
+      "returns": []
+    }
+  }
+]

src/go/failure-filer/main.go

@@ -22,7 +22,7 @@ import (
 )
 
 const ERROR_CLASS_MAX = 64
-const EXCERPT_LEN = 256
+const EXCERPT_LEN = 200
 
 type SpoolEntry struct {
 	ID         string    `json:"spool_id"`
@@ -81,7 +81,8 @@ func errorClassOf(errMsg string) string {
 	if len(tokens) == 0 {
 		return "Unknown"
 	}
-	class := strings.TrimSuffix(tokens[0], ":")
+	// oracle: tokens[0].rstrip(":") — strips ALL trailing colons
+	class := strings.TrimRight(tokens[0], ":")
 	if len(class) > ERROR_CLASS_MAX {
 		class = class[:ERROR_CLASS_MAX]
 	}
@@ -113,8 +114,10 @@ func redact(text string) string {
 
 func excerpt(errMsg string, limit int) string {
 	r := redact(errMsg)
-	if len(r) > limit {
-		r = r[:limit]
+	// oracle slices the str (codepoints), not bytes
+	runes := []rune(r)
+	if len(runes) > limit {
+		return string(runes[:limit])
 	}
 	return r
 }

src/go/failure-filer/scenario_test.go

@@ -0,0 +1,381 @@
+package main
+
+// scenario_test.go — frozen-oracle scenario battery for the pipeline
+// fns (frozen_filer_scenarios.json, generated by
+// scripts/gen_failure_auto_filer_truth.py). Each scenario runs the
+// oracle's FailureAutoFiler(wait=True) with a fake client + fake
+// journal + pinned `_now`; the Go twin replays the identical steps with
+// the identical fakes (below, mirroring the generator 1:1) and the
+// captured trace must match VERBATIM. No goroutine spawns: wait=true.
+
+import (
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"regexp"
+	"testing"
+)
+
+var tsPattern = regexp.MustCompile(`\d{4}-\d\d-\d\dT\d\d:\d\d:\d\dZ`)
+
+func normTS(s string) string { return tsPattern.ReplaceAllString(s, "<TS>") }
+
+// ---- fakes (mirror gen_failure_auto_filer_truth.FakeClient/FakeJournal) ----
+
+type fakeClient struct {
+	calls      []any
+	issues     []map[string]any
+	fail       bool
+	nullCreate bool
+}
+
+func (c *fakeClient) SearchIssues(marker string) ([]map[string]any, error) {
+	c.calls = append(c.calls, []any{"search_issues", marker})
+	if c.fail {
+		return nil, errors.New("forge down")
+	}
+	out := []map[string]any{}
+	for _, i := range c.issues {
+		cp := map[string]any{}
+		for k, v := range i {
+			cp[k] = v
+		}
+		out = append(out, cp)
+	}
+	return out, nil
+}
+
+func (c *fakeClient) AddComment(number int, body string) error {
+	c.calls = append(c.calls, []any{"add_comment", number, normTS(body)})
+	if c.fail {
+		return errors.New("forge down")
+	}
+	return nil
+}
+
+func (c *fakeClient) CreateIssue(title, body string) (map[string]any, error) {
+	c.calls = append(c.calls, []any{"create_issue", title, normTS(body)})
+	if c.fail {
+		return nil, errors.New("forge down")
+	}
+	if c.nullCreate {
+		return map[string]any{"created": false}, nil
+	}
+	return map[string]any{"number": len(c.issues) + 1, "created": true}, nil
+}
+
+type fakeJournal struct {
+	records        []map[string]any
+	consumed       []any
+	failUnconsumed bool
+	failAppend     bool
+}
+
+func (j *fakeJournal) Append(record map[string]any) error {
+	if j.failAppend {
+		return errors.New("journal append failed")
+	}
+	cp := map[string]any{}
+	for k, v := range record {
+		cp[k] = v
+	}
+	cp["spool_id"] = fmt.Sprintf("spool-det-%d", len(j.records))
+	cp["timestamp"] = "<TS>"
+	j.records = append(j.records, cp)
+	return nil
+}
+
+func (j *fakeJournal) Unconsumed() ([]map[string]any, error) {
+	if j.failUnconsumed {
+		return nil, errors.New("journal read failed")
+	}
+	consumed := map[string]bool{}
+	for _, id := range j.consumed {
+		consumed[asStr(id)] = true
+	}
+	out := []map[string]any{}
+	for _, r := range j.records {
+		if !consumed[asStr(r["spool_id"])] {
+			cp := map[string]any{}
+			for k, v := range r {
+				cp[k] = v
+			}
+			out = append(out, cp)
+		}
+	}
+	return out, nil
+}
+
+func (j *fakeJournal) MarkConsumed(spoolID string) {
+	j.consumed = append(j.consumed, spoolID)
+}
+
+// ---- trace plumbing ----
+
+func nonNil[V any](s []V) []V {
+	if s == nil {
+		return []V{}
+	}
+	return s
+}
+
+func traceOf(client *fakeClient, journal *fakeJournal) map[string]any {
+	return map[string]any{
+		"calls":    nonNil(client.calls),
+		"records":  nonNil(journal.records),
+		"consumed": nonNil(journal.consumed),
+	}
+}
+
+func canonical(v any) string {
+	b, _ := json.Marshal(v)
+	return string(b)
+}
+
+func freshFiler(client *fakeClient, journal *fakeJournal, batchSize int) *FailureAutoFiler {
+	return NewFailureAutoFiler(client, journal, batchSize, true)
+}
+
+func runScenario(t *testing.T, name string, client *fakeClient, journal *fakeJournal) map[string]any {
+	t.Helper()
+	switch name {
+	case "submit_t2_gate":
+		f := freshFiler(client, journal, 10)
+		f.Submit("sub", "ValueError: boom", 2, 100)
+		f.Drain(-1)
+		return mergeReturns(traceOf(client, journal), []any{})
+	case "submit_coalesce_create":
+		f := freshFiler(client, journal, 10)
+		for i := 0; i < 3; i++ {
+			f.Submit("api", "TimeoutError: t", 1, float64(100+i))
+		}
+		f.Drain(-1)
+		return mergeReturns(traceOf(client, journal), []any{})
+	case "file_first_only":
+		f := freshFiler(client, journal, 10)
+		r1 := f.File("s", "E: a", 1, 100)
+		r2 := f.File("s", "E: b", 1, 101)
+		r3 := f.File("s", "E: c", 1, 102)
+		return mergeReturns(traceOf(client, journal), []any{r1, r2, r3})
+	case "file_batch_full":
+		f := freshFiler(client, journal, 3)
+		r1 := f.File("s", "E: a", 1, 100)
+		r2 := f.File("s", "E: b", 1, 101)
+		r3 := f.File("s", "E: c", 1, 102)
+		return mergeReturns(traceOf(client, journal), []any{r1, r2, r3})
+	case "file_window_expiry":
+		f := freshFiler(client, journal, 10)
+		r1 := f.File("s", "E: a", 1, 100)
+		r2 := f.File("s", "E: b", 1, 100+600)
+		return mergeReturns(traceOf(client, journal), []any{r1, r2})
+	case "file_comment_existing":
+		client.issues = []map[string]any{{"number": 7, "title": "[auto-failure:s:E] s"}}
+		f := freshFiler(client, journal, 10)
+		r1 := f.File("s", "E: latest text", 1, 100)
+		return mergeReturns(traceOf(client, journal), []any{r1})
+	case "file_t2":
+		f := freshFiler(client, journal, 10)
+		r1 := f.File("s", "E: cron noise", 2, 100)
+		return mergeReturns(traceOf(client, journal), []any{r1})
+	case "file_client_failure_spools":
+		client.fail = true
+		f := freshFiler(client, journal, 10)
+		r1 := f.File("api", "TimeoutError: down", 1, 100)
+		return mergeReturns(traceOf(client, journal), []any{r1})
+	case "file_single_flight":
+		f := freshFiler(client, journal, 10)
+		f.mu.Lock()
+		f.inflight[sigKey{subsystem: "s", class: "E"}] = true
+		f.mu.Unlock()
+		r1 := f.File("s", "E: x", 1, 100)
+		return mergeReturns(traceOf(client, journal), []any{r1})
+	case "file_spool_journal_failure_never_raises":
+		client.fail = true
+		journal.failAppend = true
+		f := freshFiler(client, journal, 10)
+		r1 := f.File("api", "TimeoutError: down", 1, 100)
+		return mergeReturns(traceOf(client, journal), []any{r1})
+	case "call_dunder_is_submit":
+		f := freshFiler(client, journal, 10)
+		f.Call("api", "ValueError: boom", 1, 100)
+		f.Drain(-1)
+		return mergeReturns(traceOf(client, journal), []any{})
+	case "reconcile_groups_and_marks_consumed":
+		client.fail = true
+		f := freshFiler(client, journal, 10)
+		f.File("api", "TimeoutError: t1", 1, 100)
+		f.File("api", "TimeoutError: t2", 1, 101)
+		client.fail = false
+		report := f.Reconcile()
+		return mergeReport(traceOf(client, journal), report)
+	case "reconcile_comment_existing":
+		client.fail = true
+		f := freshFiler(client, journal, 10)
+		f.File("api", "TimeoutError: t1", 1, 100)
+		client.fail = false
+		client.issues = []map[string]any{{"number": 9, "title": "[auto-failure:api:TimeoutError] api"}}
+		report := f.Reconcile()
+		return mergeReport(traceOf(client, journal), report)
+	case "reconcile_t2_skipped":
+		f := freshFiler(client, journal, 10)
+		journal.Append(map[string]any{
+			"kind": SPOOL_KIND, "spool_id": "spool-cron", "timestamp": "<TS>",
+			"signature": []any{"cron", "Noise"}, "tier": 2, "count": 1,
+			"error": "noise", "client_error": "-",
+		})
+		report := f.Reconcile()
+		return mergeReport(traceOf(client, journal), report)
+	case "reconcile_malformed_line":
+		f := freshFiler(client, journal, 10)
+		journal.Append(map[string]any{
+			"kind": SPOOL_KIND, "spool_id": "spool-bad", "timestamp": "<TS>",
+			"signature": []any{"only-one"}, "tier": 1, "count": 1,
+			"error": "x", "client_error": "-",
+		})
+		report := f.Reconcile()
+		return mergeReport(traceOf(client, journal), report)
+	case "reconcile_failing_client_keeps_spooled":
+		client.fail = true
+		f := freshFiler(client, journal, 10)
+		f.File("api", "TimeoutError: t1", 1, 100)
+		report := f.Reconcile()
+		return mergeReport(traceOf(client, journal), report)
+	case "reconcile_journal_read_failure":
+		journal.failUnconsumed = true
+		f := freshFiler(client, journal, 10)
+		report := f.Reconcile()
+		return mergeReport(traceOf(client, journal), report)
+	case "reconcile_null_adapter_result":
+		client.fail = true
+		f := freshFiler(client, journal, 10)
+		f.File("api", "TimeoutError: t1", 1, 100)
+		nullClient := &fakeClient{nullCreate: true}
+		f.client = nullClient
+		report := f.Reconcile()
+		return mergeReport(traceOf(nullClient, journal), report)
+	case "queue_full_spools_overflow":
+		f := freshFiler(client, journal, 10)
+		f.queue = newFilerQueue(2)
+		for i := 0; i < 4; i++ {
+			f.Submit("api", "TimeoutError: t", 1, 100)
+		}
+		f.Drain(-1)
+		return mergeReturns(traceOf(client, journal), []any{})
+	case "drain_twice_idempotent":
+		f := freshFiler(client, journal, 10)
+		f.Submit("api", "TimeoutError: t", 1, 100)
+		f.Drain(-1)
+		f.Drain(-1)
+		return mergeReturns(traceOf(client, journal), []any{})
+	default:
+		t.Fatalf("unknown scenario: %s", name)
+		return nil
+	}
+}
+
+func mergeReturns(trace map[string]any, returns []any) map[string]any {
+	trace["returns"] = returns
+	return trace
+}
+
+func mergeReport(trace map[string]any, report map[string]int) map[string]any {
+	trace["report"] = report
+	return trace
+}
+
+func TestFrozenScenarioBattery(t *testing.T) {
+	raw, err := os.ReadFile("frozen_filer_scenarios.json")
+	if err != nil {
+		t.Fatalf("frozen scenarios missing: %v", err)
+	}
+	var scenarios []struct {
+		Name  string         `json:"name"`
+		Trace map[string]any `json:"trace"`
+	}
+	if err := json.Unmarshal(raw, &scenarios); err != nil {
+		t.Fatalf("unparseable: %v", err)
+	}
+	for _, sc := range scenarios {
+		sc := sc
+		t.Run(sc.Name, func(t *testing.T) {
+			client, journal := &fakeClient{}, &fakeJournal{}
+			got := runScenario(t, sc.Name, client, journal)
+			if canonical(got) != canonical(sc.Trace) {
+				t.Errorf("trace mismatch\n got: %s\nwant: %s", canonical(got), canonical(sc.Trace))
+			}
+		})
+	}
+}
+
+// JsonlSpool round-trip: append → unconsumed → mark_consumed → empty
+// (oracle sidecar dialect; consumption never rewrites the journal).
+func TestJsonlSpoolSidecarRoundTrip(t *testing.T) {
+	dir := t.TempDir()
+	path := dir + "/journal.jsonl"
+	sp := NewJsonlSpool(path)
+	if rows, _ := sp.Unconsumed(); len(rows) != 0 {
+		t.Fatalf("empty journal: unconsumed = %v", rows)
+	}
+	if err := sp.Append(map[string]any{"event_id": "e1", "kind": "canonical"}); err != nil {
+		t.Fatalf("append: %v", err)
+	}
+	if err := sp.Append(map[string]any{"kind": SPOOL_KIND, "spool_id": "spool-1", "tier": 1}); err != nil {
+		t.Fatalf("append: %v", err)
+	}
+	if err := sp.Append(map[string]any{"kind": SPOOL_KIND, "spool_id": "spool-2", "tier": 1}); err != nil {
+		t.Fatalf("append: %v", err)
+	}
+	// a malformed line rides along: never crash (oracle JSONDecodeError skip)
+	malformed, _ := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0o644)
+	malformed.WriteString("definitely not json\n")
+	malformed.Close()
+	rows, err := sp.Unconsumed()
+	if err != nil {
+		t.Fatalf("unconsumed: %v", err)
+	}
+	if len(rows) != 2 {
+		t.Fatalf("want 2 spool rows, got %d: %v", len(rows), rows)
+	}
+	if rows[0]["spool_id"] != "spool-1" {
+		t.Fatalf("row order preserved: %v", rows[0])
+	}
+	// a foreign consumed marker rides the sidecar, never the journal
+	sp.MarkConsumed("spool-1")
+	rows, _ = sp.Unconsumed()
+	if len(rows) != 1 || rows[0]["spool_id"] != "spool-2" {
+		t.Fatalf("after mark_consumed: %v", rows)
+	}
+	data, _ := os.ReadFile(path)
+	if got := countLines(string(data)); got != 4 {
+		t.Fatalf("journal must keep exactly 4 lines (never rewritten), got %d", got)
+	}
+}
+
+func countLines(s string) int {
+	n := 0
+	for _, r := range s {
+		if r == '\n' {
+			n++
+		}
+	}
+	return n
+}
+
+// runWorker smoke test, synchronously: items + shutdown sentinel are
+// queued up front, so runWorker drains the batch and returns without
+// any goroutine spawn (oracle worker loop, pinned scheduling).
+func TestRunWorkerSynchronous(t *testing.T) {
+	client, journal := &fakeClient{}, &fakeJournal{}
+	f := freshFiler(client, journal, 10)
+	f.Submit("api", "TimeoutError: t", 1, 100)
+	f.Submit("api", "TimeoutError: t", 1, 101)
+	f.queue.putSentinel()
+	f.runWorker()
+	if got := len(client.calls); got != 2 { // search + create, one coalesced batch
+		t.Fatalf("worker made %d client calls, want 2: %v", got, client.calls)
+	}
+	if u := f.queue.unfinishedTasks(); u != 0 {
+		t.Fatalf("unfinished tasks after worker exit: %d", u)
+	}
+}

src/go/failure-filer/spool.go

@@ -0,0 +1,146 @@
+package main
+
+// spool.go — the failure_auto_filer spool journal (oracle JsonlSpool, 1:1).
+//
+// Spooled records ride the EXISTING durable event journal as JSON lines
+// with `kind: failure_auto_file_spool` next to canonical envelopes.
+// Consumption NEVER rewrites the canonical journal (#122 gate): the
+// marker is an append-only sidecar `<journal>.consumed` (one spool id
+// per line, atomic appends only). Unconsumed() = spool lines minus
+// sidecar ids. Name map (oracle → here): append→Append,
+// unconsumed→Unconsumed, mark_consumed→MarkConsumed.
+
+import (
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
+	"strings"
+)
+
+const SPOOL_KIND = "failure_auto_file_spool"
+
+type JsonlSpool struct {
+	Path string
+}
+
+func NewJsonlSpool(path string) *JsonlSpool {
+	return &JsonlSpool{Path: path}
+}
+
+func (s *JsonlSpool) Append(record map[string]any) error {
+	if err := os.MkdirAll(filepath.Dir(s.Path), 0o755); err != nil {
+		return err
+	}
+	// json.Marshal of a map sorts keys — matches json.dumps(sort_keys=True)
+	data, err := json.Marshal(record)
+	if err != nil {
+		return err
+	}
+	fh, err := os.OpenFile(s.Path, os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
+	if err != nil {
+		return err
+	}
+	defer fh.Close()
+	_, err = fh.Write(append(data, '\n'))
+	return err
+}
+
+func (s *JsonlSpool) Unconsumed() ([]map[string]any, error) {
+	if _, err := os.Stat(s.Path); err != nil {
+		return []map[string]any{}, nil
+	}
+	sidecar := map[string]bool{}
+	if data, err := os.ReadFile(s.Path + ".consumed"); err == nil {
+		for _, line := range strings.Split(string(data), "\n") {
+			if t := strings.TrimSpace(line); t != "" {
+				sidecar[t] = true
+			}
+		}
+	}
+	data, err := os.ReadFile(s.Path)
+	if err != nil {
+		return []map[string]any{}, nil
+	}
+	rows := []map[string]any{}
+	for _, line := range strings.Split(string(data), "\n") {
+		if strings.TrimSpace(line) == "" {
+			continue
+		}
+		var row map[string]any
+		if json.Unmarshal([]byte(line), &row) != nil {
+			continue // malformed line: never crash (oracle JSONDecodeError → skip)
+		}
+		if row["kind"] == SPOOL_KIND && !truthy(row["consumed"]) && !sidecar[asStr(row["spool_id"])] {
+			rows = append(rows, row)
+		}
+	}
+	return rows, nil
+}
+
+func (s *JsonlSpool) MarkConsumed(spoolID string) {
+	// last resort on error: the line stays unconsumed; dedup via
+	// search_issues still prevents duplicate issues on replay
+	if err := os.MkdirAll(filepath.Dir(s.Path), 0o755); err != nil {
+		return
+	}
+	fh, err := os.OpenFile(s.Path+".consumed", os.O_APPEND|os.O_CREATE|os.O_WRONLY, 0o644)
+	if err != nil {
+		return
+	}
+	defer fh.Close()
+	_, _ = fh.WriteString(spoolID + "\n")
+}
+
+// truthy mirrors Python truthiness for the `consumed` flag probe
+// (absent/None/False/0/"" are all falsy).
+func truthy(v any) bool {
+	switch t := v.(type) {
+	case nil:
+		return false
+	case bool:
+		return t
+	case string:
+		return t != ""
+	case float64:
+		return t != 0
+	default:
+		return true
+	}
+}
+
+func asStr(v any) string {
+	switch t := v.(type) {
+	case nil:
+		return ""
+	case string:
+		return t
+	default:
+		return fmt.Sprintf("%v", t)
+	}
+}
+
+// asNumber mirrors the oracle's int(...) coercions across JSON-decoded values.
+func asNumber(v any, def float64) float64 {
+	switch t := v.(type) {
+	case float64:
+		return t
+	case int:
+		return float64(t)
+	case int64:
+		return float64(t)
+	case string:
+		var f float64
+		if _, err := fmt.Sscan(t, &f); err == nil {
+			return f
+		}
+		return def
+	case bool:
+		if t {
+			return 1
+		}
+		return 0
+	default:
+		return def
+	}
+}