sync: auto-sync from GURU-5070 at 2026-07-01 15:49:56

Author: Mike Swanson
Machine: GURU-5070
Timestamp: 2026-07-01 15:49:56

.claude/skills/errorlog-dream/scripts/selftest.py

@@ -0,0 +1,136 @@
+#!/usr/bin/env python
+"""selftest for errorlog-dream: run the analyzer against a synthetic errorlog
+in a temp dir and assert each detector fires, then assert --apply-archive
+moves exactly the old entries and leaves the marker, recent entries, and
+unparsed legacy blocks intact."""
+
+import io
+import os
+import shutil
+import sys
+import tempfile
+from datetime import datetime, timedelta, timezone
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import errorlog_dream as ed
+
+
+def build_fixture(root, today):
+    d = lambda n: (today - timedelta(days=n)).strftime("%Y-%m-%d")
+    old1 = d(95)
+    old2 = d(65)
+    recent = d(3)
+    lines = [
+        "# Error Log",
+        "",
+        ed.MARKER,
+        "",
+        # noise cluster: same machine+skill+shape on 3 distinct days
+        "%s | BOX-A | widget/api | HTTP 500 on id 12345 [ctx: cmd=list]" % d(1),
+        "",
+        "%s | BOX-A | widget/api | HTTP 500 on id 99881 [ctx: cmd=list] (x3)" % d(2),
+        "",
+        "%s | BOX-A | widget/api | HTTP 500 on id 40404 [ctx: cmd=list]" % recent,
+        "",
+        # repeat ref (2x) citing an existing memory
+        "%s | BOX-B | bash/env | [friction] used /tmp again [ctx: ref=fix_tmp_rule]" % d(4),
+        "",
+        "%s | BOX-B | bash/env | [friction] used /tmp AGAIN again [ctx: ref=fix_tmp_rule]" % recent,
+        "",
+        # machine-name case drift
+        "%s | box-b | coord | HTTP 0 talking to coord" % d(5),
+        "",
+        # resolved entry
+        "%s | BOX-A | sync/submodules | checkout aborted. [RESOLVED %s] fixed in sync.sh" % (d(6), d(5)),
+        "",
+        # correction
+        "%s | BOX-A | client/foo | [correction] assumed X; correct is Y" % d(7),
+        "",
+        # old entries -> archive candidates (two months)
+        "%s | BOX-A | oldskill | ancient failure one" % old1,
+        "",
+        "%s | BOX-B | oldskill | ancient failure two" % old2,
+        "",
+        # legacy multi-line unparsed block (no pipes on first line)
+        "Some legacy hand-written note",
+        "spanning two lines with no format.",
+        "",
+    ]
+    log = os.path.join(root, "errorlog.md")
+    with io.open(log, "w", encoding="utf-8", newline="\n") as fh:
+        fh.write("\n".join(lines))
+    mem = os.path.join(root, ".claude", "memory")
+    os.makedirs(mem)
+    with io.open(os.path.join(mem, "fix_tmp_rule.md"), "w", encoding="utf-8") as fh:
+        fh.write("---\nname: fix_tmp_rule\n---\nrule body\n")
+    return log
+
+
+def main():
+    tmp = tempfile.mkdtemp(prefix="eldream-test-")
+    failures = []
+    ok = lambda cond, name: failures.append(name) if not cond else None
+    try:
+        today = datetime.now(timezone.utc)
+        log = build_fixture(tmp, today)
+        header, entries, unparsed = ed.parse_log(log)
+
+        ok(len(entries) == 10, "parse: 10 entries (got %d)" % len(entries))
+        ok(len(unparsed) == 1, "parse: 1 unparsed block")
+        ok(any(e.count == 3 for e in entries), "parse: (x3) counter read")
+
+        r = ed.analyze(entries, unparsed, tmp, 60, today)
+        ok(r["weighted"] == 12, "weighted count incl x3 (got %d)" % r["weighted"])
+        ok(r["by_type"].get("friction") == 2 and r["by_type"].get("correction") == 1,
+           "type tally")
+        ok(any(ref == "fix_tmp_rule" and n == 2 and exists
+               for ref, n, exists, _ in r["repeat_refs"]),
+           "repeat-ref detector (existing memory)")
+        ok(any(skill == "widget/api" and w == 5 for _, skill, _, w, _ in r["noise"]),
+           "noise-cluster detector")
+        ok(len(r["resolved"]) == 1, "resolved detector")
+        ok("box-b" in r["machine_drift"], "machine-drift detector")
+        ok(len(r["archive"]) == 2, "archive candidates (got %d)" % len(r["archive"]))
+
+        report = ed.render(r, 60)
+        for token in ("STRENGTHEN?", "SUPPRESS?", "ARCHIVE?", "MACHINE-NAME DRIFT"):
+            ok(token in report, "report contains %s" % token)
+
+        # --apply-archive: moves the 2 old entries, keeps everything else
+        rc = ed.main(["--log", log, "--root", tmp, "--no-file", "--apply-archive"])
+        ok(rc == 0, "apply-archive exit 0")
+        h2, e2, u2 = ed.parse_log(log)
+        ok(len(e2) == 8, "post-archive: 8 entries kept (got %d)" % len(e2))
+        ok(len(u2) == 1, "post-archive: unparsed block untouched")
+        with io.open(log, encoding="utf-8") as fh:
+            body = fh.read()
+        ok(ed.MARKER in body, "post-archive: marker intact")
+        ok("ancient failure" not in body, "post-archive: old entries removed")
+        arch = os.path.join(tmp, "errorlog-archive")
+        months = sorted(f for f in os.listdir(arch) if f.endswith(".md"))
+        ok(len(months) == 2, "archive split by month (got %s)" % months)
+        joined = ""
+        for f in months:
+            with io.open(os.path.join(arch, f), encoding="utf-8") as fh:
+                joined += fh.read()
+        ok("ancient failure one" in joined and "ancient failure two" in joined,
+           "archived content present")
+
+        # idempotent: second run archives nothing
+        rc = ed.main(["--log", log, "--root", tmp, "--no-file", "--apply-archive"])
+        _, e3, _ = ed.parse_log(log)
+        ok(rc == 0 and len(e3) == 8, "second apply-archive is a no-op")
+    finally:
+        shutil.rmtree(tmp, ignore_errors=True)
+
+    if failures:
+        print("[ERROR] selftest FAILED:")
+        for f in failures:
+            print("  - " + f)
+        return 1
+    print("[OK] errorlog-dream selftest: all assertions passed")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())