747 lines
29 KiB
Python
747 lines
29 KiB
Python
"""Conversation-scoped hidden git versioning for host workspace."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from dataclasses import dataclass
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
from collections import deque
|
|
|
|
|
|
class VersioningError(RuntimeError):
|
|
"""Raised when hidden git versioning fails."""
|
|
|
|
|
|
@dataclass
|
|
class VersioningPaths:
|
|
save_root: Path
|
|
git_dir: Path
|
|
meta_path: Path
|
|
inputs_path: Path
|
|
snapshots_dir: Path
|
|
|
|
|
|
class ConversationVersioningManager:
|
|
"""Manage hidden git snapshots bound to a conversation."""
|
|
|
|
def __init__(self, project_path: Path | str, data_dir: Path | str, conversation_id: str):
|
|
self.project_path = Path(project_path).expanduser().resolve()
|
|
self.data_dir = Path(data_dir).expanduser().resolve()
|
|
self.conversation_id = str(conversation_id or "").strip()
|
|
if not self.conversation_id:
|
|
raise VersioningError("缺少 conversation_id")
|
|
self.paths = VersioningPaths(
|
|
save_root=(self.data_dir / "save" / self.conversation_id).resolve(),
|
|
git_dir=(self.data_dir / "save" / self.conversation_id / "git").resolve(),
|
|
meta_path=(self.data_dir / "save" / self.conversation_id / "meta.json").resolve(),
|
|
inputs_path=(self.data_dir / "save" / self.conversation_id / "inputs.jsonl").resolve(),
|
|
snapshots_dir=(self.data_dir / "save" / self.conversation_id / "snapshots").resolve(),
|
|
)
|
|
|
|
# ----------------------------
|
|
# low-level helpers
|
|
# ----------------------------
|
|
def _run_git(
|
|
self,
|
|
args: List[str],
|
|
*,
|
|
check: bool = True,
|
|
timeout_seconds: int = 180,
|
|
) -> Tuple[bool, str, str]:
|
|
git_bin = shutil.which("git")
|
|
if not git_bin:
|
|
raise VersioningError("未检测到 git 可执行文件")
|
|
cmd = [
|
|
git_bin,
|
|
"-c",
|
|
"core.quotepath=false",
|
|
f"--git-dir={self.paths.git_dir}",
|
|
f"--work-tree={self.project_path}",
|
|
*args,
|
|
]
|
|
try:
|
|
completed = subprocess.run(
|
|
cmd,
|
|
capture_output=True,
|
|
text=True,
|
|
timeout=timeout_seconds,
|
|
)
|
|
except subprocess.TimeoutExpired as exc:
|
|
raise VersioningError(f"git 执行超时: {' '.join(args)}") from exc
|
|
ok = completed.returncode == 0
|
|
stdout = (completed.stdout or "").strip()
|
|
stderr = (completed.stderr or "").strip()
|
|
if check and not ok:
|
|
raise VersioningError(stderr or stdout or f"git 命令失败: {' '.join(args)}")
|
|
return ok, stdout, stderr
|
|
|
|
def _ensure_exclude_paths(self) -> None:
|
|
"""Avoid recursive tracking if save root is inside work tree."""
|
|
try:
|
|
rel_save = self.paths.save_root.relative_to(self.project_path)
|
|
except ValueError:
|
|
return
|
|
info_exclude = self.paths.git_dir / "info" / "exclude"
|
|
info_exclude.parent.mkdir(parents=True, exist_ok=True)
|
|
existing = ""
|
|
if info_exclude.exists():
|
|
try:
|
|
existing = info_exclude.read_text(encoding="utf-8", errors="ignore")
|
|
except OSError:
|
|
existing = ""
|
|
marker = str(rel_save).replace("\\", "/").rstrip("/")
|
|
if not marker:
|
|
return
|
|
line = f"{marker}/"
|
|
if line not in existing.splitlines():
|
|
with info_exclude.open("a", encoding="utf-8") as fh:
|
|
if existing and not existing.endswith("\n"):
|
|
fh.write("\n")
|
|
fh.write(line + "\n")
|
|
|
|
def _stage_all(self) -> None:
|
|
self._ensure_exclude_paths()
|
|
# Stage everything under work tree; exclusions are handled via info/exclude.
|
|
self._run_git(["add", "-A", "--", "."])
|
|
|
|
def _get_head_commit(self) -> Optional[str]:
|
|
ok, out, _ = self._run_git(["rev-parse", "HEAD"], check=False)
|
|
return out if ok and out else None
|
|
|
|
def _is_clean(self) -> bool:
|
|
ok, out, _ = self._run_git(["status", "--porcelain"], check=False)
|
|
if not ok:
|
|
return False
|
|
return not bool(out.strip())
|
|
|
|
# ----------------------------
|
|
# metadata / records
|
|
# ----------------------------
|
|
def load_meta(self) -> Dict[str, Any]:
|
|
if not self.paths.meta_path.exists():
|
|
return {
|
|
"enabled": False,
|
|
"mode": "overwrite",
|
|
"last_commit": None,
|
|
"last_input_seq": 0,
|
|
"updated_at": None,
|
|
}
|
|
try:
|
|
return json.loads(self.paths.meta_path.read_text(encoding="utf-8")) or {}
|
|
except Exception:
|
|
return {
|
|
"enabled": False,
|
|
"mode": "overwrite",
|
|
"last_commit": None,
|
|
"last_input_seq": 0,
|
|
"updated_at": None,
|
|
}
|
|
|
|
def save_meta(self, meta: Dict[str, Any]) -> Dict[str, Any]:
|
|
payload = dict(meta or {})
|
|
payload["updated_at"] = datetime.now().isoformat()
|
|
self.paths.meta_path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.paths.meta_path.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return payload
|
|
|
|
def _read_inputs(self) -> List[Dict[str, Any]]:
|
|
if not self.paths.inputs_path.exists():
|
|
return []
|
|
rows: List[Dict[str, Any]] = []
|
|
try:
|
|
for raw in self.paths.inputs_path.read_text(encoding="utf-8", errors="ignore").splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
try:
|
|
row = json.loads(line)
|
|
except Exception:
|
|
continue
|
|
if isinstance(row, dict):
|
|
rows.append(self._normalize_checkpoint_row(row))
|
|
except OSError:
|
|
return []
|
|
return rows
|
|
|
|
def _append_input(self, row: Dict[str, Any]) -> None:
|
|
self.paths.inputs_path.parent.mkdir(parents=True, exist_ok=True)
|
|
with self.paths.inputs_path.open("a", encoding="utf-8") as fh:
|
|
fh.write(json.dumps(row, ensure_ascii=False) + "\n")
|
|
|
|
def _write_inputs(self, rows: List[Dict[str, Any]]) -> None:
|
|
self.paths.inputs_path.parent.mkdir(parents=True, exist_ok=True)
|
|
content = "\n".join(json.dumps(row, ensure_ascii=False) for row in (rows or []))
|
|
if content:
|
|
content += "\n"
|
|
self.paths.inputs_path.write_text(content, encoding="utf-8")
|
|
|
|
def _decode_git_path(self, raw_path: Any) -> str:
|
|
text = str(raw_path or "")
|
|
stripped = text.strip()
|
|
if len(stripped) >= 2 and stripped[0] == '"' and stripped[-1] == '"':
|
|
body = stripped[1:-1]
|
|
try:
|
|
decoded = bytes(body, "utf-8").decode("unicode_escape")
|
|
except Exception:
|
|
decoded = body
|
|
try:
|
|
return decoded.encode("latin-1").decode("utf-8")
|
|
except Exception:
|
|
return decoded
|
|
return text
|
|
|
|
def _normalize_checkpoint_row(self, row: Dict[str, Any]) -> Dict[str, Any]:
|
|
payload = dict(row or {})
|
|
files = payload.get("files")
|
|
if isinstance(files, list):
|
|
normalized_files: List[Dict[str, Any]] = []
|
|
for file_item in files:
|
|
if not isinstance(file_item, dict):
|
|
continue
|
|
item = dict(file_item)
|
|
item["path"] = self._decode_git_path(item.get("path"))
|
|
if item.get("old_path") is not None:
|
|
item["old_path"] = self._decode_git_path(item.get("old_path"))
|
|
normalized_files.append(item)
|
|
payload["files"] = normalized_files
|
|
return payload
|
|
|
|
def _write_snapshot(self, seq: int, payload: Dict[str, Any]) -> str:
|
|
self.paths.snapshots_dir.mkdir(parents=True, exist_ok=True)
|
|
target = self.paths.snapshots_dir / f"{int(seq)}.json"
|
|
target.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
return str(target.relative_to(self.paths.save_root))
|
|
|
|
def get_checkpoint_snapshot(self, seq: int) -> Optional[Dict[str, Any]]:
|
|
row = self.get_checkpoint(seq)
|
|
if not row:
|
|
return None
|
|
rel = row.get("snapshot_file")
|
|
if not rel:
|
|
return None
|
|
path = (self.paths.save_root / str(rel)).resolve()
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8")) or None
|
|
except Exception:
|
|
return None
|
|
|
|
# ----------------------------
|
|
# repo lifecycle
|
|
# ----------------------------
|
|
def ensure_repo(self) -> Dict[str, Any]:
|
|
self.paths.save_root.mkdir(parents=True, exist_ok=True)
|
|
self.project_path.mkdir(parents=True, exist_ok=True)
|
|
if not self.paths.git_dir.exists():
|
|
self.paths.git_dir.mkdir(parents=True, exist_ok=True)
|
|
if not (self.paths.git_dir / "HEAD").exists():
|
|
self._run_git(["init"])
|
|
self._run_git(["config", "user.name", "EasyAgent Versioning"], check=False)
|
|
self._run_git(["config", "user.email", "easyagent-versioning@local"], check=False)
|
|
self._ensure_exclude_paths()
|
|
head = self._get_head_commit()
|
|
if not head:
|
|
self._stage_all()
|
|
self._run_git(["commit", "--allow-empty", "-m", "init workspace snapshot"])
|
|
head = self._get_head_commit()
|
|
return {"ok": True, "head": head}
|
|
|
|
def set_enabled(self, enabled: bool, mode: Optional[str] = None) -> Dict[str, Any]:
|
|
meta = self.load_meta()
|
|
if enabled:
|
|
self.ensure_repo()
|
|
meta["enabled"] = bool(enabled)
|
|
meta["mode"] = "overwrite"
|
|
meta["last_commit"] = self._get_head_commit()
|
|
saved = self.save_meta(meta)
|
|
return saved
|
|
|
|
def ensure_initial_checkpoint(
|
|
self,
|
|
*,
|
|
workspace_path: Optional[str] = None,
|
|
conversation_snapshot: Optional[Dict[str, Any]] = None,
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Ensure a visible seq=0 checkpoint exists after enabling versioning.
|
|
"""
|
|
self.ensure_repo()
|
|
existing = self.get_checkpoint(0)
|
|
if existing:
|
|
return {"created": False, "row": existing, "reason": "already_exists"}
|
|
|
|
current_head = self._get_head_commit()
|
|
if not current_head:
|
|
raise VersioningError("创建初始版本点失败：未获取到 commit")
|
|
|
|
row = {
|
|
"seq": 0,
|
|
"message": "版本管理开启（初始状态）",
|
|
"message_preview": "版本管理开启（初始状态）",
|
|
"message_index": -1,
|
|
"timestamp": datetime.now().isoformat(),
|
|
"workspace_path": str(workspace_path or ""),
|
|
"commit": current_head,
|
|
"parent_commit": None,
|
|
"insertions": 0,
|
|
"deletions": 0,
|
|
"files_changed": 0,
|
|
"files": [],
|
|
"changed": False,
|
|
"run_status": "initial",
|
|
}
|
|
if isinstance(conversation_snapshot, dict):
|
|
row["snapshot_file"] = self._write_snapshot(0, conversation_snapshot)
|
|
self._append_input(row)
|
|
meta = self.load_meta()
|
|
meta["last_commit"] = current_head
|
|
if int(meta.get("last_input_seq") or 0) < 0:
|
|
meta["last_input_seq"] = 0
|
|
self.save_meta(meta)
|
|
return {"created": True, "row": row, "reason": "created"}
|
|
|
|
def ensure_baseline_for_first_input(self) -> Dict[str, Any]:
|
|
"""
|
|
Ensure the baseline commit reflects current workspace before first manual input checkpoint.
|
|
|
|
This is intentionally hidden from checkpoint list (no seq row created).
|
|
"""
|
|
self.ensure_repo()
|
|
meta = self.load_meta()
|
|
if self.get_checkpoint(0):
|
|
return {
|
|
"created": False,
|
|
"skipped": True,
|
|
"reason": "initial_checkpoint_exists",
|
|
"head": self._get_head_commit(),
|
|
}
|
|
if int(meta.get("last_input_seq") or 0) > 0:
|
|
return {
|
|
"created": False,
|
|
"skipped": True,
|
|
"reason": "already_has_input_checkpoint",
|
|
"head": self._get_head_commit(),
|
|
}
|
|
|
|
previous_head = self._get_head_commit()
|
|
self._stage_all()
|
|
_, staged_numstat, _ = self._run_git(["diff", "--cached", "--numstat"], check=False)
|
|
has_changes = bool((staged_numstat or "").strip())
|
|
if has_changes:
|
|
self._run_git(["commit", "-m", "baseline before input#1"])
|
|
current_head = self._get_head_commit()
|
|
meta["last_commit"] = current_head
|
|
self.save_meta(meta)
|
|
return {
|
|
"created": bool(has_changes),
|
|
"skipped": False,
|
|
"reason": "baseline_committed" if has_changes else "baseline_unchanged",
|
|
"head": current_head,
|
|
"parent_commit": previous_head,
|
|
"has_changes": has_changes,
|
|
}
|
|
|
|
# ----------------------------
|
|
# diff / checkpoint
|
|
# ----------------------------
|
|
def _parse_numstat(self, text: str) -> Dict[str, Dict[str, int]]:
|
|
stats: Dict[str, Dict[str, int]] = {}
|
|
for line in (text or "").splitlines():
|
|
parts = line.split("\t")
|
|
if len(parts) < 3:
|
|
continue
|
|
ins_raw, del_raw = parts[0], parts[1]
|
|
path = self._decode_git_path(parts[2])
|
|
ins = 0 if ins_raw == "-" else int(ins_raw or 0)
|
|
dele = 0 if del_raw == "-" else int(del_raw or 0)
|
|
stats[path] = {"insertions": ins, "deletions": dele}
|
|
return stats
|
|
|
|
def _build_diff_summary(self, parent_commit: Optional[str], commit: str) -> Dict[str, Any]:
|
|
if parent_commit:
|
|
_, numstat_text, _ = self._run_git(["diff", "--numstat", parent_commit, commit], check=False)
|
|
_, status_text, _ = self._run_git(["diff", "--name-status", parent_commit, commit], check=False)
|
|
else:
|
|
_, numstat_text, _ = self._run_git(["show", "--numstat", "--format=", commit], check=False)
|
|
_, status_text, _ = self._run_git(
|
|
["diff-tree", "--root", "--no-commit-id", "--name-status", "-r", commit],
|
|
check=False,
|
|
)
|
|
|
|
numstats = self._parse_numstat(numstat_text)
|
|
files: List[Dict[str, Any]] = []
|
|
total_insertions = 0
|
|
total_deletions = 0
|
|
for raw in (status_text or "").splitlines():
|
|
line = raw.strip()
|
|
if not line:
|
|
continue
|
|
parts = line.split("\t")
|
|
code = parts[0] if parts else ""
|
|
path = self._decode_git_path(parts[-1] if len(parts) >= 2 else "")
|
|
old_path = self._decode_git_path(parts[1]) if code.startswith("R") and len(parts) >= 3 else None
|
|
file_stat = numstats.get(path, {"insertions": 0, "deletions": 0})
|
|
ins = int(file_stat.get("insertions") or 0)
|
|
dele = int(file_stat.get("deletions") or 0)
|
|
total_insertions += ins
|
|
total_deletions += dele
|
|
files.append(
|
|
{
|
|
"path": path,
|
|
"status": code,
|
|
"old_path": old_path,
|
|
"insertions": ins,
|
|
"deletions": dele,
|
|
}
|
|
)
|
|
return {
|
|
"insertions": total_insertions,
|
|
"deletions": total_deletions,
|
|
"files_changed": len(files),
|
|
"files": files,
|
|
}
|
|
|
|
def create_checkpoint(
|
|
self,
|
|
*,
|
|
message: str,
|
|
message_index: int,
|
|
workspace_path: Optional[str] = None,
|
|
conversation_snapshot: Optional[Dict[str, Any]] = None,
|
|
run_status: Optional[str] = None,
|
|
) -> Dict[str, Any]:
|
|
self.ensure_repo()
|
|
previous_head = self._get_head_commit()
|
|
self._stage_all()
|
|
_, staged_numstat, _ = self._run_git(["diff", "--cached", "--numstat"], check=False)
|
|
has_changes = bool((staged_numstat or "").strip())
|
|
if has_changes:
|
|
next_seq = int(self.load_meta().get("last_input_seq") or 0) + 1
|
|
title = (message or "").strip().splitlines()[0][:72] if message else ""
|
|
commit_msg = f"input#{next_seq} {title}".strip()
|
|
self._run_git(["commit", "-m", commit_msg])
|
|
current_head = self._get_head_commit()
|
|
if not current_head:
|
|
raise VersioningError("创建版本快照失败：未获取到 commit")
|
|
|
|
diff_summary = (
|
|
self._build_diff_summary(previous_head, current_head)
|
|
if current_head != previous_head
|
|
else {"insertions": 0, "deletions": 0, "files_changed": 0, "files": []}
|
|
)
|
|
|
|
meta = self.load_meta()
|
|
seq = int(meta.get("last_input_seq") or 0) + 1
|
|
row = {
|
|
"seq": seq,
|
|
"message": (message or "").strip(),
|
|
"message_preview": ((message or "").strip().splitlines() or [""])[0][:200],
|
|
"message_index": int(message_index),
|
|
"timestamp": datetime.now().isoformat(),
|
|
"workspace_path": str(workspace_path or ""),
|
|
"commit": current_head,
|
|
"parent_commit": previous_head,
|
|
"insertions": int(diff_summary.get("insertions") or 0),
|
|
"deletions": int(diff_summary.get("deletions") or 0),
|
|
"files_changed": int(diff_summary.get("files_changed") or 0),
|
|
"files": diff_summary.get("files") or [],
|
|
"changed": bool(current_head != previous_head),
|
|
}
|
|
if isinstance(conversation_snapshot, dict):
|
|
row["snapshot_file"] = self._write_snapshot(seq, conversation_snapshot)
|
|
if run_status:
|
|
row["run_status"] = str(run_status)
|
|
self._append_input(row)
|
|
meta["last_input_seq"] = seq
|
|
meta["last_commit"] = current_head
|
|
self.save_meta(meta)
|
|
return row
|
|
|
|
def list_checkpoints(self) -> List[Dict[str, Any]]:
|
|
rows = self._read_inputs()
|
|
rows.sort(key=lambda item: int(item.get("seq") or 0), reverse=True)
|
|
return rows
|
|
|
|
def get_checkpoint(self, seq: int) -> Optional[Dict[str, Any]]:
|
|
for row in self._read_inputs():
|
|
if int(row.get("seq") or 0) == int(seq):
|
|
return row
|
|
return None
|
|
|
|
def _collect_file_patch_lines(
|
|
self,
|
|
*,
|
|
parent_commit: Optional[str],
|
|
commit: str,
|
|
path: str,
|
|
old_path: Optional[str] = None,
|
|
max_lines: int = 600,
|
|
) -> Dict[str, Any]:
|
|
target_paths: List[str] = []
|
|
for candidate in [old_path, path]:
|
|
value = str(candidate or "").strip()
|
|
if value and value not in target_paths:
|
|
target_paths.append(value)
|
|
if not target_paths:
|
|
return {"lines": [], "truncated": False}
|
|
|
|
if parent_commit:
|
|
_, patch_text, _ = self._run_git(
|
|
["diff", "--no-color", "--unified=200", parent_commit, commit, "--", *target_paths],
|
|
check=False,
|
|
)
|
|
else:
|
|
_, patch_text, _ = self._run_git(
|
|
["show", "--no-color", "--unified=200", "--format=", commit, "--", *target_paths],
|
|
check=False,
|
|
)
|
|
parsed = self._extract_contextual_patch_lines(patch_text or "", max_lines=max_lines)
|
|
lines = parsed.get("lines") or []
|
|
truncated = bool(parsed.get("truncated"))
|
|
# 回退策略：若按路径过滤拿不到行，但该文件确实有增删，尝试从整次 commit patch 中按文件段提取。
|
|
if not lines:
|
|
fallback = self._collect_file_patch_lines_from_full_diff(
|
|
parent_commit=parent_commit,
|
|
commit=commit,
|
|
target_paths=target_paths,
|
|
max_lines=max_lines,
|
|
)
|
|
lines = fallback.get("lines") or []
|
|
truncated = bool(fallback.get("truncated")) or truncated
|
|
return {"lines": lines, "truncated": truncated}
|
|
|
|
def _collect_file_patch_lines_from_full_diff(
|
|
self,
|
|
*,
|
|
parent_commit: Optional[str],
|
|
commit: str,
|
|
target_paths: List[str],
|
|
max_lines: int = 600,
|
|
) -> Dict[str, Any]:
|
|
if parent_commit:
|
|
_, patch_text, _ = self._run_git(
|
|
["diff", "--no-color", "--unified=200", parent_commit, commit],
|
|
check=False,
|
|
)
|
|
else:
|
|
_, patch_text, _ = self._run_git(
|
|
["show", "--no-color", "--unified=200", "--format=", commit],
|
|
check=False,
|
|
)
|
|
|
|
targets = {str(p or "").lstrip("./") for p in target_paths if str(p or "").strip()}
|
|
chunk_lines: List[str] = []
|
|
lines: List[Dict[str, Any]] = []
|
|
truncated = False
|
|
in_target_section = False
|
|
for raw in (patch_text or "").splitlines():
|
|
if raw.startswith("diff --git "):
|
|
if in_target_section and chunk_lines and len(lines) < max_lines:
|
|
parsed_chunk = self._extract_contextual_patch_lines("\n".join(chunk_lines), max_lines=max_lines - len(lines))
|
|
lines.extend(parsed_chunk.get("lines") or [])
|
|
truncated = truncated or bool(parsed_chunk.get("truncated"))
|
|
chunk_lines = []
|
|
in_target_section = False
|
|
try:
|
|
parts = raw.split(" ")
|
|
a_path = self._decode_git_path((parts[2] or "").replace("a/", "", 1)).lstrip("./")
|
|
b_path = self._decode_git_path((parts[3] or "").replace("b/", "", 1)).lstrip("./")
|
|
if a_path in targets or b_path in targets:
|
|
in_target_section = True
|
|
except Exception:
|
|
in_target_section = False
|
|
continue
|
|
if not in_target_section:
|
|
continue
|
|
chunk_lines.append(raw)
|
|
if len(lines) >= max_lines:
|
|
truncated = True
|
|
break
|
|
if in_target_section and chunk_lines and len(lines) < max_lines and not truncated:
|
|
parsed_chunk = self._extract_contextual_patch_lines("\n".join(chunk_lines), max_lines=max_lines - len(lines))
|
|
lines.extend(parsed_chunk.get("lines") or [])
|
|
truncated = truncated or bool(parsed_chunk.get("truncated"))
|
|
if len(lines) > max_lines:
|
|
lines = lines[:max_lines]
|
|
truncated = True
|
|
return {"lines": lines, "truncated": truncated}
|
|
|
|
def get_checkpoint_detail(self, seq: int, *, include_patch: bool = True) -> Optional[Dict[str, Any]]:
|
|
row = self.get_checkpoint(seq)
|
|
if not row:
|
|
return None
|
|
payload = dict(row)
|
|
files = payload.get("files")
|
|
if not include_patch or not isinstance(files, list):
|
|
return payload
|
|
parent_commit = payload.get("parent_commit")
|
|
commit = payload.get("commit")
|
|
normalized_files: List[Dict[str, Any]] = []
|
|
for file_item in files:
|
|
if not isinstance(file_item, dict):
|
|
continue
|
|
item = dict(file_item)
|
|
if commit and str(item.get("path") or "").strip():
|
|
patch = self._collect_file_patch_lines(
|
|
parent_commit=parent_commit,
|
|
commit=commit,
|
|
path=str(item.get("path") or ""),
|
|
old_path=item.get("old_path"),
|
|
)
|
|
item["patch_lines"] = patch.get("lines") or []
|
|
item["patch_truncated"] = bool(patch.get("truncated"))
|
|
else:
|
|
item["patch_lines"] = []
|
|
item["patch_truncated"] = False
|
|
normalized_files.append(item)
|
|
payload["files"] = normalized_files
|
|
return payload
|
|
|
|
def get_latest_checkpoint(self) -> Optional[Dict[str, Any]]:
|
|
rows = self.list_checkpoints()
|
|
return rows[0] if rows else None
|
|
|
|
def prune_checkpoints_after(self, seq: int) -> Dict[str, Any]:
|
|
target = int(seq)
|
|
rows = self._read_inputs()
|
|
if not rows:
|
|
return {"kept": 0, "removed": 0, "max_seq": 0}
|
|
|
|
kept_rows = [row for row in rows if int(row.get("seq") or 0) <= target]
|
|
removed_rows = [row for row in rows if int(row.get("seq") or 0) > target]
|
|
|
|
for row in removed_rows:
|
|
rel = row.get("snapshot_file")
|
|
if not rel:
|
|
continue
|
|
try:
|
|
path = (self.paths.save_root / str(rel)).resolve()
|
|
if path.exists() and path.is_file():
|
|
path.unlink()
|
|
except Exception:
|
|
continue
|
|
|
|
self._write_inputs(kept_rows)
|
|
max_seq = max((int(r.get("seq") or 0) for r in kept_rows), default=0)
|
|
latest_kept = None
|
|
for r in kept_rows:
|
|
if int(r.get("seq") or 0) == max_seq:
|
|
latest_kept = r
|
|
meta = self.load_meta()
|
|
meta["last_input_seq"] = max_seq
|
|
meta["last_commit"] = (latest_kept or {}).get("commit") or self._get_head_commit()
|
|
self.save_meta(meta)
|
|
return {
|
|
"kept": len(kept_rows),
|
|
"removed": len(removed_rows),
|
|
"max_seq": max_seq,
|
|
"last_commit": meta.get("last_commit"),
|
|
}
|
|
|
|
# ----------------------------
|
|
# restore / status
|
|
# ----------------------------
|
|
def restore_to_commit(self, commit: str) -> Dict[str, Any]:
|
|
if not commit:
|
|
raise VersioningError("缺少 commit")
|
|
self.ensure_repo()
|
|
ok, _, _ = self._run_git(["cat-file", "-e", f"{commit}^{{commit}}"], check=False)
|
|
if not ok:
|
|
raise VersioningError("目标 commit 不存在")
|
|
self._run_git(["reset", "--hard", commit])
|
|
self._run_git(["clean", "-fd"])
|
|
meta = self.load_meta()
|
|
meta["last_restored_commit"] = commit
|
|
meta["last_commit"] = self._get_head_commit()
|
|
self.save_meta(meta)
|
|
return {"success": True, "commit": commit}
|
|
|
|
def detect_mismatch(self, latest_commit: Optional[str], expected_workspace_path: Optional[str] = None) -> Dict[str, Any]:
|
|
expected = str(expected_workspace_path or "").strip()
|
|
current = str(self.project_path)
|
|
workspace_matched = True if not expected else (Path(expected).expanduser().resolve() == self.project_path)
|
|
if not (self.paths.git_dir / "HEAD").exists():
|
|
return {
|
|
"has_repo": False,
|
|
"head": None,
|
|
"dirty": False,
|
|
"mismatch": False,
|
|
"workspace_matched": workspace_matched,
|
|
}
|
|
if not workspace_matched:
|
|
return {
|
|
"has_repo": True,
|
|
"head": self._get_head_commit(),
|
|
"dirty": False,
|
|
"mismatch": False,
|
|
"workspace_matched": False,
|
|
"expected_workspace_path": expected,
|
|
"current_workspace_path": current,
|
|
}
|
|
head = self._get_head_commit()
|
|
dirty = not self._is_clean()
|
|
mismatch = bool(dirty or (latest_commit and head and latest_commit != head))
|
|
return {
|
|
"has_repo": True,
|
|
"head": head,
|
|
"dirty": dirty,
|
|
"mismatch": mismatch,
|
|
"workspace_matched": True,
|
|
"expected_workspace_path": expected or current,
|
|
"current_workspace_path": current,
|
|
}
|
|
def _extract_contextual_patch_lines(self, patch_text: str, max_lines: int = 600) -> Dict[str, Any]:
|
|
"""
|
|
Extract +/- lines plus 3 context lines before and after changed regions.
|
|
"""
|
|
lines: List[Dict[str, Any]] = []
|
|
truncated = False
|
|
pre_context: deque[str] = deque(maxlen=3)
|
|
post_context_needed = 0
|
|
|
|
for raw in (patch_text or "").splitlines():
|
|
if raw.startswith("diff --git "):
|
|
pre_context.clear()
|
|
post_context_needed = 0
|
|
continue
|
|
if raw.startswith("index ") or raw.startswith("--- ") or raw.startswith("+++ "):
|
|
continue
|
|
if raw.startswith("@@"):
|
|
pre_context.clear()
|
|
post_context_needed = 0
|
|
continue
|
|
|
|
if not raw:
|
|
prefix = " "
|
|
content = ""
|
|
else:
|
|
prefix = raw[0]
|
|
content = raw[1:] if len(raw) > 1 else ""
|
|
|
|
if prefix == " ":
|
|
if post_context_needed > 0:
|
|
lines.append({"type": "context", "content": content})
|
|
post_context_needed -= 1
|
|
else:
|
|
pre_context.append(content)
|
|
elif prefix in {"+", "-"}:
|
|
if pre_context:
|
|
for ctx in pre_context:
|
|
lines.append({"type": "context", "content": ctx})
|
|
pre_context.clear()
|
|
lines.append({"type": "add" if prefix == "+" else "remove", "content": content})
|
|
post_context_needed = 3
|
|
else:
|
|
continue
|
|
|
|
if len(lines) >= max_lines:
|
|
truncated = True
|
|
break
|
|
|
|
return {"lines": lines[:max_lines], "truncated": truncated}
|