Merge nucleic/dapper-feather-lynx-w4wa into dev

2026-07-22 04:47:03 -07:00
parent b13b39e729
commit 7443a7f431
3 changed files with 367 additions and 0 deletions
@@ -0,0 +1,135 @@
+#!/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()