Gabriel Vidal
9258267fbb
Replays the transcript's accumulated Write/Edit/MultiEdit toolUseResults (and Bash rm deletes) into per-file before/after content, cut off at each repo's last conversation commit, and serves it as /api/conversation-uncommitted-diff (same DiffResult model, stats variant for the panel). The detail panel's Commits card links it above the commit rows; /diff/uncommitted/<id> renders it in the shared DiffView. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
375 lines
16 KiB
Python
375 lines
16 KiB
Python
"""Uncommitted-changes diff, reconstructed from a conversation's transcript.
|
|
|
|
The container can't see the conversation's working tree (worktrees live outside
|
|
the mounted repos), so the "current changes, not committed yet" view is rebuilt
|
|
purely from the transcript: every Write/Edit/MultiEdit ``toolUseResult`` carries
|
|
the target path, the full new content (writes), the exact replacement strings
|
|
(edits) and a ``structuredPatch``; ``rm`` commands in Bash calls mark deletes.
|
|
Replaying those in timestamp order yields each file's before/after content. The
|
|
conversation's commits (per repo, from ``gitdiff.conversation_commits``) set a
|
|
cutoff: events at or before a repo's last commit count as committed, and the
|
|
baseline for the diff is the file's replayed content at that cutoff — so the
|
|
result approximates ``git diff HEAD`` in the (possibly gone) working tree.
|
|
|
|
When a file's full content can't be reconstructed (its first sighting is an
|
|
Edit without ``originalFile``), we fall back to showing its post-cutoff
|
|
``structuredPatch`` hunks verbatim — approximate, but honest per-edit diffs.
|
|
|
|
Only paths inside the homelab checkout (or one of its worktrees) are considered:
|
|
edits under the untracked root ``data/`` tree, ``node_modules``, ``/tmp`` or the
|
|
user's home dot-dirs can never be committed, so they'd pollute the view forever.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import difflib
|
|
import json
|
|
import os.path
|
|
import re
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
from gitdiff import _parse_unified_diff
|
|
|
|
# A path inside the homelab checkout or any of its worktrees/clones
|
|
# (`/home/<u>/homelab/…`, `/home/<u>/homelab-<wt>/…`) → repo-relative tail.
|
|
_REPO_PATH_RE = re.compile(r"^/home/[^/]+/homelab[^/]*/(.+)$")
|
|
_PROJECT_RE = re.compile(r"^projects/([A-Za-z0-9._][\w.-]*)(?:/|$)")
|
|
_SERVICE_RE = re.compile(r"^services/([A-Za-z0-9._][\w.-]*)(?:/|$)")
|
|
# Repo-relative prefixes that are never committed (untracked/generated).
|
|
_SKIP_REL_RE = re.compile(r"^(?:data/|\.env)|(?:^|/)node_modules/")
|
|
|
|
# `rm …` at a command position (start, or right after a shell separator) — the
|
|
# same anchoring trick conversations.py uses for new-worktree.sh, so prose or
|
|
# grep patterns *about* rm don't register as deletes.
|
|
_RM_RE = re.compile(r"(?:^|[\n;&|(]|\b(?:sudo|command)[ \t])[ \t]*rm[ \t]+([^\n|&;<>]*)")
|
|
_GLOBBY = set("*?[]{}$`\"'\\")
|
|
|
|
|
|
def _ts(iso: str | None) -> datetime | None:
|
|
if not iso:
|
|
return None
|
|
try:
|
|
return datetime.fromisoformat(iso.replace("Z", "+00:00"))
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def _rel_and_token(path: str) -> tuple[str, str] | None:
|
|
"""Map an absolute edited path → (repo-relative path, repo token), or None
|
|
when it lives outside the homelab checkout / is never-committed noise."""
|
|
m = _REPO_PATH_RE.match(path or "")
|
|
if not m:
|
|
return None
|
|
rel = m.group(1)
|
|
if _SKIP_REL_RE.search(rel):
|
|
return None
|
|
pm = _PROJECT_RE.match(rel)
|
|
if pm:
|
|
return rel, f"project:{pm.group(1)}"
|
|
sm = _SERVICE_RE.match(rel)
|
|
if sm:
|
|
return rel, f"service:{sm.group(1)}"
|
|
return rel, "super:_"
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# transcript → chronological edit events
|
|
# --------------------------------------------------------------------------- #
|
|
def _rm_paths(command: str, cwd: str | None) -> list[str]:
|
|
"""Absolute paths an `rm` in a Bash command plausibly deleted."""
|
|
out: list[str] = []
|
|
for m in _RM_RE.finditer(command or ""):
|
|
for tok in m.group(1).split():
|
|
if tok.startswith("-"):
|
|
continue
|
|
if any(c in _GLOBBY for c in tok):
|
|
continue
|
|
if not tok.startswith("/"):
|
|
if not cwd:
|
|
continue
|
|
tok = os.path.join(cwd, tok)
|
|
out.append(os.path.normpath(tok))
|
|
return out
|
|
|
|
|
|
def _events_from_file(path: Path) -> list[dict]:
|
|
"""File-mutation events (edit/write/delete) mined from one JSONL transcript."""
|
|
events: list[dict] = []
|
|
last_ts = ""
|
|
try:
|
|
fh = path.open(encoding="utf-8", errors="replace")
|
|
except OSError:
|
|
return events
|
|
with fh:
|
|
for line in fh:
|
|
has_patch = '"structuredPatch"' in line
|
|
has_rm = '"Bash"' in line and "rm " in line
|
|
if not has_patch and not has_rm:
|
|
continue
|
|
try:
|
|
o = json.loads(line)
|
|
except (json.JSONDecodeError, UnicodeDecodeError):
|
|
continue
|
|
ts = o.get("timestamp") or last_ts
|
|
last_ts = ts
|
|
tr = o.get("toolUseResult")
|
|
if has_patch and isinstance(tr, dict) and tr.get("filePath"):
|
|
ev = {"ts": ts, "file": str(tr["filePath"]),
|
|
"patch": tr.get("structuredPatch") or [],
|
|
"orig": tr.get("originalFile")}
|
|
if "oldString" in tr:
|
|
ev.update(kind="edit",
|
|
edits=[(tr.get("oldString") or "",
|
|
tr.get("newString") or "",
|
|
bool(tr.get("replaceAll")))])
|
|
elif isinstance(tr.get("edits"), list): # MultiEdit
|
|
ev.update(kind="edit", edits=[
|
|
(e.get("oldString") or e.get("old_string") or "",
|
|
e.get("newString") or e.get("new_string") or "",
|
|
bool(e.get("replaceAll") or e.get("replace_all")))
|
|
for e in tr["edits"] if isinstance(e, dict)])
|
|
elif tr.get("type") in ("create", "update"):
|
|
ev.update(kind="write", content=tr.get("content"),
|
|
create=tr.get("type") == "create")
|
|
else:
|
|
ev.update(kind="patch")
|
|
events.append(ev)
|
|
continue
|
|
if has_rm:
|
|
msg = o.get("message") or {}
|
|
content = msg.get("content")
|
|
if not isinstance(content, list):
|
|
continue
|
|
for blk in content:
|
|
if not (isinstance(blk, dict) and blk.get("type") == "tool_use"
|
|
and blk.get("name") == "Bash"):
|
|
continue
|
|
cmd = (blk.get("input") or {}).get("command") or ""
|
|
for p in _rm_paths(cmd, o.get("cwd")):
|
|
events.append({"ts": ts, "file": p, "kind": "delete"})
|
|
return events
|
|
|
|
|
|
def _all_events(transcript: Path) -> list[dict]:
|
|
"""The conversation's events plus its subagents', merged chronologically."""
|
|
events = _events_from_file(transcript)
|
|
subdir = transcript.parent / transcript.stem / "subagents"
|
|
if subdir.is_dir():
|
|
for sp in sorted(subdir.glob("agent-*.jsonl")):
|
|
events += _events_from_file(sp)
|
|
events.sort(key=lambda e: e["ts"] or "")
|
|
return events
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# replay → per-file before/after
|
|
# --------------------------------------------------------------------------- #
|
|
def _apply_edits(content: str, edits: list[tuple[str, str, bool]]) -> str | None:
|
|
"""Apply Edit-tool replacements; None when an oldString no longer matches
|
|
(external/user modification — full content is no longer trustworthy)."""
|
|
for old, new, replace_all in edits:
|
|
if not old:
|
|
return None
|
|
if old not in content:
|
|
return None
|
|
content = content.replace(old, new) if replace_all \
|
|
else content.replace(old, new, 1)
|
|
return content
|
|
|
|
|
|
def _replay_file(events: list[dict], cutoff: datetime | None) -> dict | None:
|
|
"""Replay one file's events; snapshot the baseline at the commit cutoff.
|
|
|
|
Returns ``{baseline: (known, content, exists), final: (…), patches: […]}``
|
|
or None when every event predates the cutoff (i.e. it's all committed).
|
|
"""
|
|
known, content, exists = False, None, None # exists=None → never seen yet
|
|
baseline = None
|
|
patches: list = []
|
|
|
|
for e in events:
|
|
et = _ts(e["ts"])
|
|
is_post = cutoff is None or et is None or et > cutoff
|
|
if is_post and baseline is None:
|
|
b_known, b_content, b_exists = known, content, exists
|
|
if b_exists is None: # first sighting is post-cutoff → infer prior
|
|
b_exists = not (e["kind"] == "write" and e.get("create"))
|
|
orig = e.get("orig")
|
|
if not b_exists:
|
|
b_known, b_content = True, None
|
|
elif isinstance(orig, str):
|
|
b_known, b_content = True, orig
|
|
baseline = (b_known, b_content, b_exists)
|
|
kind = e["kind"]
|
|
if kind == "delete":
|
|
known, content, exists = True, None, False
|
|
continue
|
|
if kind == "write":
|
|
body = e.get("content")
|
|
known = isinstance(body, str)
|
|
content = body if known else None
|
|
exists = True
|
|
elif kind == "edit":
|
|
base = content if known and content is not None else (
|
|
e["orig"] if isinstance(e.get("orig"), str) else None)
|
|
content = _apply_edits(base, e["edits"]) if base is not None else None
|
|
known = content is not None
|
|
exists = True
|
|
else: # bare patch (NotebookEdit & co) — content can't be replayed
|
|
known, content, exists = False, None, True
|
|
if is_post and e.get("patch"):
|
|
patches.append(e["patch"])
|
|
|
|
if baseline is None:
|
|
return None
|
|
return {"baseline": baseline, "final": (known, content, exists),
|
|
"patches": patches}
|
|
|
|
|
|
# --------------------------------------------------------------------------- #
|
|
# before/after → the DiffResult file model
|
|
# --------------------------------------------------------------------------- #
|
|
def _unified_chunk(rel: str, status: str, a: str | None, b: str | None) -> str | None:
|
|
a_lines = a.splitlines() if a is not None else []
|
|
b_lines = b.splitlines() if b is not None else []
|
|
body = list(difflib.unified_diff(
|
|
a_lines, b_lines,
|
|
fromfile="/dev/null" if a is None else f"a/{rel}",
|
|
tofile="/dev/null" if b is None else f"b/{rel}",
|
|
lineterm=""))
|
|
if not body:
|
|
return None
|
|
head = [f"diff --git a/{rel} b/{rel}"]
|
|
if status == "added":
|
|
head.append("new file mode 100644")
|
|
elif status == "deleted":
|
|
head.append("deleted file mode 100644")
|
|
return "\n".join(head + body)
|
|
|
|
|
|
def _patch_file(rel: str, status: str, patches: list) -> dict | None:
|
|
"""Fallback file model: the raw post-cutoff structuredPatch hunks, verbatim.
|
|
Successive hunks describe successive intermediate versions, not one net
|
|
diff — the whole response is flagged ``approx`` for this reason."""
|
|
hunks: list[dict] = []
|
|
add = dele = 0
|
|
for sp in patches:
|
|
for h in sp if isinstance(sp, list) else []:
|
|
try:
|
|
old_no, new_no = int(h["oldStart"]), int(h["newStart"])
|
|
header = (f"@@ -{h['oldStart']},{h['oldLines']}"
|
|
f" +{h['newStart']},{h['newLines']} @@")
|
|
raw_lines = h.get("lines") or []
|
|
except (KeyError, TypeError, ValueError):
|
|
continue
|
|
lines: list[dict] = []
|
|
hs, hn = old_no, new_no
|
|
for raw in raw_lines:
|
|
raw = raw if isinstance(raw, str) else str(raw)
|
|
tag, body = raw[:1], raw[1:]
|
|
if tag == "+":
|
|
lines.append({"type": "add", "oldNo": None, "newNo": hn, "text": body})
|
|
hn += 1
|
|
add += 1
|
|
elif tag == "-":
|
|
lines.append({"type": "del", "oldNo": hs, "newNo": None, "text": body})
|
|
hs += 1
|
|
dele += 1
|
|
else:
|
|
lines.append({"type": "ctx", "oldNo": hs, "newNo": hn, "text": body})
|
|
hs += 1
|
|
hn += 1
|
|
hunks.append({"header": header,
|
|
"oldStart": old_no, "oldLines": int(h["oldLines"]),
|
|
"newStart": new_no, "newLines": int(h["newLines"]),
|
|
"lines": lines})
|
|
if not hunks:
|
|
return None
|
|
return {"path": rel, "oldPath": None, "status": status,
|
|
"additions": add, "deletions": dele, "binary": False,
|
|
"truncated": False, "hunks": hunks}
|
|
|
|
|
|
def uncommitted_diff(transcript: Path, title: str, commits: list[dict]) -> dict:
|
|
"""The conversation's net not-yet-committed diff, as a DiffResult dict."""
|
|
# Per-repo cutoff = its newest conversation commit; edits after it are the
|
|
# uncommitted tail. A repo with no commits has no cutoff (all uncommitted).
|
|
cutoffs: dict[str, datetime] = {}
|
|
for c in commits or []:
|
|
dt = _ts(c.get("date"))
|
|
if dt and (c.get("repo") not in cutoffs or dt > cutoffs[c["repo"]]):
|
|
cutoffs[c["repo"]] = dt
|
|
|
|
by_file: dict[str, tuple[str, list[dict]]] = {}
|
|
for e in _all_events(transcript):
|
|
if e["kind"] == "delete":
|
|
# `rm dir` (or a worktree teardown) may cover several tracked files.
|
|
hit = False
|
|
for path, (rel, evs) in by_file.items():
|
|
if path == e["file"] or path.startswith(e["file"] + "/"):
|
|
evs.append(dict(e, file=path))
|
|
hit = True
|
|
if hit:
|
|
continue
|
|
rt = _rel_and_token(e["file"])
|
|
if not rt:
|
|
continue
|
|
by_file.setdefault(e["file"], (rt[0], []))[1].append(e)
|
|
|
|
unified_chunks: list[str] = []
|
|
chunk_tokens: list[str] = []
|
|
patch_files: list[dict] = []
|
|
for path, (rel, evs) in sorted(by_file.items(), key=lambda kv: kv[1][0]):
|
|
token = _rel_and_token(path)[1] # type: ignore[index]
|
|
st = _replay_file(evs, cutoffs.get(token))
|
|
if st is None:
|
|
continue # fully committed
|
|
(b_known, b_content, b_exists) = st["baseline"]
|
|
(f_known, f_content, f_exists) = st["final"]
|
|
if f_exists is False: # deleted at the end
|
|
if b_exists is False:
|
|
continue # created and deleted after the cutoff — net nothing
|
|
if b_known and b_content is not None:
|
|
chunk = _unified_chunk(rel, "deleted", b_content, None)
|
|
if chunk:
|
|
unified_chunks.append(chunk)
|
|
chunk_tokens.append(token)
|
|
else:
|
|
patch_files.append({
|
|
"path": rel, "oldPath": None, "status": "deleted",
|
|
"additions": 0, "deletions": 0, "binary": False,
|
|
"truncated": False, "hunks": [], "repo": token})
|
|
continue
|
|
status = "added" if b_exists is False else "modified"
|
|
if b_known and f_known:
|
|
if b_exists is not False and (b_content or "") == (f_content or ""):
|
|
continue # touched, but net-identical to the committed state
|
|
chunk = _unified_chunk(
|
|
rel, status,
|
|
None if b_exists is False else (b_content or ""),
|
|
f_content or "")
|
|
if chunk:
|
|
unified_chunks.append(chunk)
|
|
chunk_tokens.append(token)
|
|
continue
|
|
pf = _patch_file(rel, status, st["patches"])
|
|
if pf:
|
|
pf["repo"] = token
|
|
patch_files.append(pf)
|
|
|
|
files = _parse_unified_diff("\n".join(unified_chunks))
|
|
for f, token in zip(files, chunk_tokens):
|
|
f["repo"] = token
|
|
files += patch_files
|
|
files.sort(key=lambda f: f["path"])
|
|
add = sum(f["additions"] for f in files)
|
|
dele = sum(f["deletions"] for f in files)
|
|
n = len(files)
|
|
return {
|
|
"title": title or "Uncommitted changes",
|
|
"subtitle": (f"uncommitted · {n} file{'s' if n != 1 else ''} · "
|
|
"replayed from the transcript's edits"),
|
|
"files": files, "additions": add, "deletions": dele,
|
|
"commits": None, "approx": True,
|
|
}
|