"""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.""" TRACKING_MODE_WORKSPACE_AND_CONVERSATION = "workspace_and_conversation" TRACKING_MODE_CONVERSATION_ONLY = "conversation_only" SYSTEM_AUTO_DIRS = ("compact_result", "skills", "user_upload") 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(), ) @classmethod def normalize_tracking_mode(cls, mode: Optional[str]) -> str: raw = str(mode or "").strip().lower() if raw == cls.TRACKING_MODE_CONVERSATION_ONLY: return cls.TRACKING_MODE_CONVERSATION_ONLY return cls.TRACKING_MODE_WORKSPACE_AND_CONVERSATION # ---------------------------- # 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.""" 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 = "" existing_lines = set(existing.splitlines()) lines_to_add: List[str] = [] try: rel_save = self.paths.save_root.relative_to(self.project_path) marker = str(rel_save).replace("\\", "/").rstrip("/") if marker: lines_to_add.append(f"{marker}/") except ValueError: pass for dirname in self.SYSTEM_AUTO_DIRS: clean = str(dirname or "").strip().strip("/").replace("\\", "/") if not clean: continue # 同时忽略根目录与任意层级同名目录 lines_to_add.append(f"/{clean}/") lines_to_add.append(f"{clean}/") pending = [line for line in lines_to_add if line not in existing_lines] if not pending: return if existing and not existing.endswith("\n"): pending.insert(0, "") with info_exclude.open("a", encoding="utf-8") as fh: fh.write("\n".join(pending) + "\n") def _drop_ignored_from_index(self) -> None: """ Ensure ignored system directories are not kept as tracked files in hidden git index. """ for dirname in self.SYSTEM_AUTO_DIRS: clean = str(dirname or "").strip().strip("/").replace("\\", "/") if not clean: continue self._run_git(["rm", "-r", "--cached", "--ignore-unmatch", "--", clean], check=False) def _stage_all(self) -> None: self._ensure_exclude_paths() self._drop_ignored_from_index() # 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", "tracking_mode": self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION, "last_commit": None, "last_input_seq": 0, "updated_at": None, } try: payload = json.loads(self.paths.meta_path.read_text(encoding="utf-8")) or {} payload["tracking_mode"] = self.normalize_tracking_mode(payload.get("tracking_mode")) return payload except Exception: return { "enabled": False, "mode": "overwrite", "tracking_mode": self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION, "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["tracking_mode"] = self.normalize_tracking_mode(payload.get("tracking_mode")) 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 {}) payload["tracking_mode"] = self.normalize_tracking_mode(payload.get("tracking_mode")) 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, *, tracking_mode: Optional[str] = None, ) -> Dict[str, Any]: meta = self.load_meta() normalized_tracking_mode = self.normalize_tracking_mode( tracking_mode or meta.get("tracking_mode") ) if enabled and normalized_tracking_mode == self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION: self.ensure_repo() meta["enabled"] = bool(enabled) meta["mode"] = "overwrite" meta["tracking_mode"] = normalized_tracking_mode 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, tracking_mode: Optional[str] = None, ) -> Dict[str, Any]: """ Ensure a visible seq=0 checkpoint exists after enabling versioning. """ normalized_tracking_mode = self.normalize_tracking_mode(tracking_mode) if normalized_tracking_mode == self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION: 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 normalized_tracking_mode == self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION and 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", "tracking_mode": normalized_tracking_mode, } if isinstance(conversation_snapshot, dict): row["snapshot_file"] = self._write_snapshot(0, conversation_snapshot) self._append_input(row) meta = self.load_meta() meta["tracking_mode"] = normalized_tracking_mode 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, *, tracking_mode: Optional[str] = None, ) -> 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). """ normalized_tracking_mode = self.normalize_tracking_mode(tracking_mode) if normalized_tracking_mode == self.TRACKING_MODE_CONVERSATION_ONLY: return { "created": False, "skipped": True, "reason": "conversation_only_mode", "head": self._get_head_commit(), } 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, tracking_mode: Optional[str] = None, ) -> Dict[str, Any]: meta = self.load_meta() normalized_tracking_mode = self.normalize_tracking_mode( tracking_mode or meta.get("tracking_mode") ) previous_head: Optional[str] = self._get_head_commit() current_head: Optional[str] = previous_head if normalized_tracking_mode == self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION: 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(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 normalized_tracking_mode == self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION and current_head != previous_head else {"insertions": 0, "deletions": 0, "files_changed": 0, "files": []} ) 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( normalized_tracking_mode == self.TRACKING_MODE_WORKSPACE_AND_CONVERSATION and current_head != previous_head ), "tracking_mode": normalized_tracking_mode, } 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 meta["tracking_mode"] = normalized_tracking_mode 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, *, tracking_mode: Optional[str] = None, ) -> Dict[str, Any]: normalized_tracking_mode = self.normalize_tracking_mode(tracking_mode) if normalized_tracking_mode == self.TRACKING_MODE_CONVERSATION_ONLY: return { "has_repo": (self.paths.git_dir / "HEAD").exists(), "head": self._get_head_commit(), "dirty": False, "mismatch": False, "workspace_matched": True, } 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}