agentdiff 0.1.16

Audit and trace autonomous AI code contributions in git repositories
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131

#!/usr/bin/env python3
"""
AgentDiff capture script for VS Code GitHub Copilot.
Receives events from the agentdiff-copilot VS Code extension via stdin.
Writes to <repo>/.git/agentdiff/session.jsonl.
"""
import os
import sys
import json
import subprocess
from datetime import datetime, timezone


def debug_enabled() -> bool:
    return os.environ.get("AGENTDIFF_DEBUG", "").lower() in {"1", "true", "yes", "on"}


def debug_log(message: str) -> None:
    if not debug_enabled():
        return
    log_dir = os.path.expanduser("~/.agentdiff/logs")
    os.makedirs(log_dir, exist_ok=True)
    path = os.path.join(log_dir, "capture-copilot.log")
    ts = datetime.now(timezone.utc).isoformat()
    with open(path, "a", encoding="utf-8") as f:
        f.write(f"{ts} {message}\n")


def find_repo_root(cwd: str) -> str:
    try:
        result = subprocess.run(
            ["git", "rev-parse", "--show-toplevel"],
            capture_output=True, text=True, cwd=cwd
        )
        return result.stdout.strip() if result.returncode == 0 else cwd
    except Exception:
        return cwd


def is_git_repo(path: str) -> bool:
    return bool(path) and os.path.exists(os.path.join(path, ".git"))


def get_session_log(cwd: str) -> str:
    override = os.environ.get("AGENTDIFF_SESSION_LOG")
    if override:
        parent = os.path.dirname(override)
        if parent:
            os.makedirs(parent, exist_ok=True)
        return override

    repo_root = find_repo_root(cwd)
    if is_git_repo(repo_root):
        base = os.path.join(repo_root, ".git", "agentdiff")
        os.makedirs(base, exist_ok=True)
        return os.path.join(base, "session.jsonl")

    spill_root = os.environ.get(
        "AGENTDIFF_SPILLOVER", os.path.expanduser("~/.agentdiff/spillover")
    )
    os.makedirs(spill_root, exist_ok=True)
    slug = cwd.replace("/", "-").replace("\\", "-") or "unknown"
    return os.path.join(spill_root, f"{slug}.jsonl")


def main():
    input_data = sys.stdin.read()
    if not input_data.strip():
        sys.exit(0)
    debug_log(f"raw={input_data[:2000]}")

    try:
        payload = json.loads(input_data)
    except json.JSONDecodeError:
        sys.exit(0)

    abs_file = payload.get("file_path") or ""
    if not abs_file:
        debug_log("skip: missing file_path")
        sys.exit(0)

    cwd = payload.get("cwd") or os.path.dirname(abs_file) or os.getcwd()
    repo_root = find_repo_root(cwd)
    in_repo = is_git_repo(repo_root)

    # Resolve relative file path within repo
    if in_repo:
        try:
            rel_file = os.path.relpath(abs_file, repo_root)
        except ValueError:
            rel_file = abs_file
    else:
        rel_file = abs_file

    event = payload.get("event", "inline")
    tool_map = {
        "inline":      "copilot-inline",
        "save":        "copilot-save",
        "chat_edit":   "copilot-chat",
        "manual":      "copilot-manual",
    }
    tool = tool_map.get(event, f"copilot-{event}")

    lines = payload.get("lines") or []
    if not isinstance(lines, list):
        lines = []

    entry = {
        "timestamp": datetime.now(timezone.utc).isoformat(),
        "agent": "copilot",
        "model": payload.get("model") or "copilot-unknown",
        "session_id": payload.get("session_id") or "unknown",
        "tool": tool,
        "file": rel_file,
        "abs_file": abs_file,
        "prompt": payload.get("prompt"),
        "acceptance": "verbatim",
        "lines": lines,
    }

    session_log = get_session_log(cwd)
    with open(session_log, "a", encoding="utf-8") as f:
        f.write(json.dumps(entry) + "\n")
    debug_log(
        f"wrote entry tool={tool} file={entry['file']} lines={entry.get('lines')} "
        f"repo_root={repo_root!r} session_log={session_log!r}"
    )


if __name__ == "__main__":
    main()