W4118

collect_llm_logs.py

Download, or run it from your team repo:

curl -O https://w4118.github.io/tools/collect_llm_logs.py
python3 collect_llm_logs.py abc1234
#!/usr/bin/env python3
"""Copy your coding-agent transcripts into your homework repo.

    python3 collect_llm_logs.py abc1234
    python3 collect_llm_logs.py abc1234 --repo /home/abc1234/f26-hmwk4-team5

Run it inside your team repo, on every machine where you used a coding agent
(your VM, your laptop, or both). Replace abc1234 with your UNI.

What it does:
  1. Looks in the places where coding agents (Claude Code, Codex, Google Antigravity,
     Cursor, GitHub Copilot CLI, VS Code Copilot Chat, Cline/Roo Code)
     save their transcripts on this machine.
  2. Picks the sessions that ran inside your repo (or in a folder below it).
  3. Copies those files, unchanged, into llm_logs/<your UNI>/<tool>/ in your repo.

Changelog:
  v1  2026-10-18  Initial version.
"""

import argparse
import json
import os
import re
import shutil
import subprocess
import sys
from pathlib import Path
from urllib.parse import unquote, urlparse

HOME = Path.home()
GITHUB_LIMIT = 95 * 1024 * 1024  # GitHub rejects files over 100 MB


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------

def norm(path):
    """A path as a comparable string: forward slashes, no trailing slash,
    lowercase on Windows (where paths are case-insensitive)."""
    s = str(path).replace("\\", "/").rstrip("/")
    return s.lower() if os.name == "nt" else s


def in_repo(path, repos):
    """True if `path` is one of `repos` or a folder below one of them.
    (A plain prefix test would also match team5-old or team50.)"""
    p = norm(path)
    return any(p == r or p.startswith(r + "/") for r in repos)


def cwd_in_jsonl(file, max_lines=50):
    """The first "cwd" value found in the first lines of a JSONL transcript
    (Claude Code, Codex and Copilot CLI record the working directory there)."""
    def find(obj):
        if isinstance(obj, dict):
            if isinstance(obj.get("cwd"), str):
                return obj["cwd"]
            for v in obj.values():
                found = find(v)
                if found:
                    return found
        return None
    lines = read_text(file).splitlines()[:max_lines]
    for line in lines:
        try:
            found = find(json.loads(line))
        except ValueError:
            continue
        if found:
            return found
    return None


def read_text(file):
    """File contents as text; .zst files (old Codex sessions) are decompressed
    in memory only, the file on disk stays as it is."""
    file = Path(file)
    if file.suffix != ".zst":
        return file.read_text(errors="replace")
    try:
        from compression import zstd  # Python 3.14+
        return zstd.decompress(file.read_bytes()).decode(errors="replace")
    except ImportError:
        pass
    try:
        out = subprocess.run(["zstd", "-dcq", str(file)], capture_output=True, check=True)
        return out.stdout.decode(errors="replace")
    except (OSError, subprocess.CalledProcessError):
        print(f"  WARNING: cannot read {file} (install zstd: sudo apt install zstd, "
              "or brew install zstd), then run this script again")
        return ""


def copy(src, dest_dir, name=None):
    """Copy one file or folder unchanged (contents and timestamps)."""
    dest = Path(dest_dir) / (name or Path(src).name)
    dest.parent.mkdir(parents=True, exist_ok=True)
    if Path(src).is_dir():
        shutil.copytree(src, dest, dirs_exist_ok=True)
    else:
        shutil.copy2(src, dest)
    return dest


def env_dir(var, default):
    return Path(os.environ[var]) if os.environ.get(var) else default


# ---------------------------------------------------------------------------
# One function per tool. Each returns how many sessions it copied.
# ---------------------------------------------------------------------------

def claude_code(repos, out):
    """Claude Code (CLI and IDE extensions): ~/.claude/projects/<folder>/<session>.jsonl.
    A session may have a folder of the same name with subagent transcripts."""
    root = env_dir("CLAUDE_CONFIG_DIR", HOME / ".claude") / "projects"
    n = 0
    for f in sorted(root.glob("*/*.jsonl")):
        cwd = cwd_in_jsonl(f)
        if cwd and in_repo(cwd, repos):
            copy(f, out / f.parent.name)
            extra = f.with_suffix("")  # <session>/ with subagents/, tool-results/
            if extra.is_dir():
                copy(extra, out / f.parent.name)
            n += 1
    return n


def codex(repos, out):
    """OpenAI Codex (CLI, desktop app, IDE extension):
    ~/.codex/sessions/YYYY/MM/DD/rollout-*.jsonl and ~/.codex/archived_sessions/.
    Sessions older than a few days are compressed (.jsonl.zst); they are copied as is."""
    root = env_dir("CODEX_HOME", HOME / ".codex")
    n = 0
    for sub in ("sessions", "archived_sessions"):
        for f in sorted((root / sub).rglob("rollout-*")):
            cwd = cwd_in_jsonl(f)
            if cwd and in_repo(cwd, repos):
                copy(f, out)
                n += 1
    return n


def mentions_repo(text, repos):
    """True if `text` contains the repo path followed by a quote, '/', whitespace
    or the end (so team5 does not match team50 or team5-old)."""
    text = norm(text)
    return any(re.search(re.escape(r) + r'(?=["/\s]|$)', text) for r in repos)


def antigravity(repos, out):
    """Google Antigravity (replaces Gemini CLI).
    CLI (agy): ~/.gemini/antigravity-cli/brain/<conversation>/.system_generated/logs/
    with transcript_full.jsonl (every step, untruncated) and transcript.jsonl.
    The transcript does not record the folder agy ran in, so a conversation counts
    as this repo's if it mentions the repo path (agy works with absolute paths).
    Editor: no plain message log. The conversation itself is a binary file
    (~/.gemini/antigravity/conversations/<conversation>.pb or .db), and brain/<conversation>/
    holds its plans and walkthroughs. Both are copied as they are when the
    brain folder mentions the repo."""
    gem = HOME / ".gemini"
    n = 0
    for logs in sorted((gem / "antigravity-cli" / "brain").glob("*/.system_generated/logs")):
        files = sorted(logs.glob("*.jsonl"))
        if mentions_repo("\n".join(f.read_text(errors="replace") for f in files), repos):
            conv = logs.parent.parent.name
            for f in files:
                copy(f, out / "cli" / conv)
            n += 1
    for store in ("antigravity", "antigravity-ide"):
        for brain in sorted((gem / store / "brain").glob("*")):
            if not brain.is_dir():
                continue
            text = "\n".join(f.read_text(errors="replace") for f in brain.rglob("*")
                              if f.is_file() and f.suffix in (".md", ".json", ".resolved", ""))
            if not mentions_repo(text, repos):
                continue
            copy(brain, out / "editor" / brain.name, "brain")
            for f in (gem / store / "conversations").glob(brain.name + ".*"):
                copy(f, out / "editor" / brain.name)
            n += 1
    return n


def cursor(repos, out):
    """Cursor agent: ~/.cursor/projects/<folder>/agent-transcripts/, where
    <folder> is the workspace path with every non-alphanumeric character
    turned into '-' (on Remote-SSH this is on the VM)."""
    def slug(p):
        return "".join(c if c.isalnum() else "-" for c in norm(p).lstrip("/"))
    wanted = {slug(r) for r in repos}
    n = 0
    for d in sorted((HOME / ".cursor" / "projects").glob("*/agent-transcripts")):
        if slug(d.parent.name) in wanted:
            copy(d, out / d.parent.name)
            n += len([p for p in d.rglob("*") if p.is_file()])
    return n


def copilot_cli(repos, out):
    """GitHub Copilot CLI: ~/.copilot/session-state/<id>/events.jsonl, with the
    session's folder in workspace.yaml ("cwd: /path"). Older versions wrote
    ~/.copilot/session-state/<id>.jsonl."""
    root = env_dir("COPILOT_HOME", HOME / ".copilot") / "session-state"
    n = 0
    for d in sorted(p for p in root.glob("*") if p.is_dir()):
        cwd = None
        ws = d / "workspace.yaml"
        if ws.is_file():
            for line in ws.read_text(errors="replace").splitlines():
                if line.startswith("cwd:"):
                    cwd = line[4:].strip().strip("'\"")
        if cwd is None and (d / "events.jsonl").is_file():
            cwd = cwd_in_jsonl(d / "events.jsonl")
        if cwd and in_repo(cwd, repos):
            copy(d, out)
            n += 1
    for f in sorted(root.glob("*.jsonl")):  # older layout
        cwd = cwd_in_jsonl(f)
        if cwd and in_repo(cwd, repos):
            copy(f, out)
            n += 1
    return n


def vscode_copilot_chat(repos, out):
    """VS Code Copilot Chat: workspaceStorage/<hash>/chatSessions/ (and the
    Copilot extension's own folder), where <hash>/workspace.json names the
    workspace folder. Remote-SSH workspaces are stored on your laptop with the
    VM path in workspace.json: run on the laptop with --repo <path on the VM>."""
    roots = [
        HOME / ".vscode-server" / "data" / "User" / "workspaceStorage",
        HOME / ".config" / "Code" / "User" / "workspaceStorage",
        HOME / "Library" / "Application Support" / "Code" / "User" / "workspaceStorage",
    ]
    if os.environ.get("APPDATA"):
        roots.append(Path(os.environ["APPDATA"]) / "Code" / "User" / "workspaceStorage")
    n = 0
    for root in roots:
        for ws in sorted(root.glob("*/workspace.json")):
            try:
                uri = json.loads(ws.read_text()).get("folder", "")
            except ValueError:
                continue
            path = unquote(urlparse(uri).path)  # works for file:// and vscode-remote://
            if os.name == "nt" and path[:1] == "/" and path[2:3] == ":":
                path = path[1:]  # file:///c:/Users/... -> c:/Users/...
            if not in_repo(path, repos):
                continue
            for sub in ("chatSessions", "GitHub.copilot-chat/transcripts"):  # not the search index
                if (ws.parent / sub).is_dir():
                    copy(ws.parent / sub, out / ws.parent.name)
                    n += len([p for p in (ws.parent / sub).rglob("*") if p.is_file()])
    return n


def cline(repos, out):
    """Cline and Roo Code (VS Code extensions): globalStorage/<extension>/tasks/<id>/*.json.
    A task belongs to the repo if its files mention the repo path. Only the
    .json transcript files are copied, not the checkpoints/ folder."""
    bases = [
        HOME / ".vscode-server" / "data" / "User" / "globalStorage",
        HOME / ".cursor-server" / "data" / "User" / "globalStorage",
        HOME / ".config" / "Code" / "User" / "globalStorage",
        HOME / "Library" / "Application Support" / "Code" / "User" / "globalStorage",
    ]
    if os.environ.get("APPDATA"):
        bases.append(Path(os.environ["APPDATA"]) / "Code" / "User" / "globalStorage")
    n = 0
    for base in bases:
        for ext in ("saoudrizwan.claude-dev", "rooveterinaryinc.roo-cline"):
            for task in sorted((base / ext / "tasks").glob("*")):
                files = sorted(task.glob("*.json"))
                if mentions_repo("\n".join(f.read_text(errors="replace") for f in files), repos):
                    for f in files:
                        copy(f, out / ext / task.name)
                    n += 1
    return n


TOOLS = [
    ("Claude Code", "claude-code", claude_code),
    ("Codex", "codex", codex),
    ("Google Antigravity", "antigravity", antigravity),
    ("Cursor", "cursor", cursor),
    ("GitHub Copilot CLI", "copilot-cli", copilot_cli),
    ("VS Code Copilot Chat", "vscode-copilot", vscode_copilot_chat),
    ("Cline / Roo Code", "cline", cline),
]


# ---------------------------------------------------------------------------
# Main
# ---------------------------------------------------------------------------

def git_toplevel(folder):
    try:
        out = subprocess.run(["git", "-C", str(folder), "rev-parse", "--show-toplevel"],
                             capture_output=True, text=True, check=True)
        return Path(out.stdout.strip())
    except (OSError, subprocess.CalledProcessError):
        return None


def main():
    ap = argparse.ArgumentParser(description="Copy your coding-agent transcripts into llm_logs/<UNI>/.")
    ap.add_argument("uni", help="your UNI, e.g. abc1234")
    ap.add_argument("--repo", help="the folder the agent worked in, if it is not the repo you run this from")
    args = ap.parse_args()

    uni = args.uni.strip().lower()
    if not uni.isalnum() or uni == "abc1234":
        sys.exit("Give your own UNI, e.g.: python3 collect_llm_logs.py abc1234")

    here = git_toplevel(Path.cwd())
    if here is None and not args.repo:
        sys.exit("Run this inside your team repo (or use --repo).")
    target = Path(args.repo).expanduser() if args.repo else here
    # Match both the path as given and its real path (symlinks, /private/tmp on macOS).
    repos = sorted({norm(target), norm(target.resolve())})
    base = (here or Path.cwd()) / "llm_logs" / uni

    print(f"Looking for sessions that ran in {target}")
    total = 0
    for label, folder, collect in TOOLS:
        try:
            n = collect(repos, base / folder)
        except OSError as e:
            print(f"  {label}: error reading its files ({e}); ask on Ed")
            continue
        if n:
            print(f"  {label}: copied {n} into {base / folder}")
            total += n

    if total == 0:
        print("No agent transcripts found on this machine for this repo.\n"
              "If you used an agent here, check that you ran this inside the repo the agent "
              "worked in (or pass --repo), and ask on Ed if it still finds nothing.")
        return
    big = [p for p in base.rglob("*") if p.is_file() and p.stat().st_size > GITHUB_LIMIT]
    for p in big:
        print(f"  {p} is too big for GitHub: run  gzip '{p}'  and commit the .gz file")
    print(f"\nNow commit them:\n  git add -f llm_logs/{uni} && git commit -m 'llm logs' && git push")


if __name__ == "__main__":
    main()