myxce/scripts/buglog.py

#!/usr/bin/env python3
"""Read and edit .wolf/buglog.json without re-serializing it.

Re-serializing the whole buglog (json.dumps / jq) rewrites ~1,200 unrelated
lines because existing entries mix raw UTF-8 and \\uXXXX escapes, so every
write here edits only the lines of the entry it touches. Every write is
re-parsed before it is saved and saved atomically (tmp file + rename), the
same way OpenWolf's own hooks write it.

The `status` values and when to use each are defined in
.claude/rules/buglog-status.md.

Commands:
  add       --error MSG [--root-cause T] [--fix T] [--file F] [--tags a,b]
            [--related bug-1,bug-2] [--status S] [--note T]
            Append a new entry with the next free id; prints the id.
  status    ID STATUS [--note T] [--duplicate-of ID] [--repro T]
            Set status + status_updated (today) + status_note. --note is
            required except for `fixed`; duplicate needs --duplicate-of,
            cannot-reproduce needs --repro.
  note      ID TEXT       Append an "UPDATE (date): TEXT" paragraph to `fix`.
  recur     ID [--note T] The bug was seen again: bump occurrences/last_seen;
                          cannot-reproduce -> investigating, fixed/wontfix -> open.
  set       ID key=<json> ...   Set arbitrary single-line fields.
  show      ID            Print one entry.
  list      [--status S,S | --open] [--tag T] [--grep RE] [--since DATE]
            One line per matching entry.
  triage                  List entries with no (or an unknown) status.
  next-id                 Print the id `add` would use.

Importable: set_fields(path, bug_id, {key: value}), BuglogEditError.
"""
import argparse
import datetime
import json
import os
import re
import sys
from pathlib import Path

DEFAULT_PATH = Path(__file__).resolve().parent.parent / ".wolf" / "buglog.json"

STATUSES = ["open", "partial", "investigating", "fixed",
            "cannot-reproduce", "wontfix", "duplicate"]
OPEN_STATUSES = {"open", "partial", "investigating"}


class BuglogEditError(Exception):
    pass


def _today():
    return datetime.date.today().isoformat()


def _now():
    return datetime.datetime.now(datetime.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")


def _load(path):
    return json.loads(Path(path).read_text(encoding="utf-8"))["bugs"]


def _save(path, text):
    """Refuse to write text that no longer parses; write atomically."""
    json.loads(text)
    path = Path(path)
    tmp = path.with_name(f"{path.name}.{os.getpid()}.tmp")
    tmp.write_text(text, encoding="utf-8")
    os.replace(tmp, path)


def _entry_span(lines, bug_id):
    """(start, end, indent) of the entry: start = its "id" line, end = its '}'."""
    id_re = re.compile(r'^(\s+)"id": ' + re.escape(json.dumps(bug_id)) + r",?$")
    starts = [i for i, ln in enumerate(lines) if id_re.match(ln)]
    if len(starts) != 1:
        raise BuglogEditError(f"{bug_id}: expected exactly one entry, found {len(starts)}")
    start = starts[0]
    indent = id_re.match(lines[start]).group(1)
    close_re = re.compile(r"^" + re.escape(indent[:-2]) + r"\},?$")
    end = next(i for i in range(start, len(lines)) if close_re.match(lines[i]))
    return start, end, indent


def set_fields(path, bug_id, fields):
    """Set each key in `fields` on entry `bug_id`, touching only its lines.

    An existing single-line field is replaced in place; a new one is inserted
    right after the entry's "id" line. Multi-line (array/object) fields are
    refused -- edit those by hand.
    """
    path = Path(path)
    lines = path.read_text(encoding="utf-8").split("\n")
    start, end, indent = _entry_span(lines, bug_id)

    for key, value in fields.items():
        rendered = json.dumps(value, ensure_ascii=False)
        key_re = re.compile(r"^" + re.escape(indent) + re.escape(json.dumps(key)) + r": (.*?)(,?)$")
        hit = next((i for i in range(start, end) if key_re.match(lines[i])), None)
        if hit is not None:
            old, comma = key_re.match(lines[hit]).groups()
            if old in ("[", "{"):
                raise BuglogEditError(f"{bug_id}.{key}: multi-line value, edit by hand")
            lines[hit] = f"{indent}{json.dumps(key)}: {rendered}{comma}"
        else:
            lines.insert(start + 1, f"{indent}{json.dumps(key)}: {rendered},")
            end += 1

    text = "\n".join(lines)
    entry = next(b for b in json.loads(text)["bugs"] if b.get("id") == bug_id)
    for key, value in fields.items():
        if entry[key] != value:
            raise BuglogEditError(f"{bug_id}.{key}: wrote {entry[key]!r}, wanted {value!r}")
    _save(path, text)


def next_id(bugs):
    """One past the highest numeric id. (OpenWolf's own hook uses len(bugs)+1,
    which collides with existing ids once any were renumbered or skipped.)"""
    nums = [int(m.group(1)) for b in bugs
            if (m := re.fullmatch(r"bug-(\d+)", str(b.get("id", ""))))]
    return f"bug-{max(nums, default=0) + 1:03d}"


def append_entry(path, entry):
    """Append `entry` as the last element of the `bugs` array."""
    path = Path(path)
    lines = path.read_text(encoding="utf-8").split("\n")
    close = max(i for i, ln in enumerate(lines) if ln == "  ]")
    prev = close - 1
    if lines[prev] == "    }":
        lines[prev] = "    },"
    elif lines[prev] != "  \"bugs\": [":
        raise BuglogEditError(f"unexpected line before bugs array end: {lines[prev]!r}")
    body = json.dumps(entry, indent=2, ensure_ascii=False).split("\n")
    lines[close:close] = ["    " + ln for ln in body]
    text = "\n".join(lines)
    if json.loads(text)["bugs"][-1] != entry:
        raise BuglogEditError("appended entry did not round-trip")
    _save(path, text)


def _split(value):
    return [v.strip() for v in (value or "").split(",") if v.strip()]


def _text(value):
    if isinstance(value, list):
        return " || ".join(str(v) for v in value)
    return value or ""


def _find(bugs, bug_id):
    for b in bugs:
        if b.get("id") == bug_id:
            return b
    raise BuglogEditError(f"{bug_id}: no such entry")


def _line(b):
    note = b.get("status_note") or _text(b.get("error_message"))
    return f"{b['id']:<10} {b.get('status') or '-':<16} {note[:110]}"


def cmd_add(a, path):
    if a.status not in STATUSES:
        sys.exit(f"unknown status {a.status!r}; one of {', '.join(STATUSES)}")
    bugs = _load(path)
    bid = next_id(bugs)
    now = _now()
    entry = {"id": bid, "status": a.status, "status_updated": _today()}
    if a.note:
        entry["status_note"] = a.note
    entry.update({
        "timestamp": now,
        "error_message": a.error,
        "file": a.file or "",
        "root_cause": a.root_cause or "",
        "fix": a.fix or "",
        "tags": _split(a.tags),
        "related_bugs": _split(a.related),
        "occurrences": 1,
        "last_seen": now,
    })
    append_entry(path, entry)
    print(bid)


def cmd_status(a, path):
    if a.status not in STATUSES:
        sys.exit(f"unknown status {a.status!r}; one of {', '.join(STATUSES)}")
    if a.status == "duplicate" and not a.duplicate_of:
        sys.exit("duplicate needs --duplicate-of")
    if a.status == "cannot-reproduce" and not a.repro:
        sys.exit("cannot-reproduce needs --repro (what was tried, e.g. '0/40 runs, 2026-08-01')")
    if a.note is None and a.status != "fixed":
        sys.exit(f"{a.status} needs --note (what remains / why / what was seen)")
    # Always (re)write the note so a stale one never outlives its status.
    fields = {"status": a.status, "status_updated": _today(),
              "status_note": a.note or ""}
    if a.duplicate_of:
        fields["duplicate_of"] = a.duplicate_of
    if a.repro:
        fields["repro_attempts"] = a.repro
    set_fields(path, a.id, fields)
    print(_line(_find(_load(path), a.id)))


def cmd_note(a, path):
    fix = _find(_load(path), a.id).get("fix") or ""
    if isinstance(fix, list):
        raise BuglogEditError(f"{a.id}.fix is a list, edit by hand")
    para = f"UPDATE ({_today()}): {a.text}"
    set_fields(path, a.id, {"fix": f"{fix}\n\n{para}" if fix else para})


def cmd_recur(a, path):
    b = _find(_load(path), a.id)
    fields = {"occurrences": int(b.get("occurrences") or 1) + 1, "last_seen": _now()}
    reopen = {"cannot-reproduce": "investigating", "fixed": "open", "wontfix": "open"}
    if b.get("status") in reopen:
        fields.update(status=reopen[b["status"]], status_updated=_today())
        fields["status_note"] = a.note or f"recurred {_today()} (was {b['status']})"
    elif a.note:
        fields["status_note"] = a.note
    set_fields(path, a.id, fields)
    print(_line(_find(_load(path), a.id)))


def cmd_set(a, path):
    fields = {}
    for kv in a.pairs:
        key, sep, raw = kv.partition("=")
        if not sep or not key:
            sys.exit(f"bad argument {kv!r}: expected key=value")
        try:
            fields[key] = json.loads(raw)
        except json.JSONDecodeError:
            fields[key] = raw
    set_fields(path, a.id, fields)


def cmd_show(a, path):
    print(json.dumps(_find(_load(path), a.id), indent=2, ensure_ascii=False))


def cmd_list(a, path):
    want = OPEN_STATUSES if a.open else set(_split(a.status)) or None
    rx = re.compile(a.grep, re.I) if a.grep else None
    for b in _load(path):
        if want is not None and b.get("status") not in want:
            continue
        if a.tag and a.tag not in (b.get("tags") or []):
            continue
        if a.since and max(b.get("last_seen") or "", b.get("status_updated") or "") < a.since:
            continue
        if rx and not rx.search(json.dumps(b, ensure_ascii=False)):
            continue
        print(_line(b))


def cmd_triage(a, path):
    for b in _load(path):
        if b.get("status") not in STATUSES:
            print(_line(b))


def main(argv=None):
    p = argparse.ArgumentParser(description=__doc__.split("\n")[0],
                                epilog="Status values: " + ", ".join(STATUSES))
    p.add_argument("--path", default=str(DEFAULT_PATH), help=argparse.SUPPRESS)
    sub = p.add_subparsers(dest="cmd", required=True)

    s = sub.add_parser("add")
    s.add_argument("--error", required=True)
    s.add_argument("--root-cause")
    s.add_argument("--fix")
    s.add_argument("--file")
    s.add_argument("--tags")
    s.add_argument("--related")
    s.add_argument("--status", default="open")
    s.add_argument("--note")
    s.set_defaults(fn=cmd_add)

    s = sub.add_parser("status")
    s.add_argument("id")
    s.add_argument("status")
    s.add_argument("--note")
    s.add_argument("--duplicate-of")
    s.add_argument("--repro")
    s.set_defaults(fn=cmd_status)

    s = sub.add_parser("note")
    s.add_argument("id")
    s.add_argument("text")
    s.set_defaults(fn=cmd_note)

    s = sub.add_parser("recur")
    s.add_argument("id")
    s.add_argument("--note")
    s.set_defaults(fn=cmd_recur)

    s = sub.add_parser("set")
    s.add_argument("id")
    s.add_argument("pairs", nargs="+")
    s.set_defaults(fn=cmd_set)

    s = sub.add_parser("show")
    s.add_argument("id")
    s.set_defaults(fn=cmd_show)

    s = sub.add_parser("list")
    s.add_argument("--status")
    s.add_argument("--open", action="store_true")
    s.add_argument("--tag")
    s.add_argument("--grep")
    s.add_argument("--since")
    s.set_defaults(fn=cmd_list)

    s = sub.add_parser("triage")
    s.set_defaults(fn=cmd_triage)

    s = sub.add_parser("next-id")
    s.set_defaults(fn=lambda a, path: print(next_id(_load(path))))

    a = p.parse_args(argv)
    try:
        a.fn(a, a.path)
    except BuglogEditError as e:
        sys.exit(str(e))


if __name__ == "__main__":
    main()