claude-harness/docs/retro/scripts/hook_blocks_export.py

#!/usr/bin/env python3
"""Выгрузка блокировок хуков из транскриптов Claude Code в JSONL — сырьё для регрессионных тестов хуков (B2).

Запуск: python3 hook_blocks_export.py [--since ГГГГ-ММ-ДД] [--id-prefix <дата>] <каталог-транскриптов> [...]
Одна строка — одна блокировка: сессия, время, cwd и ветка на момент вызова, инструмент, команда или путь файла,
первая строка ответа хука. Поле expected не заполняется: вердикт (block/allow) проставляется при разметке.
Выгружаются только ошибки хуков (результат начинается с `PreToolUse:<инструмент> hook error`); вывод команд,
в котором встретилось слово BLOCKED (файлы памяти, исходники хуков), не попадает.
Транскрипты сабагентов (<сессия>/subagents/*.jsonl) выгружаются с сессией-родителем. --since — только вызовы
с меткой времени не раньше даты; --id-prefix (по умолчанию — дата --since) — поле id вида <префикс>-NNN (срез
ретро, B12). Похожие на секреты подстроки заменяются на <redacted> до обрезки.
"""
import glob
import json
import os
import re
import sys

HOOK_ERROR = re.compile(r"\s*PreToolUse:\S+ hook error")

RE_SECRET = re.compile(
    r'(?i)gh[pousr]_[A-Za-z0-9]{20,}|sk-[A-Za-z0-9_-]{20,}|\b[0-9a-f]{40}\b'
    r'|(?P<kv>(?:token|password|passwd|secret|api[_-]?key)\s*[=:]\s*)\S+'
    r'|(?P<flag>--(?:token|password)\s+)\S+|(?P<auth>Authorization:\s*\S+\s+)\S+'
)


def redact(s):
    def sub(m):
        keep = m.group("kv") or m.group("flag") or m.group("auth") or ""
        return keep + "<redacted>"
    return RE_SECRET.sub(sub, s)


def clip(s, n, tail=False):
    """redact, затем обрезка: секрет на границе обрезки не утекает частично."""
    s = redact(s or "")
    return s[-n:] if tail else s[:n]


def hook_of(txt, tool):
    if "[guard-main]" in txt:
        return "guard-main(Bash)" if tool == "Bash" else "guard-main(Edit/Write)"
    if "[require-plan]" in txt:
        return "require-plan"
    if "shell write" in txt:  # формулировки до 0.4.0
        return "guard-main(Bash)"
    if "you are on" in txt:
        return "guard-main(Edit/Write)"
    if "without a plan" in txt:
        return "require-plan"
    if "commit/push в main" in txt or "ветке main" in txt or "reference-old-arm" in txt or "docs/source" in txt:
        return "workspace-guard"
    return "other"


def _records(path):
    for line in open(path, encoding="utf-8", errors="replace"):
        try:
            j = json.loads(line)
        except ValueError:
            continue
        if isinstance(j, dict):
            yield j


def _blocks(path, sid, since):
    """Блокировки одного транскрипта; с since — только вызовы с меткой времени не раньше даты."""
    pending = {}  # tool_use_id -> (name, input, cwd, branch, ts)
    for j in _records(path):
        m = j.get("message") or {}
        cont = m.get("content")
        if j.get("type") == "assistant":
            for b in cont or []:
                if isinstance(b, dict) and b.get("type") == "tool_use":
                    pending[b.get("id")] = (b.get("name"), b.get("input") or {}, j.get("cwd", ""),
                                            j.get("gitBranch", ""), j.get("timestamp", ""))
        elif j.get("type") == "user" and isinstance(cont, list):
            for b in cont:
                if not (isinstance(b, dict) and b.get("type") == "tool_result"):
                    continue
                txt = b.get("content", "")
                if isinstance(txt, list):
                    txt = "\n".join(x.get("text", "") for x in txt if isinstance(x, dict))
                if not isinstance(txt, str) or not HOOK_ERROR.match(txt):
                    continue  # обычный вывод команды, даже если в нём есть слово BLOCKED
                name, inp, cwd, branch, ts = pending.get(b.get("tool_use_id"), ("?", {}, "", "", ""))
                if name not in ("Bash", "Edit", "Write", "MultiEdit"):
                    continue
                if since and (ts or "")[:10] < since:
                    continue
                msg = next((l for l in txt.splitlines() if "BLOCKED" in l), "")
                yield {
                    "session": sid, "ts": ts, "cwd": cwd, "branch": branch, "tool": name,
                    "command": redact(inp.get("command", "")) if name == "Bash" else None,
                    "file_path": inp.get("file_path") if name != "Bash" else None,
                    "hook": hook_of(txt, name), "message": clip(msg, 300), "expected": None,
                }


def export(dirs, since=None, prefix=None):
    """Записи выгрузки по каталогам транскриптов: сессия, затем её сабагенты (<сессия>/subagents/*.jsonl)."""
    n = 0
    for d in dirs:
        for f in sorted(glob.glob(os.path.join(os.path.expanduser(d), "*.jsonl"))):
            sid = os.path.basename(f)[:8]
            subs = sorted(glob.glob(os.path.join(f[:-len(".jsonl")], "subagents", "*.jsonl")))
            for path in [f] + subs:
                for rec in _blocks(path, sid, since):
                    if prefix:
                        n += 1
                        rec = dict({"id": "%s-%03d" % (prefix, n)}, **rec)
                    yield rec


def main(argv):
    args, dirs, since, prefix = argv[1:], [], None, None
    i = 0
    while i < len(args):
        if args[i] == "--since" and i + 1 < len(args):
            since, i = args[i + 1], i + 2
        elif args[i] == "--id-prefix" and i + 1 < len(args):
            prefix, i = args[i + 1], i + 2
        else:
            dirs.append(args[i])
            i += 1
    for rec in export(dirs, since, prefix or since):
        print(json.dumps(rec, ensure_ascii=False))
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))