nash/corpus/extract_host_corpus.py

#!/usr/bin/env python3
"""Extract the real `host_exec` command corpus from session transcripts (docs/NASH.md §10.6).

Sweeps a directory tree of transcript captures (JSON or JSONL — the Claude stream-json capture
files and session transcript stores both work) for `host_exec` tool calls, dedupes the command
strings, auto-classifies each as replayable or `excluded:<reason>` (network / host-mutating /
signing / toolchain — exclusions are explicit, never silent), and appends NEW entries to the
host corpus, preserving anything already there (including the self-test entries the harness
requires).

Usage: extract_host_corpus.py TRANSCRIPTS_DIR [--corpus PATH] [--dry-run]
"""

import argparse
import json
import os
import re
import sys

EXCLUSION_PATTERNS = [
    # (reason, regex over the whole command) — first hit wins; order: most-specific first.
    ("network", r"\b(curl|wget|git\s+(push|pull|fetch|clone)|gh\s|ssh\s|scp\s|rsync\s.*:)"),
    ("mutates-host", r"\b(brew\s+(install|upgrade|uninstall)|sudo\s|defaults\s+write|killall\s"
                     r"|launchctl\s|xcode-select\s+--install)"),
    ("signing-identity", r"\b(codesign|notarytool|productsign|security\s+import)"),
    # Toolchain builds replay only against a real tree (--worktree), not the synthetic fixtures.
    ("toolchain — replay with --worktree against a real package",
     r"\b(xcodebuild|swift\s+(build|test|run)|xcrun\s|make\b)"),
]


def classify(cmd):
    for reason, pattern in EXCLUSION_PATTERNS:
        if re.search(pattern, cmd):
            return f"excluded:{reason}"
    return "replayable"


def walk_json(node, found):
    """Recursively collect host_exec tool-call commands from any JSON shape."""
    if isinstance(node, dict):
        name = node.get("name", "")
        if isinstance(name, str) and name.endswith("host_exec"):
            command = (node.get("input") or {}).get("command")
            if isinstance(command, str) and command.strip():
                found.append(command.strip())
        for value in node.values():
            walk_json(value, found)
    elif isinstance(node, list):
        for value in node:
            walk_json(value, found)


def commands_in_file(path):
    found = []
    try:
        with open(path, encoding="utf-8", errors="replace") as f:
            text = f.read()
    except OSError:
        return found
    # Whole-file JSON first; else JSONL line by line.
    try:
        walk_json(json.loads(text), found)
        return found
    except ValueError:
        pass
    for line in text.splitlines():
        line = line.strip()
        if not line or "host_exec" not in line:
            continue
        try:
            walk_json(json.loads(line), found)
        except ValueError:
            continue
    return found


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("transcripts_dir")
    ap.add_argument(
        "--corpus",
        default=os.path.join(
            os.path.dirname(os.path.abspath(__file__)), "host-corpus.jsonl"),
    )
    ap.add_argument("--dry-run", action="store_true")
    args = ap.parse_args()

    commands = []
    for root, _dirs, files in os.walk(args.transcripts_dir):
        for name in files:
            if name.endswith((".json", ".jsonl")):
                commands.extend(commands_in_file(os.path.join(root, name)))

    existing_cmds = set()
    next_index = 1
    if os.path.exists(args.corpus):
        with open(args.corpus) as f:
            for line in f:
                line = line.strip()
                if line:
                    entry = json.loads(line)
                    existing_cmds.add(entry["cmd"])
                    match = re.match(r"hx-(\d+)$", entry["id"])
                    if match:
                        next_index = max(next_index, int(match.group(1)) + 1)

    new_entries = []
    seen = set()
    for cmd in commands:
        if cmd in existing_cmds or cmd in seen:
            continue
        seen.add(cmd)
        entry = {"id": f"hx-{next_index}", "cmd": cmd}
        next_index += 1
        klass = classify(cmd)
        if klass != "replayable":
            entry["class"] = klass
        new_entries.append(entry)

    print(f"transcripts: {args.transcripts_dir}")
    print(f"host_exec commands found: {len(commands)}  unique-new: {len(new_entries)}")
    for entry in new_entries:
        tag = entry.get("class", "replayable")
        print(f"  [{tag}] {entry['id']}: {entry['cmd']!r}")
    if args.dry_run or not new_entries:
        return
    with open(args.corpus, "a") as f:
        for entry in new_entries:
            f.write(json.dumps(entry) + "\n")
    print(f"appended {len(new_entries)} entries -> {args.corpus}")


if __name__ == "__main__":
    main()