diff --git a/CoderMind/.gitignore b/CoderMind/.gitignore index 835fb40..298698b 100644 --- a/CoderMind/.gitignore +++ b/CoderMind/.gitignore @@ -239,3 +239,5 @@ plans/ .mcp.json .github/agents/ .github/prompts/ + +.claude/commands/ diff --git a/CoderMind/pyproject.toml b/CoderMind/pyproject.toml index f73a08c..7485fd2 100644 --- a/CoderMind/pyproject.toml +++ b/CoderMind/pyproject.toml @@ -1,6 +1,6 @@ [project] name = "cmind-cli" -version = "0.1.9" +version = "0.1.10" description = "CoderMind CLI - A tool to generate feature trees for repository planning and code generation." requires-python = ">=3.12" dependencies = [ diff --git a/CoderMind/scripts/check_code_gen.py b/CoderMind/scripts/check_code_gen.py index f2165ec..5a0da18 100644 --- a/CoderMind/scripts/check_code_gen.py +++ b/CoderMind/scripts/check_code_gen.py @@ -13,9 +13,14 @@ import json import argparse +import sys from pathlib import Path from typing import Dict, Any, List, Tuple +SCRIPTS_DIR = Path(__file__).resolve().parent +if str(SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPTS_DIR)) + # Import centralized paths and state loader from common.paths import ( TASKS_FILE, @@ -24,6 +29,8 @@ cmd_for, REPO_DIR, ) +from common.run_events import ArtifactEvent, CommandRun, StepEvent, VerificationEvent +from common.run_report import write_command_report from common.execution_state import load_code_gen_state from common.execution_state import load_code_gen_state as _load_state, save_code_gen_state as _save_state from common.execution_state import complete_batch as _complete_batch @@ -427,6 +434,46 @@ def determine_state( return result +def _write_code_gen_report(result: Dict[str, Any]) -> str | None: + try: + stats = result.get("stats") or {} + report_path = write_command_report(CommandRun( + command="code_gen", + title="CoderMind code_gen Progress View", + status=result.get("type"), + summary=[ + {"label": "state", "value": result.get("type", "")}, + {"label": "total", "value": stats.get("total_tasks", 0)}, + {"label": "completed", "value": stats.get("completed", 0)}, + {"label": "failed", "value": stats.get("failed", 0)}, + {"label": "remaining", "value": stats.get("remaining", 0)}, + {"label": "current batch", "value": (result.get("current_batch") or {}).get("batch_id", "")}, + {"label": "next batch", "value": result.get("next_batch", "")}, + ], + steps=[ + StepEvent(name="determine_state", status=result.get("type"), reason=result.get("message", "")), + StepEvent(name="current_batch", status=(result.get("current_batch") or {}).get("phase", "none"), reason=(result.get("current_batch") or {}).get("file_path", "")), + StepEvent(name="next_action", status="available" if result.get("next_action") else "missing", reason=result.get("next_action", "")), + ], + artifacts=[ + ArtifactEvent(label="tasks", path=TASKS_FILE), + ArtifactEvent(label="code_gen_state", path=STATE_FILE), + ], + verification=[VerificationEvent(name="state", status=result.get("type"), detail=result.get("message", ""))], + evidence={ + "type": result.get("type"), + "stats": stats, + "current_batch": result.get("current_batch"), + "next_batch": result.get("next_batch"), + "next_action": result.get("next_action"), + }, + )) + return str(report_path) + except Exception as exc: + result["report_error"] = str(exc) + return None + + def print_status(result: Dict[str, Any], json_output: bool = False) -> None: """Print the status in human-readable or JSON format.""" if json_output: @@ -474,7 +521,9 @@ def print_status(result: Dict[str, Any], json_output: bool = False) -> None: # Next batch info if result.get("next_batch"): print(f"\n Next Batch: {result['next_batch']}") - + if result.get("report_path"): + print(f"\n Report: {result['report_path']}") + # Guidance print("\n " + "─" * 60) @@ -516,6 +565,9 @@ def main(): args = parser.parse_args() result = determine_state(args.tasks, args.state) + report_path = _write_code_gen_report(result) + if report_path: + result["report_path"] = report_path print_status(result, json_output=args.json) # Return exit code based on state diff --git a/CoderMind/scripts/common/git_utils.py b/CoderMind/scripts/common/git_utils.py index 3c72866..b08e45c 100644 --- a/CoderMind/scripts/common/git_utils.py +++ b/CoderMind/scripts/common/git_utils.py @@ -85,12 +85,12 @@ class GitRunner: def __init__( self, repo_path: str, - main_branch: str = "main", + main_branch: str = MAIN_BRANCH, logger: Optional[logging.Logger] = None ): self.repo_path = Path(repo_path) self.logger = logger or logging.getLogger(__name__) - self.main_branch = self.MAIN_BRANCH + self.main_branch = main_branch # Ensure repo exists and is a git repo self._ensure_git_repository() @@ -149,7 +149,7 @@ def _ensure_git_repository(self) -> None: if not git_dir.exists(): self.logger.info("Initializing git repository...") self.repo_path.mkdir(parents=True, exist_ok=True) - self.run_git(["init", "-b", self.MAIN_BRANCH]) + self.run_git(["init", "-b", self.main_branch]) # Configure safe directory self.run_git([ @@ -391,11 +391,11 @@ def get_diff( to_commit: str = "HEAD" ) -> str: """Get diff between commits. - + Args: from_commit: Start commit (default: parent of to_commit) to_commit: End commit (default: HEAD) - + Returns: Diff content as string """ @@ -403,9 +403,24 @@ def get_diff( result = self.run_git(["diff", from_commit, to_commit]) else: result = self.run_git(["diff", f"{to_commit}^", to_commit]) - + return result.stdout if result.success else "" - + + def get_file_diffs( + self, + from_commit: Optional[str] = None, + to_commit: str = "HEAD", + files: Optional[List[str]] = None, + ) -> List[Dict[str, str]]: + """Get per-file diff rows between commits.""" + return file_diffs_between( + self.repo_path, + from_commit, + to_commit, + files=files, + py_only=False, + ) + def get_changed_files( self, from_commit: Optional[str] = None, @@ -565,6 +580,34 @@ def read_head(repo_dir: str | Path) -> Optional[dict]: } +def git_workspace_prefix(workspace_dir: str | Path) -> str: + """Return the path from the git root to ``workspace_dir``. + + Returns ``""`` when ``workspace_dir`` is the git root or when git metadata + cannot be read. This keeps callers safe outside git repositories. + """ + if not workspace_dir: + return "" + workspace_path = Path(workspace_dir) + if not workspace_path.is_dir(): + return "" + + git_root = _run_git_readonly( + ["rev-parse", "--show-toplevel"], + workspace_path, + ) + if not git_root: + return "" + + try: + rel = workspace_path.resolve().relative_to(Path(git_root).resolve()) + except ValueError: + return "" + + rel_str = rel.as_posix() + return "" if rel_str == "." else rel_str + + # --------------------------------------------------------------------------- # Diff helpers — produce ``(modified, renames)`` from various git scopes. # --------------------------------------------------------------------------- @@ -589,6 +632,52 @@ def read_head(repo_dir: str | Path) -> Optional[dict]: _GIT_STATUS_COPY_PREFIX = "C" +def _parse_name_status_rows( + raw: Optional[str], + *, + py_only: bool = True, +) -> List[Dict[str, str]]: + rows: List[Dict[str, str]] = [] + if not raw: + return rows + + def _keep(p: str) -> bool: + return (not py_only) or p.endswith(".py") + + for line in raw.splitlines(): + parts = line.split("\t") + if len(parts) < 2: + continue + status = parts[0] + if status.startswith(_GIT_STATUS_RENAME_PREFIX) or status.startswith( + _GIT_STATUS_COPY_PREFIX + ): + if len(parts) < 3: + continue + old_path, new_path = parts[1], parts[2] + if _keep(new_path) or _keep(old_path): + rows.append({ + "file": new_path, + "change_type": "rename" if status.startswith(_GIT_STATUS_RENAME_PREFIX) else "copy", + "old_file": old_path, + "status": status, + }) + continue + path = parts[1] + if not _keep(path): + continue + if status == _GIT_STATUS_ADDED: + change_type = "add" + elif status == _GIT_STATUS_DELETED: + change_type = "delete" + elif status == _GIT_STATUS_MODIFIED: + change_type = "modify" + else: + continue + rows.append({"file": path, "change_type": change_type, "status": status}) + return rows + + def _parse_name_status( raw: Optional[str], *, @@ -616,40 +705,76 @@ def _parse_name_status( if not raw: return modified, renames - def _keep(p: str) -> bool: - return (not py_only) or p.endswith(".py") - - for line in raw.splitlines(): - parts = line.split("\t") - if len(parts) < 2: - continue - status = parts[0] - if status.startswith(_GIT_STATUS_RENAME_PREFIX) or status.startswith( - _GIT_STATUS_COPY_PREFIX - ): - if len(parts) < 3: + for row in _parse_name_status_rows(raw, py_only=py_only): + path = row.get("file", "") + if row.get("change_type") in ("rename", "copy"): + old_path = row.get("old_file", "") + if old_path and path: + renames[old_path] = path + if py_only and not path.endswith(".py"): continue - old_path, new_path = parts[1], parts[2] - if _keep(new_path) or _keep(old_path): - renames[old_path] = new_path - # update_files() treats the OLD path as a deletion (via - # ``renames``) and the NEW path as something it must - # reparse — so we surface the new path through the - # modified list as well. - if _keep(new_path): - modified.append(new_path) - continue - path = parts[1] - if not _keep(path): - continue - if status in (_GIT_STATUS_ADDED, _GIT_STATUS_DELETED, _GIT_STATUS_MODIFIED): + if path: modified.append(path) - # Type / unmerged / other status letters → ignore (caller will - # fall back to full sync via the safety threshold if there are - # many of them). return modified, renames +def _diff_range_args( + from_commit: Optional[str] = None, + to_commit: str = "HEAD", +) -> List[str]: + if from_commit: + return [from_commit, to_commit] + return [f"{to_commit}^", to_commit] + + +def file_diffs_between( + repo_dir: str | Path, + from_commit: Optional[str] = None, + to_commit: str = "HEAD", + *, + files: Optional[List[str]] = None, + py_only: bool = False, +) -> List[Dict[str, str]]: + """Return per-file diff rows without mutating git state.""" + if not repo_dir: + return [] + repo_path = Path(repo_dir) + if not repo_path.is_dir(): + return [] + + range_args = _diff_range_args(from_commit, to_commit) + raw_status = _run_git_readonly( + ["diff", "--relative", "--name-status", "-M", *range_args], + repo_path, + ) + rows = _parse_name_status_rows(raw_status, py_only=py_only) + if not rows: + return [] + + selected = {str(path) for path in files or [] if path} + diff_rows: List[Dict[str, str]] = [] + for row in rows: + file_path = row.get("file", "") + old_file = row.get("old_file", "") + if selected and file_path not in selected and old_file not in selected: + continue + pathspecs = [path for path in [old_file, file_path] if path] + raw_diff = _run_git_readonly( + ["diff", *range_args, "--", *pathspecs], + repo_path, + timeout=10.0, + ) + diff_row = { + "file": file_path, + "change_type": row.get("change_type", "modify"), + "diff": raw_diff or "", + } + if old_file: + diff_row["old_file"] = old_file + diff_rows.append(diff_row) + return diff_rows + + def staged_changes( repo_dir: str | Path, ) -> Tuple[List[str], Dict[str, str]]: @@ -669,14 +794,14 @@ def staged_changes( if not repo_path.is_dir(): return [], {} raw = _run_git_readonly( - ["diff", "--cached", "--name-status", "-M", "HEAD"], + ["diff", "--cached", "--relative", "--name-status", "-M", "HEAD"], repo_path, ) if raw is None: # ``HEAD`` may not exist yet (unborn branch); try without it so # the very first commit's staged files still get picked up. raw = _run_git_readonly( - ["diff", "--cached", "--name-status", "-M"], + ["diff", "--cached", "--relative", "--name-status", "-M"], repo_path, ) return _parse_name_status(raw) @@ -706,7 +831,7 @@ def working_tree_changes( return [], {} raw = _run_git_readonly( - ["diff", "--name-status", "-M", "HEAD"], + ["diff", "--relative", "--name-status", "-M", "HEAD"], repo_path, ) modified, renames = _parse_name_status(raw) @@ -749,7 +874,7 @@ def changed_files_between( if not repo_path.is_dir(): return [], {} raw = _run_git_readonly( - ["diff", "--name-status", "-M", f"{old_ref}..{new_ref}"], + ["diff", "--relative", "--name-status", "-M", f"{old_ref}..{new_ref}"], repo_path, ) return _parse_name_status(raw) diff --git a/CoderMind/scripts/common/paths.py b/CoderMind/scripts/common/paths.py index bebd0d0..15c62b0 100644 --- a/CoderMind/scripts/common/paths.py +++ b/CoderMind/scripts/common/paths.py @@ -301,7 +301,10 @@ def cmd_for(script_relpath: str) -> str: RPG_EDIT_PLAN_FILE = DATA_DIR / "rpg_edit_plan.json" RPG_EDIT_IMPACT_FILE = DATA_DIR / "rpg_edit_impact.json" +RPG_EDIT_VALIDATE_FILE = DATA_DIR / "rpg_edit_validate.json" +RPG_EDIT_LOCATE_FILE = DATA_DIR / "rpg_edit_locate.json" RPG_EDIT_CODE_RESULT_FILE = DATA_DIR / "rpg_edit_code_result.json" +RPG_EDIT_APPLY_RESULT_FILE = DATA_DIR / "rpg_edit_apply_result.json" RPG_EDIT_REVIEW_RESULT_FILE = DATA_DIR / "rpg_edit_review_result.json" diff --git a/CoderMind/scripts/common/run_events.py b/CoderMind/scripts/common/run_events.py new file mode 100644 index 0000000..99ccf11 --- /dev/null +++ b/CoderMind/scripts/common/run_events.py @@ -0,0 +1,238 @@ +"""Structured event contract for command run reports.""" + +from __future__ import annotations + +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Mapping, Sequence + + +def _to_plain(value: Any) -> Any: + if hasattr(value, "to_dict"): + return value.to_dict() + if isinstance(value, Path): + return str(value) + if isinstance(value, Mapping): + return {str(key): _to_plain(item_value) for key, item_value in value.items()} + if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)): + return [_to_plain(item) for item in value] + return value + + +def _compact(values: Mapping[str, Any]) -> dict[str, Any]: + compacted: dict[str, Any] = {} + for key, value in values.items(): + plain = _to_plain(value) + if plain is None or plain == "" or plain == [] or plain == {}: + continue + compacted[key] = plain + return compacted + + +def _as_list(value: Any) -> list[Any]: + if value is None: + return [] + if isinstance(value, Sequence) and not isinstance(value, (str, bytes, bytearray)): + return [_to_plain(item) for item in value] + return [_to_plain(value)] + + +def _artifact_status(path: Any, status: Any = None) -> Any: + if status not in (None, ""): + return status + if path in (None, ""): + return "missing" + try: + return "available" if Path(str(path)).expanduser().exists() else "missing" + except (OSError, ValueError): + return "missing" + + +@dataclass +class StepEvent: + name: Any = "step" + status: Any = "recorded" + reason: Any = None + duration: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "name": self.name, + "status": self.status, + "reason": self.reason, + "duration": self.duration, + }) + + +@dataclass +class RetrievalEvent: + query: Any = None + tool: Any = None + hits: Any = None + reason: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "query": self.query, + "tool": self.tool, + "hits": self.hits, + "reason": self.reason, + }) + + +@dataclass +class RPGDeltaEvent: + node_id: Any = None + name: Any = None + type: Any = None + path: Any = None + change: Any = None + score: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "node_id": self.node_id, + "name": self.name, + "type": self.type, + "path": self.path, + "change": self.change, + "score": self.score, + }) + + +@dataclass +class DepGraphDeltaEvent: + dep_node_id: Any = None + path: Any = None + source_feature: Any = None + change: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "dep_node_id": self.dep_node_id, + "path": self.path, + "source_feature": self.source_feature, + "change": self.change, + }) + + +@dataclass +class CodeDeltaEvent: + file: Any = None + change_type: Any = None + before: Any = None + after: Any = None + diff: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "file": self.file, + "change_type": self.change_type, + "before": self.before, + "after": self.after, + "diff": self.diff, + }) + + +@dataclass +class VerificationEvent: + name: Any = "verification" + status: Any = None + detail: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "name": self.name, + "status": self.status, + "detail": self.detail, + }) + + +@dataclass +class UserDecisionEvent: + decision: Any = None + branch: Any = None + before_state: Any = None + rollback_path: Any = None + confirmed: Any = None + apply_status: Any = None + test_status: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "decision": self.decision, + "branch": self.branch, + "before_state": self.before_state, + "rollback_path": self.rollback_path, + "confirmed": self.confirmed, + "apply_status": self.apply_status, + "test_status": self.test_status, + }) + + +@dataclass +class ArtifactEvent: + label: Any = "artifact" + path: Any = None + status: Any = None + + def to_dict(self) -> dict[str, Any]: + return _compact({ + "label": self.label, + "path": self.path, + "status": _artifact_status(self.path, self.status), + }) + + +@dataclass +class CommandRun: + command: str = "command" + status: Any = None + title: Any = None + timestamp: Any = None + summary: Any = None + steps: Any = None + retrievals: Any = None + rpg_deltas: Any = None + dep_graph_deltas: Any = None + code_deltas: Any = None + verification: Any = None + user_decisions: Any = None + artifacts: Any = None + evidence: Any = None + + def to_dict(self) -> dict[str, Any]: + data: dict[str, Any] = { + "command": self.command, + "status": _to_plain(self.status), + "title": _to_plain(self.title), + "timestamp": _to_plain(self.timestamp), + "summary": _as_list(self.summary), + "steps": _as_list(self.steps), + "retrievals": _as_list(self.retrievals), + "rpg_deltas": _as_list(self.rpg_deltas), + "dep_graph_deltas": _as_list(self.dep_graph_deltas), + "code_deltas": _as_list(self.code_deltas), + "verification": _as_list(self.verification), + "user_decisions": _as_list(self.user_decisions), + "artifacts": _as_list(self.artifacts), + "evidence": _to_plain(self.evidence) if self.evidence is not None else {}, + } + return _compact(data) + + +__all__ = [ + "CommandRun", + "StepEvent", + "RetrievalEvent", + "RPGDeltaEvent", + "DepGraphDeltaEvent", + "CodeDeltaEvent", + "VerificationEvent", + "UserDecisionEvent", + "ArtifactEvent", + "_compact", + "_to_plain", + "_as_list", + "_artifact_status", +] diff --git a/CoderMind/scripts/common/run_report.py b/CoderMind/scripts/common/run_report.py new file mode 100644 index 0000000..d0bae9d --- /dev/null +++ b/CoderMind/scripts/common/run_report.py @@ -0,0 +1,611 @@ +"""Shared HTML renderer for CoderMind command run reports.""" + +from __future__ import annotations + +import json +import re +from datetime import datetime, timezone +from html import escape +from pathlib import Path +from typing import Any, Mapping, Sequence +from urllib.parse import quote + +from common.paths import REPORTS_DIR +from common.run_events import _to_plain + +_MAX_SUMMARY_CARDS = 7 + + +def write_command_report( + run: Any, + *, + report_dir: str | Path | None = None, + timestamp: str | datetime | None = None, +) -> Path: + """Write a sanitized Explain View HTML report and return its path.""" + if hasattr(run, "to_dict"): + data = run.to_dict() + elif isinstance(run, Mapping): + data = dict(run) + else: + raise TypeError("write_command_report() expects a CommandRun or mapping") + + data = _to_plain(data) + if not isinstance(data, Mapping): + raise TypeError("CommandRun.to_dict() must return a mapping") + + command = str(data.get("command") or "command") + status = data.get("status") + title = data.get("title") + timestamp = timestamp if timestamp is not None else data.get("timestamp") + report_dir = report_dir if report_dir is not None else data.get("report_dir") + + generated_at = _display_timestamp(timestamp) + filename_ts = _filename_timestamp(timestamp) + safe_command = _slug(command) + target_dir = Path(report_dir) if report_dir is not None else REPORTS_DIR + target_dir.mkdir(parents=True, exist_ok=True) + report_path = _unique_report_path(target_dir / f"cmind_run_{safe_command}_{filename_ts}.html") + + evidence = dict(data) + evidence_data = evidence.get("evidence") if isinstance(evidence.get("evidence"), Mapping) else {} + retrievals = data.get("retrievals") or evidence_data.get("retrievals") + code_deltas = data.get("code_deltas") or evidence_data.get("code_deltas") + focused_graph = data.get("focused_graph") or evidence_data.get("focused_graph") + user_decisions = data.get("user_decisions") or evidence_data.get("user_decisions") + + page_title = title or f"CoderMind {command} Explain View" + html = _render_page( + title=page_title, + command=command, + generated_at=generated_at, + status=status, + summary_cards=_normalize_cards(data.get("summary")), + stages=_normalize_stages(data.get("steps")), + retrievals=_normalize_retrievals(retrievals), + rpg_nodes=_normalize_nodes(data.get("rpg_deltas"), dep_graph=False), + dep_nodes=_normalize_nodes(data.get("dep_graph_deltas"), dep_graph=True), + code_deltas=_normalize_code_deltas(code_deltas), + focused_graph=_normalize_focused_graph(focused_graph), + artifacts=_normalize_artifacts(data.get("artifacts")), + verification=_normalize_verification(data.get("verification")), + user_decisions=_normalize_user_decisions(user_decisions), + evidence=evidence, + ) + report_path.write_text(html, encoding="utf-8") + return report_path + + +def _slug(value: str) -> str: + slug = re.sub(r"[^A-Za-z0-9_.-]+", "_", value).strip("._-") + return slug or "command" + + +def _unique_report_path(path: Path) -> Path: + if not path.exists(): + return path + stem = path.stem + suffix = path.suffix + for index in range(2, 1000): + candidate = path.with_name(f"{stem}_{index}{suffix}") + if not candidate.exists(): + return candidate + return path.with_name(f"{stem}_{datetime.now(timezone.utc).strftime('%f')}{suffix}") + + +def _display_timestamp(value: str | datetime | None) -> str: + if isinstance(value, datetime): + return value.astimezone(timezone.utc).isoformat(timespec="seconds") + if value is not None: + return str(value) + return datetime.now(timezone.utc).isoformat(timespec="seconds") + + +def _filename_timestamp(value: str | datetime | None) -> str: + if isinstance(value, datetime): + raw = value.astimezone(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + elif value is not None: + raw = str(value) + else: + raw = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + return _slug(raw) + + +def _as_sequence(value: Any) -> list[Any]: + if value is None: + return [] + if isinstance(value, (str, bytes, Path)): + return [value] + if isinstance(value, Mapping): + return [value] + if isinstance(value, Sequence): + return list(value) + return [value] + + +def _normalize_cards(value: Any) -> list[dict[str, Any]]: + cards: list[dict[str, Any]] = [] + for item in _as_sequence(value): + if isinstance(item, Mapping): + cards.append({ + "label": item.get("label") or "Summary", + "value": item.get("value", ""), + "detail": item.get("detail"), + }) + else: + cards.append({"label": "Summary", "value": item}) + return cards[:_MAX_SUMMARY_CARDS] + + +def _normalize_stages(value: Any) -> list[dict[str, Any]]: + stages: list[dict[str, Any]] = [] + for item in _as_sequence(value): + if isinstance(item, Mapping): + stages.append({ + "name": item.get("name") or "stage", + "status": item.get("status", "recorded"), + "reason": item.get("reason", ""), + "duration": item.get("duration"), + }) + else: + stages.append({"name": item, "status": "recorded"}) + return stages + + +def _normalize_retrievals(value: Any) -> list[dict[str, Any]]: + retrievals: list[dict[str, Any]] = [] + for item in _as_sequence(value): + if isinstance(item, Mapping): + hits = item.get("hits") + retrievals.append({ + "query": item.get("query", ""), + "tool": item.get("tool", ""), + "reason": item.get("reason", ""), + "hits": _as_sequence(hits), + }) + else: + retrievals.append({"query": item, "hits": []}) + return retrievals + + +def _normalize_nodes(value: Any, *, dep_graph: bool = False) -> list[dict[str, Any]]: + nodes: list[dict[str, Any]] = [] + id_key = "dep_node_id" if dep_graph else "node_id" + for item in _as_sequence(value): + if isinstance(item, Mapping): + entry = dict(item) + else: + entry = {id_key: item} + nodes.append(entry) + return nodes + + +def _normalize_code_deltas(value: Any) -> list[dict[str, Any]]: + deltas: list[dict[str, Any]] = [] + for item in _as_sequence(value): + if isinstance(item, Mapping): + deltas.append({ + "file": item.get("file") or item.get("path") or "", + "change_type": item.get("change_type") or item.get("status") or "", + "before": item.get("before"), + "after": item.get("after"), + "diff": item.get("diff", ""), + }) + else: + deltas.append({"file": item, "change_type": "recorded", "diff": ""}) + return deltas + + +def _normalize_focused_graph(value: Any) -> dict[str, Any]: + if value in (None, "", [], {}): + return {} + if isinstance(value, Mapping): + return dict(value) + return {"detail": value} + + +def _normalize_artifacts(value: Any) -> list[dict[str, Any]]: + artifacts: list[dict[str, Any]] = [] + for item in _as_sequence(value): + if isinstance(item, Mapping): + path = item.get("path") + artifacts.append({ + "label": item.get("label") or path or "artifact", + "path": path, + "status": _artifact_status(path, item.get("status")), + "detail": item.get("detail"), + }) + else: + artifacts.append({"label": Path(str(item)).name or "artifact", "path": item, "status": _artifact_status(item)}) + return artifacts + + +def _artifact_status(path: Any, status: Any = None) -> Any: + if status not in (None, ""): + return status + if path in (None, ""): + return "missing" + try: + return "available" if Path(str(path)).expanduser().exists() else "missing" + except (OSError, ValueError): + return "missing" + + +def _normalize_verification(value: Any) -> list[dict[str, Any]]: + checks: list[dict[str, Any]] = [] + for item in _as_sequence(value): + if isinstance(item, Mapping): + checks.append({ + "name": item.get("name") or "verification", + "status": item.get("status", ""), + "detail": item.get("detail"), + }) + else: + checks.append({"name": "verification", "status": item}) + return checks + + +def _normalize_user_decisions(value: Any) -> list[dict[str, Any]]: + decisions: list[dict[str, Any]] = [] + for item in _as_sequence(value): + if isinstance(item, Mapping): + entry = dict(item) + else: + entry = {"decision": item} + decisions.append({ + "decision": entry.get("decision", ""), + "before_state": entry.get("before_state"), + "confirmed": entry.get("confirmed"), + "branch": entry.get("branch", ""), + "apply_status": entry.get("apply_status", ""), + "test_status": entry.get("test_status", ""), + "rollback_path": entry.get("rollback_path", ""), + }) + return decisions + + +def _render_page( + *, + title: str, + command: str, + generated_at: str, + status: str | None, + summary_cards: list[dict[str, Any]], + stages: list[dict[str, Any]], + retrievals: list[dict[str, Any]], + rpg_nodes: list[dict[str, Any]], + dep_nodes: list[dict[str, Any]], + code_deltas: list[dict[str, Any]], + focused_graph: dict[str, Any], + artifacts: list[dict[str, Any]], + verification: list[dict[str, Any]], + user_decisions: list[dict[str, Any]], + evidence: Mapping[str, Any], +) -> str: + status_html = f"{_h(status)}" if status else "" + return f""" + + + + +{_h(title)} + + + +
+
+

{_h(title)}

+
Command: {_h(command)}Generated: {_h(generated_at)}{status_html}
+
+{_render_summary_cards(summary_cards)} +{_render_timeline(stages)} +{_render_safety_boundary(user_decisions)} +{_render_verification(verification)} +{_render_retrievals(retrievals)} +{_render_node_table("Focused RPG node evidence", rpg_nodes)} +{_render_node_table("Focused dependency node evidence", dep_nodes)} +{_render_code_deltas(code_deltas)} +{_render_focused_graph(focused_graph)} +{_render_artifacts(artifacts)} +{_render_evidence(evidence)} +
+ + +""" + + +def _render_summary_cards(cards: list[dict[str, Any]]) -> str: + if not cards: + body = "

No summary cards recorded.

" + else: + rendered_cards = [] + for card in cards: + detail = card.get("detail") + detail_html = "" + if detail is not None: + detail_html = f"
{_h(detail)}
" + value = card.get("value", "") + long_value = len(str(value)) > 48 + card_class = "card card-wide" if long_value else "card" + value_class = "card-value card-value-long" if long_value else "card-value" + rendered_cards.append( + f"
{_h(card.get('label', 'Summary'))}
" + f"
{_h(value)}
" + f"{detail_html}
" + ) + body = '
' + "".join(rendered_cards) + "
" + return f"

Summary

{body}
" + + +def _render_timeline(stages: list[dict[str, Any]]) -> str: + if not stages: + body = "

No stages recorded.

" + else: + items = [] + for stage in stages: + duration = stage.get("duration") + duration_text = f"{_h(duration)}s" if duration not in (None, "") else "" + items.append( + "
  • " + f"
    {_h(stage.get('name', 'stage'))}" + f"{_h(stage.get('status', 'recorded'))}{duration_text}
    " + f"
    {_h(stage.get('reason', ''))}
    " + "
    • " + ) + body = '
      ' + "".join(items) + "
    " + return f"

    Stage timeline

    {body}
    " + + +def _render_verification(checks: list[dict[str, Any]]) -> str: + if not checks: + body = "

    No verification status recorded.

    " + else: + rows = [] + for check in checks: + rows.append( + "" + f"{_h(check.get('name') or 'verification')}" + f"{_h(check.get('status', ''))}" + f"{_h(check.get('detail', ''))}" + "" + ) + body = "" + "".join(rows) + "
    CheckStatusDetail
    " + return f"

    Verification status

    {body}
    " + + +def _render_safety_boundary(decisions: list[dict[str, Any]]) -> str: + if not decisions: + return "" + rows = [] + for decision in decisions: + confirmed = decision.get("confirmed") + if confirmed is True: + confirmation = "confirmed" + elif confirmed is False: + confirmation = "not confirmed" + else: + confirmation = "" + rows.append( + "" + f"{_h(decision.get('decision', ''))}" + f"{_h(decision.get('before_state'))}" + f"{_h(confirmation)}" + f"{_h(decision.get('branch', ''))}" + f"{_h(decision.get('apply_status', ''))}" + f"{_h(decision.get('test_status', ''))}" + f"{_h(decision.get('rollback_path', ''))}" + "" + ) + body = ( + "" + "" + "" + + "".join(rows) + + "
    DecisionBefore stateConfirmationBranchApply statusTest statusRollback path
    " + ) + return f"

    Safety boundary

    {body}
    " + + +def _render_retrievals(retrievals: list[dict[str, Any]]) -> str: + if not retrievals: + return "" + rows = [] + for retrieval in retrievals: + hits = retrieval.get("hits") or [] + hit_items = [] + for hit in hits: + if isinstance(hit, Mapping): + label = hit.get("node_id") or hit.get("dep_node_id") or hit.get("path") or hit.get("file") or hit.get("name") or "hit" + reason = hit.get("reason") or hit.get("score") or hit.get("status") or "" + hit_items.append(f"
  • {_h(label)} {_h(reason)}
    • ") + else: + hit_items.append(f"
  • {_h(hit)}
    • ") + hits_html = "No hits recorded." + if hit_items: + hits_html = '" + rows.append( + "" + f"{_h(retrieval.get('tool', ''))}" + f"{_h(retrieval.get('query', ''))}" + f"{_h(retrieval.get('reason', ''))}" + f"{hits_html}" + "" + ) + body = "" + "".join(rows) + "
    ToolQueryReasonHits
    " + return f"

    Retrieval evidence

    {body}
    " + + +def _render_code_deltas(deltas: list[dict[str, Any]]) -> str: + if not deltas: + return "" + blocks = [] + for delta in deltas: + diff = delta.get("diff", "") + diff_html = "

    No diff recorded.

    " + if diff: + diff_html = f"
    View diff
    {_h(diff)}
    " + before_after = "" + if delta.get("before") is not None or delta.get("after") is not None: + before_after = ( + "
    Before/after" + f"
    {_h({'before': delta.get('before'), 'after': delta.get('after')})}
    " + "
    " + ) + blocks.append( + "
    " + "
    " + f"{_h(delta.get('file', ''))}" + f"{_h(delta.get('change_type', ''))}" + "
    " + f"{diff_html}{before_after}" + "
    " + ) + return f"

    Code deltas

    {''.join(blocks)}
    " + + +def _render_focused_graph(focused_graph: dict[str, Any]) -> str: + if not focused_graph: + return "" + path = focused_graph.get("path") or focused_graph.get("artifact_path") or focused_graph.get("html_path") + href = _artifact_href(path) if path else "#" + rows = [ + ("Status", focused_graph.get("status", "recorded"), False), + ("Graph artifact", f"{_h(path or '')}" if path else "", True), + ("Selected RPG nodes", ", ".join(str(v) for v in focused_graph.get("selected_rpg_nodes") or focused_graph.get("rpg_node_ids") or []), False), + ("Selected dependency nodes", ", ".join(str(v) for v in focused_graph.get("selected_dep_nodes") or focused_graph.get("dep_node_ids") or []), False), + ("Included RPG nodes", focused_graph.get("rpg_node_count") or focused_graph.get("rpg_nodes") or "", False), + ("Included dependency nodes", focused_graph.get("dep_node_count") or focused_graph.get("dep_nodes") or "", False), + ] + table_rows = [] + for label, value, is_html in rows: + if value in (None, ""): + continue + rendered = str(value) if is_html else _h(value) + table_rows.append(f"{_h(label)}{rendered}") + table = "" + "".join(table_rows) + "
    " if table_rows else "

    No focused graph metadata recorded.

    " + metadata = json.dumps(focused_graph, indent=2, ensure_ascii=False, default=_json_default) + inspector = f"
    Inspector metadata
    {_h(metadata)}
    " + return f"

    Focused graph evidence

    {table}{inspector}
    " + + +def _render_node_table(title: str, nodes: list[dict[str, Any]]) -> str: + if not nodes: + body = "

    No node evidence recorded.

    " + elif any("dep_node_id" in node for node in nodes): + rows = [] + for node in nodes: + rows.append( + "" + f"{_h(node.get('dep_node_id', ''))}" + f"{_h(node.get('path', ''))}" + f"{_h(node.get('source_feature', ''))}" + f"{_h(node.get('change', ''))}" + "" + ) + body = "" + "".join(rows) + "
    IDPathSource featureChange
    " + else: + rows = [] + for node in nodes: + rows.append( + "" + f"{_h(node.get('node_id', ''))}" + f"{_h(node.get('name', ''))}" + f"{_h(node.get('type', ''))}" + f"{_h(node.get('path', ''))}" + f"{_h(node.get('score', ''))}" + "" + ) + body = "" + "".join(rows) + "
    IDNameTypePathScore/status
    " + return f"

    {_h(title)}

    {body}
    " + + +def _render_artifacts(artifacts: list[dict[str, Any]]) -> str: + if not artifacts: + body = "

    No artifact links recorded.

    " + else: + rows = [] + for artifact in artifacts: + path = artifact.get("path") + href = _artifact_href(path) + rows.append( + "" + f"{_h(artifact.get('label', 'artifact'))}" + f"{_h(path or '')}" + f"{_h(_artifact_status(path, artifact.get('status')))}" + "" + ) + body = "" + "".join(rows) + "
    ArtifactPathStatus
    " + return f"

    Artifact links

    {body}
    " + + +def _render_evidence(evidence: Mapping[str, Any]) -> str: + data = json.dumps(evidence, indent=2, ensure_ascii=False, default=_json_default) + return f"
    Evidence JSON
    {_h(data)}
    " + + +def _artifact_href(path: Any) -> str: + if path is None: + return "#" + try: + return Path(str(path)).expanduser().resolve().as_uri() + except Exception: + return "file://" + quote(str(path), safe="/._-~") + + +def _json_default(value: Any) -> Any: + if isinstance(value, Path): + return str(value) + if hasattr(value, "to_dict"): + return value.to_dict() + if hasattr(value, "__dict__"): + return value.__dict__ + return str(value) + + +def _h(value: Any) -> str: + if value is None: + return "" + if isinstance(value, (dict, list, tuple)): + value = json.dumps(value, ensure_ascii=False, default=_json_default) + return escape(str(value), quote=False) + + +def _h_attr(value: Any) -> str: + if value is None: + return "" + return escape(str(value), quote=True) diff --git a/CoderMind/scripts/common/session_manager.py b/CoderMind/scripts/common/session_manager.py index df8fd5a..daa0a23 100644 --- a/CoderMind/scripts/common/session_manager.py +++ b/CoderMind/scripts/common/session_manager.py @@ -225,7 +225,7 @@ class NullSessionManager(SessionManager): """ def before(self, ctx: TraceContext, prompt: str) -> None: - pass + return None def after(self, purpose: str) -> Optional[Path]: return None diff --git a/CoderMind/scripts/common/utils.py b/CoderMind/scripts/common/utils.py index 1dd1874..a4f3809 100644 --- a/CoderMind/scripts/common/utils.py +++ b/CoderMind/scripts/common/utils.py @@ -857,9 +857,6 @@ def get_skeleton( class _CompressTransformer(cst.CSTTransformer): """Replace function bodies with ``...`` while preserving structure.""" - def __init__(self): - pass - def _is_import_stmt(self, stmt: cst.CSTNode) -> bool: if not m.matches(stmt, m.SimpleStatementLine()): return False diff --git a/CoderMind/scripts/plan.py b/CoderMind/scripts/plan.py index 90c2cd5..a6ad4aa 100644 --- a/CoderMind/scripts/plan.py +++ b/CoderMind/scripts/plan.py @@ -58,7 +58,20 @@ # Sub-scripts live in the same directory as this file (bundled under # cmind_cli/core_pack/scripts/ in the installed wheel). _SCRIPTS_DIR = Path(__file__).resolve().parent - +if str(_SCRIPTS_DIR) not in sys.path: + sys.path.insert(0, str(_SCRIPTS_DIR)) + +from common.paths import ( + BASE_CLASSES_FILE, + DATA_FLOW_FILE, + DATA_FLOW_VIZ_FILE, + INTERFACES_FILE, + SKELETON_FILE, + SKELETON_SUMMARY_FILE, + TASKS_FILE, +) +from common.run_events import ArtifactEvent, CommandRun, StepEvent, VerificationEvent +from common.run_report import write_command_report # --------------------------------------------------------------------------- # Stage table — single source of truth for the pipeline. @@ -278,8 +291,9 @@ def decide(states: list[StageState], force: bool) -> None: state.reason = "up-to-date" else: state.will_run = True - if state.type == "warning": - state.reason = "warning: cross-stage contract violation; rebuild stage and downstream" + # Preserve useful messages for warnings (don't overwrite state.message) + if state.message: + state.reason = state.message else: state.reason = f"type={state.type}" cascade = True @@ -292,6 +306,96 @@ def decide(states: list[StageState], force: bool) -> None: _GLYPH = {"update": "✓", "init": "·", "warning": "!", "error": "✗"} +def _stage_rows(states: list[StageState]) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for state in states: + rows.append({ + "name": state.stage.name, + "status": "run" if state.will_run else "skip", + "type": state.type, + "done": state.done, + "reason": state.reason or state.message, + "message": state.message, + }) + return rows + + +def _plan_artifacts() -> list[dict[str, Any]]: + paths = { + "skeleton": SKELETON_FILE, + "skeleton_summary": SKELETON_SUMMARY_FILE, + "data_flow": DATA_FLOW_FILE, + "data_flow_viz": DATA_FLOW_VIZ_FILE, + "base_classes": BASE_CLASSES_FILE, + "interfaces": INTERFACES_FILE, + "tasks": TASKS_FILE, + } + return [ + {"label": label, "path": str(path), "status": "available" if path.exists() else "missing"} + for label, path in paths.items() + ] + + +def _write_plan_report( + states: list[StageState], + *, + mode: str, + elapsed: float | None = None, + post_steps: list[dict[str, Any]] | None = None, +) -> str | None: + try: + done = sum(1 for s in states if s.done) + runnable = sum(1 for s in states if s.will_run) + cards = [ + {"label": "mode", "value": mode}, + {"label": "stages", "value": len(states)}, + {"label": "done", "value": done}, + {"label": "to run", "value": runnable}, + {"label": "skipped", "value": len(states) - runnable}, + ] + if elapsed is not None: + cards.append({"label": "elapsed", "value": f"{elapsed:.1f}s"}) + if post_steps is not None: + cards.append({"label": "post steps", "value": len(post_steps)}) + stage_rows = _stage_rows(states) + artifact_rows = _plan_artifacts() + steps = [ + StepEvent(name=row["name"], status=row["status"], reason=row.get("reason", "")) + for row in stage_rows + ] + for post in post_steps or []: + steps.append(StepEvent( + name=post.get("name", "post-step"), + status="ok" if post.get("returncode") == 0 else "warning", + reason=f"exit {post.get('returncode')}", + )) + report_path = write_command_report(CommandRun( + command="plan", + title="CoderMind plan Explain View", + status=mode, + summary=cards, + steps=steps, + artifacts=[ + ArtifactEvent(label=row["label"], path=row["path"], status=row.get("status")) + for row in artifact_rows + ], + verification=[ + VerificationEvent(name=s.stage.name, status=s.type, detail=s.message or s.reason) + for s in states + ], + evidence={ + "mode": mode, + "elapsed": elapsed, + "stages": stage_rows, + "post_steps": post_steps or [], + "artifacts": artifact_rows, + }, + )) + return str(report_path) + except Exception: + return None + + def _format_table(states: list[StageState]) -> str: rows = ["Stage Type Done Action"] rows.append("-" * 50) @@ -322,6 +426,7 @@ def _emit_check_only_json(states: list[StageState]) -> None: done = sum(1 for s in states if s.done) total = len(states) next_pending = next((s.stage.name for s in states if not s.done), None) + report_path = _write_plan_report(states, mode="check-only") payload = { "total": total, "done": done, @@ -332,10 +437,14 @@ def _emit_check_only_json(states: list[StageState]) -> None: "type": s.type, "message": s.message, "done": s.done, + "will_run": s.will_run, + "reason": s.reason, } for s in states ], } + if report_path: + payload["report_path"] = report_path print(json.dumps(payload, indent=2)) @@ -436,6 +545,9 @@ def main(argv: Optional[list[str]] = None) -> int: _emit_check_only_json(states) else: _print_probe_summary(states) + report_path = _write_plan_report(states, mode="check-only") + if report_path: + print(f"Report: {report_path}") return 0 # --- Step: prerequisite check ----------------------------------------- @@ -521,15 +633,20 @@ def main(argv: Optional[list[str]] = None) -> int: # --- Step: post-pipeline helpers -------------------------------------- print() print("Running post-pipeline helpers ...") + post_results: list[dict[str, Any]] = [] for post in POST_STEPS: print(f"▶ {post}") rc = _run_stage(invoker, post, []) + post_results.append({"name": post, "returncode": rc}) if rc != 0: print(f" warning: {post} exited with {rc} (continuing)") total_elapsed = time.monotonic() - started + report_path = _write_plan_report(states, mode="complete", elapsed=total_elapsed, post_steps=post_results) print() print(f"Plan complete in {total_elapsed:.1f}s.") + if report_path: + print(f"Report: {report_path}") print("Next: `/cmind.code_gen` to generate source code.") print("Graph: see the 'Writing visualization to:' line above for the generated HTML path.") return 0 diff --git a/CoderMind/scripts/rpg/dep_graph.py b/CoderMind/scripts/rpg/dep_graph.py index 093228f..5397895 100644 --- a/CoderMind/scripts/rpg/dep_graph.py +++ b/CoderMind/scripts/rpg/dep_graph.py @@ -214,6 +214,7 @@ def __init__(self, repo_dir: str): _DERIVED_FIELDS = frozenset({ "imports_from", "calls", "called_by", "inherits", "inherited_by", + "rpg_nodes", }) # ------------------------------------------------------------------ diff --git a/CoderMind/scripts/rpg/graph_query.py b/CoderMind/scripts/rpg/graph_query.py index 734d07f..cc119a6 100644 --- a/CoderMind/scripts/rpg/graph_query.py +++ b/CoderMind/scripts/rpg/graph_query.py @@ -135,18 +135,26 @@ def from_files(cls, rpg_path: str, dep_graph_path: str = "") -> "GraphQueryEngin # Helpers # ------------------------------------------------------------------ - def _normalize_path(self, meta_path: str) -> str: - """Strip redundant code_dir prefix from RPG meta.path. - - For legacy data where ``_dep_graph_code_dir`` was ``"repo"``, - this converts e.g. ``repo/routes/auth.py`` to ``routes/auth.py`` - so it aligns with dep_graph node IDs. In the unified - workspace==repo layout the prefix is empty and this is a no-op. - """ + def _normalize_path(self, meta_path: Any) -> Any: + """Strip redundant code_dir prefix from RPG meta.path.""" + if isinstance(meta_path, list): + return [self._normalize_path(path) for path in meta_path] + if not isinstance(meta_path, str): + return meta_path if self._code_dir_prefix and meta_path.startswith(self._code_dir_prefix): return meta_path[len(self._code_dir_prefix):] return meta_path + def _path_search_text(self, meta_path: Any) -> str: + normalized_path = self._normalize_path(meta_path) + if isinstance(normalized_path, list): + return " ".join( + self._path_search_text(path) for path in normalized_path + ) + if isinstance(normalized_path, str): + return normalized_path.split(":", 1)[0].lower() + return str(normalized_path).lower() if normalized_path else "" + def _get_feature_path(self, node_id: str) -> str: """Build ancestor chain for an RPG node (excluding repo root).""" parts: list[str] = [] @@ -306,8 +314,8 @@ def _search_rpg_tree(self, query: str, top_k: int) -> List[Dict[str, Any]]: elif query in name.lower(): score = 75 else: - file_part = meta_path.split(":")[0] if ":" in meta_path else meta_path - if query in file_part.lower(): + path_text = self._path_search_text(meta_path) + if path_text and query in path_text: score = 60 elif query in nid.lower(): score = 55 diff --git a/CoderMind/scripts/rpg_edit/apply.py b/CoderMind/scripts/rpg_edit/apply.py index 2421576..cb6d1c5 100644 --- a/CoderMind/scripts/rpg_edit/apply.py +++ b/CoderMind/scripts/rpg_edit/apply.py @@ -8,6 +8,7 @@ import argparse import json +import shlex import shutil import subprocess import sys @@ -21,7 +22,15 @@ if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) -from common.paths import REPO_RPG_FILE, DEP_GRAPH_FILE, REPO_DIR, RPG_EDIT_PLAN_FILE # noqa: E402 +from common.paths import ( # noqa: E402 + REPO_RPG_FILE, + DEP_GRAPH_FILE, + REPO_DIR, + RPG_EDIT_PLAN_FILE, + RPG_EDIT_APPLY_RESULT_FILE, + cmd_for, +) +from common.git_utils import read_head # noqa: E402 def _backup(rpg_path: Path, dep_graph_path: Path, ts: str) -> Dict[str, str]: @@ -50,6 +59,61 @@ def _rollback(backups: Dict[str, str], rpg_path: Path, dep_graph_path: Path) -> shutil.copy2(backups["dep_graph"], dep_graph_path) +def _is_rpg_edit_branch(branch: Any) -> bool: + return isinstance(branch, str) and branch.startswith("rpg-edit/") + + +def _rollback_command(timestamp: str | None, before_state: dict | None) -> str | None: + if not timestamp: + return None + command = f"{cmd_for('rpg_edit/apply.py')} --rollback {shlex.quote(str(timestamp))}" + branch = before_state.get("head_branch") if isinstance(before_state, dict) else None + if _is_rpg_edit_branch(branch): + command += f" --rollback-branch {shlex.quote(str(branch))}" + return command + + +def _rollback_path(backups: Dict[str, str]) -> str | None: + return backups.get("rpg") or backups.get("dep_graph") + + +def _persist_apply_result(result: Dict[str, Any]) -> None: + RPG_EDIT_APPLY_RESULT_FILE.parent.mkdir(parents=True, exist_ok=True) + RPG_EDIT_APPLY_RESULT_FILE.write_text( + json.dumps(result, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + +def _record_apply_result( + result: Dict[str, Any], + *, + backup_timestamp: str | None = None, + backups: Dict[str, str] | None = None, + applied_features: list | None = None, + dep_graph_refreshed: bool | None = None, + before_state: dict | None = None, + confirmed: bool | None = None, +) -> Dict[str, Any]: + backups = backups or {} + if backup_timestamp is not None: + result.setdefault("backup_timestamp", backup_timestamp) + result.setdefault("rollback_command", _rollback_command(backup_timestamp, before_state)) + result.setdefault("backups", backups) + result.setdefault("applied_features", applied_features or []) + if dep_graph_refreshed is not None: + result.setdefault("dep_graph_refreshed", dep_graph_refreshed) + rollback_path = _rollback_path(backups) + if rollback_path: + result.setdefault("rollback_path", rollback_path) + if before_state is not None: + result.setdefault("before_state", before_state) + if confirmed is not None: + result.setdefault("confirmed", confirmed) + _persist_apply_result(result) + return result + + def apply_feature_changes(svc, changes: list) -> list: """Apply feature_changes to the RPG in memory. @@ -128,6 +192,8 @@ def main(): parser.add_argument("--backup-ts", type=str, default=None, help="Reuse existing backup timestamp (skip new backup)") parser.add_argument("--skip-tests", action="store_true") + parser.add_argument("--confirmed", action="store_const", const=True, default=None, + help="Record that the apply boundary was explicitly confirmed") parser.add_argument("--rollback", type=str, default=None, help="Rollback to a previous timestamp backup") parser.add_argument("--rollback-branch", type=str, default=None, @@ -145,6 +211,8 @@ def main(): from common.logging_setup import setup_file_logging setup_file_logging("rpg_edit") + before_state = read_head(REPO_DIR) + # Handle rollback if args.rollback: rpg_backup = args.rpg.with_suffix(f".before-edit-{args.rollback}.json") @@ -180,10 +248,23 @@ def main(): "message": f"git invocation failed: {exc}", } + rollback_backups = {} + if rpg_backup.exists(): + rollback_backups["rpg"] = str(rpg_backup) + if dg_backup.exists(): + rollback_backups["dep_graph"] = str(dg_backup) result = {"type": "rollback", "restored": restored, "timestamp": args.rollback} if branch_result: result["branch"] = branch_result + _record_apply_result( + result, + backup_timestamp=args.rollback, + backups=rollback_backups, + dep_graph_refreshed=False, + before_state=before_state, + confirmed=args.confirmed, + ) print(json.dumps(result, indent=2) if args.json else f"Rolled back: {restored}" + (f"; branch={branch_result}" if branch_result else "")) @@ -192,6 +273,12 @@ def main(): # Load plan if not args.plan.exists(): result = {"type": "error", "message": f"Plan not found: {args.plan}"} + _record_apply_result( + result, + dep_graph_refreshed=False, + before_state=before_state, + confirmed=args.confirmed, + ) print(json.dumps(result) if args.json else f"Error: {result['message']}") return 1 @@ -214,6 +301,7 @@ def main(): # --- Phase: rpg-only or all → apply feature_changes --- applied_features = [] + dep_graph_refreshed = False if args.phase in ("rpg-only", "all"): feature_changes = plan.get("feature_changes", []) applied_features = apply_feature_changes(svc, feature_changes) if feature_changes else [] @@ -226,12 +314,20 @@ def main(): "backup_timestamp": ts, "backups": backups, } + _record_apply_result( + result, + backup_timestamp=ts, + backups=backups, + applied_features=applied_features, + dep_graph_refreshed=dep_graph_refreshed, + before_state=before_state, + confirmed=args.confirmed, + ) print(json.dumps(result, indent=2) if args.json else f"RPG updated ({len(applied_features)} features). Backup: {ts}") return 0 # --- Phase: dep-refresh or all → refresh dep_graph --- - dep_graph_refreshed = False if args.phase in ("dep-refresh", "all"): # Workspace root is the project repo root. Explicit ``--repo`` # still wins for tests / brownfield setups where the code lives @@ -252,6 +348,15 @@ def main(): "message": f"dep_graph refresh failed: {exc}", "backup_timestamp": ts, } + _record_apply_result( + result, + backup_timestamp=ts, + backups=backups, + applied_features=applied_features, + dep_graph_refreshed=dep_graph_refreshed, + before_state=before_state, + confirmed=args.confirmed, + ) print(json.dumps(result, indent=2) if args.json else f"Error: {result['message']}") return 1 @@ -264,6 +369,15 @@ def main(): "dep_graph_refreshed": dep_graph_refreshed, "backup_timestamp": ts, } + _record_apply_result( + result, + backup_timestamp=ts, + backups=backups, + applied_features=applied_features, + dep_graph_refreshed=dep_graph_refreshed, + before_state=before_state, + confirmed=args.confirmed, + ) print(json.dumps(result, indent=2) if args.json else f"dep_graph refreshed: {dep_graph_refreshed}. Backup: {ts}") return 0 @@ -293,9 +407,19 @@ def main(): "type": "test_failed", "applied_features": applied_features, "test_output": test_result["output"], + "test_result": test_result, "rolled_back": True, "backup_timestamp": ts, } + _record_apply_result( + result, + backup_timestamp=ts, + backups=backups, + applied_features=applied_features, + dep_graph_refreshed=dep_graph_refreshed, + before_state=before_state, + confirmed=args.confirmed, + ) print(json.dumps(result, indent=2) if args.json else f"Tests failed. Rolled back to {ts}.") return 1 @@ -309,6 +433,15 @@ def main(): "backup_timestamp": ts, "backups": backups, } + _record_apply_result( + result, + backup_timestamp=ts, + backups=backups, + applied_features=applied_features, + dep_graph_refreshed=dep_graph_refreshed, + before_state=before_state, + confirmed=args.confirmed, + ) print(json.dumps(result, indent=2) if args.json else "EditPlan applied successfully.") return 0 diff --git a/CoderMind/scripts/rpg_edit/code.py b/CoderMind/scripts/rpg_edit/code.py index c070be7..6d0172e 100644 --- a/CoderMind/scripts/rpg_edit/code.py +++ b/CoderMind/scripts/rpg_edit/code.py @@ -43,6 +43,7 @@ RPG_FILE, REPO_DIR, RPG_EDIT_PLAN_FILE, + RPG_EDIT_CODE_RESULT_FILE, DATA_DIR, WORKSPACE_ROOT, cmd_for, @@ -52,6 +53,14 @@ logger = logging.getLogger(__name__) +def _write_code_result(result: Dict[str, Any]) -> None: + RPG_EDIT_CODE_RESULT_FILE.parent.mkdir(parents=True, exist_ok=True) + RPG_EDIT_CODE_RESULT_FILE.write_text( + json.dumps(result, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + # --------------------------------------------------------------------------- # Prompt templates # --------------------------------------------------------------------------- @@ -584,6 +593,7 @@ def apply_code_changes( fp = c.get("file_path") if fp and fp not in done_files: done_files.append(fp) + last_error = None iter_info["detail"] = None break elif status == "partial": @@ -684,6 +694,7 @@ def main() -> int: max_iterations=args.max_iterations, timeout=args.timeout, ) + _write_code_result(result) if args.json: print(json.dumps(result, indent=2, ensure_ascii=False)) diff --git a/CoderMind/scripts/rpg_edit/locate.py b/CoderMind/scripts/rpg_edit/locate.py index 082b819..43492c6 100644 --- a/CoderMind/scripts/rpg_edit/locate.py +++ b/CoderMind/scripts/rpg_edit/locate.py @@ -20,7 +20,15 @@ if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) -from common.paths import REPO_RPG_FILE # noqa: E402 +from common.paths import REPO_RPG_FILE, RPG_EDIT_LOCATE_FILE # noqa: E402 + + +def _write_locate_result(result: dict) -> None: + RPG_EDIT_LOCATE_FILE.parent.mkdir(parents=True, exist_ok=True) + RPG_EDIT_LOCATE_FILE.write_text( + json.dumps(result, indent=2, ensure_ascii=False), + encoding="utf-8", + ) def _build_tree_summary(svc, max_depth: int = 3, max_lines: int = 150) -> List[str]: @@ -198,6 +206,7 @@ def main(): # when search results are poor (e.g. editing features that don't exist yet). tree_lines = _build_tree_summary(svc) output["tree_summary"] = tree_lines + _write_locate_result(output) if args.json: print(json.dumps(output, indent=2, ensure_ascii=False)) diff --git a/CoderMind/scripts/rpg_edit/review.py b/CoderMind/scripts/rpg_edit/review.py index 85b8386..ab8bc0e 100644 --- a/CoderMind/scripts/rpg_edit/review.py +++ b/CoderMind/scripts/rpg_edit/review.py @@ -34,10 +34,455 @@ if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) -from common.paths import REPO_DIR, cmd_for, RPG_EDIT_PLAN_FILE, RPG_EDIT_IMPACT_FILE # noqa: E402 +from common.paths import ( # noqa: E402 + REPO_DIR, + REPORTS_DIR, + REPO_RPG_FILE, + cmd_for, + RPG_EDIT_PLAN_FILE, + RPG_EDIT_IMPACT_FILE, + RPG_EDIT_VALIDATE_FILE, + RPG_EDIT_LOCATE_FILE, + RPG_EDIT_CODE_RESULT_FILE, + RPG_EDIT_APPLY_RESULT_FILE, + RPG_EDIT_REVIEW_RESULT_FILE, +) +from common.run_events import ( # noqa: E402 + ArtifactEvent, + CodeDeltaEvent, + CommandRun, + DepGraphDeltaEvent, + RPGDeltaEvent, + RetrievalEvent, + StepEvent, + UserDecisionEvent, + VerificationEvent, +) +from common.git_utils import file_diffs_between, read_head # noqa: E402 +from common.run_report import write_command_report # noqa: E402 logger = logging.getLogger(__name__) + +def _write_review_result(result: Dict[str, Any]) -> None: + RPG_EDIT_REVIEW_RESULT_FILE.parent.mkdir(parents=True, exist_ok=True) + RPG_EDIT_REVIEW_RESULT_FILE.write_text( + json.dumps(result, indent=2, ensure_ascii=False), + encoding="utf-8", + ) + + +def _load_json_artifact(path: Optional[Path]) -> Any: + if path is None or not path.exists(): + return None + try: + return json.loads(path.read_text(encoding="utf-8")) + except Exception as exc: + return {"_error": str(exc), "_path": str(path)} + + +def _load_review_artifacts(plan_path: Path, impact_path: Optional[Path]) -> Dict[str, Any]: + return { + "validate": _load_json_artifact(RPG_EDIT_VALIDATE_FILE), + "locate": _load_json_artifact(RPG_EDIT_LOCATE_FILE), + "plan": _load_json_artifact(plan_path), + "impact": _load_json_artifact(impact_path), + "code_result": _load_json_artifact(RPG_EDIT_CODE_RESULT_FILE), + "apply_result": _load_json_artifact(RPG_EDIT_APPLY_RESULT_FILE), + } + + +def _artifact_links(plan_path: Path, impact_path: Optional[Path]) -> List[Dict[str, Any]]: + paths = { + "validate": RPG_EDIT_VALIDATE_FILE, + "locate": RPG_EDIT_LOCATE_FILE, + "plan": plan_path, + "impact": impact_path, + "code_result": RPG_EDIT_CODE_RESULT_FILE, + "apply_result": RPG_EDIT_APPLY_RESULT_FILE, + "review_result": RPG_EDIT_REVIEW_RESULT_FILE, + } + links: List[Dict[str, Any]] = [] + for label, path in paths.items(): + if path is None: + continue + status = "available" if path.exists() or label == "review_result" else "missing" + links.append({"label": label, "path": str(path), "status": status}) + return links + + +def _impact_results(artifacts: Dict[str, Any]) -> Dict[str, Any]: + impact = artifacts.get("impact") if isinstance(artifacts.get("impact"), dict) else {} + results = impact.get("results") if isinstance(impact.get("results"), dict) else {} + return results + + +def _selected_candidate_rows(artifacts: Dict[str, Any]) -> List[Dict[str, Any]]: + locate = artifacts.get("locate") if isinstance(artifacts.get("locate"), dict) else {} + plan = artifacts.get("plan") if isinstance(artifacts.get("plan"), dict) else {} + affected = [node_id for node_id in plan.get("affected_nodes") or [] if node_id] + candidates = [c for c in locate.get("results") or [] if isinstance(c, dict)] + if not affected: + return candidates + + impact_results = _impact_results(artifacts) + candidates_by_id = {c.get("node_id"): c for c in candidates if c.get("node_id") in affected} + rows: List[Dict[str, Any]] = [] + for node_id in affected: + impact = impact_results.get(node_id) if isinstance(impact_results.get(node_id), dict) else {} + candidate = dict(candidates_by_id.get(node_id) or {"node_id": node_id}) + if impact: + candidate.setdefault("name", impact.get("name", "")) + if not candidate.get("dep_nodes"): + candidate["dep_nodes"] = impact.get("dep_nodes") or [] + if not candidate.get("status") and (impact.get("error") or impact.get("message")): + candidate["status"] = impact.get("error") or impact.get("message") + rows.append(candidate) + return rows + + +def _listify(value: Any) -> List[Any]: + if value in (None, ""): + return [] + if isinstance(value, (list, tuple, set)): + return [item for item in value if item not in (None, "")] + return [value] + + +def _retrieval_hit_reason(candidate: Dict[str, Any], impact: Dict[str, Any]) -> str: + parts: List[str] = [] + if candidate.get("score") not in (None, ""): + parts.append(f"locate score={candidate.get('score')}") + if candidate.get("feature_path"): + parts.append(f"feature path={candidate.get('feature_path')}") + dep_count = len(_listify(candidate.get("dep_nodes"))) + if dep_count: + parts.append(f"{dep_count} dep nodes") + if impact.get("error"): + parts.append(f"impact error={impact.get('error')}") + if impact.get("message"): + parts.append(str(impact.get("message"))) + summary = impact.get("impact_summary") if isinstance(impact.get("impact_summary"), dict) else {} + callers = summary.get("total_callers", len(impact.get("callers") or [])) + files = summary.get("affected_file_count", len(impact.get("affected_files") or [])) + if callers or files: + parts.append(f"impact callers={callers}, affected_files={files}") + return "; ".join(parts) or "selected by review plan" + + +def _retrieval_rows(artifacts: Dict[str, Any], candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + rows: List[Dict[str, Any]] = [] + locate = artifacts.get("locate") if isinstance(artifacts.get("locate"), dict) else {} + impact_results = _impact_results(artifacts) + if locate or candidates: + hits: List[Dict[str, Any]] = [] + for candidate in candidates: + node_id = candidate.get("node_id") + impact = impact_results.get(node_id) if isinstance(impact_results.get(node_id), dict) else {} + hits.append({ + "node_id": node_id, + "name": candidate.get("name"), + "path": candidate.get("path") or candidate.get("meta_path"), + "score": candidate.get("score"), + "reason": _retrieval_hit_reason(candidate, impact), + }) + rows.append({ + "query": locate.get("query", ""), + "tool": str(RPG_EDIT_LOCATE_FILE), + "reason": f"{len(hits)} selected from {len(locate.get('results') or [])} locate candidates", + "hits": hits, + }) + if impact_results: + hits = [] + for node_id, impact in impact_results.items(): + impact = impact if isinstance(impact, dict) else {} + hits.append({ + "node_id": node_id, + "name": impact.get("name"), + "reason": _retrieval_hit_reason({"node_id": node_id}, impact), + }) + rows.append({ + "query": ", ".join(str(node_id) for node_id in impact_results), + "tool": str(RPG_EDIT_IMPACT_FILE), + "reason": f"{len(impact_results)} impact result sets", + "hits": hits, + }) + return rows + + +def _code_delta_rows(artifacts: Dict[str, Any]) -> List[Dict[str, Any]]: + code_result = artifacts.get("code_result") if isinstance(artifacts.get("code_result"), dict) else {} + files = [str(path) for path in _listify(code_result.get("files_modified"))] + commit_sha = code_result.get("commit_sha") + rows: List[Dict[str, Any]] = [] + if commit_sha: + try: + rows = file_diffs_between( + REPO_DIR, + to_commit=str(commit_sha), + files=files or None, + py_only=False, + ) + except Exception as exc: + rows = [{"file": path, "change_type": "modify", "diff": "", "error": str(exc)} for path in files] + seen = {row.get("file") for row in rows} + for path in files: + if path not in seen: + rows.append({"file": path, "change_type": "modify", "diff": ""}) + return rows + + +def _focused_graph_output_path() -> Path: + REPORTS_DIR.mkdir(parents=True, exist_ok=True) + return REPORTS_DIR / f"rpg_edit_focused_graph_{time.time_ns()}.html" + + +def _focused_graph_artifact(candidates: List[Dict[str, Any]], artifacts: Dict[str, Any]) -> Dict[str, Any]: + dep_rows = _dep_node_rows(candidates) + selected_rpg = sorted({str(row.get("node_id")) for row in candidates if row.get("node_id")}) + selected_dep = sorted({str(row.get("node_id")) for row in dep_rows if row.get("node_id")}) + if not selected_rpg and not selected_dep: + return {} + + metadata: Dict[str, Any] = { + "status": "recorded", + "selected_rpg_nodes": selected_rpg, + "selected_dep_nodes": selected_dep, + } + rpg_data = _load_json_artifact(REPO_RPG_FILE) + if not isinstance(rpg_data, dict): + metadata.update({"status": "missing", "reason": f"RPG file not available: {REPO_RPG_FILE}"}) + return metadata + + try: + from rpg_visualize import build_focused_graph_data, generate_html + + focused_data = build_focused_graph_data( + rpg_data, + rpg_node_ids=selected_rpg, + dep_node_ids=selected_dep, + ) + focused_meta = focused_data.get("_focused_graph") if isinstance(focused_data.get("_focused_graph"), dict) else {} + metadata.update(focused_meta) + if not (focused_meta.get("matched_rpg_nodes") or focused_meta.get("matched_dep_nodes")): + metadata.update({"status": "unavailable", "reason": "selected nodes not found in current RPG"}) + return metadata + graph_path = _focused_graph_output_path() + graph_path.write_text(generate_html(focused_data), encoding="utf-8") + metadata.update({"status": "available", "path": str(graph_path)}) + except Exception as exc: + metadata.update({"status": "error", "reason": str(exc)}) + return metadata + + +def _dep_node_path(dep_id: Any) -> str: + if dep_id in (None, ""): + return "" + dep_id_text = str(dep_id) + return dep_id_text.split(":", 1)[0] if ":" in dep_id_text else dep_id_text + + +def _dep_node_rows(candidates: List[Dict[str, Any]]) -> List[Dict[str, Any]]: + rows: List[Dict[str, Any]] = [] + for candidate in candidates: + for dep_id in candidate.get("dep_nodes") or []: + rows.append({ + "node_id": dep_id, + "source_feature": candidate.get("node_id"), + "path": _dep_node_path(dep_id) or candidate.get("meta_path"), + }) + return rows + + +def _review_summary_cards(result: Dict[str, Any], artifacts: Dict[str, Any]) -> List[Dict[str, Any]]: + plan = artifacts.get("plan") if isinstance(artifacts.get("plan"), dict) else {} + locate = artifacts.get("locate") if isinstance(artifacts.get("locate"), dict) else {} + code_result = artifacts.get("code_result") if isinstance(artifacts.get("code_result"), dict) else {} + return [ + {"label": "review", "value": result.get("type", "review")}, + {"label": "success", "value": result.get("success", result.get("type") == "skipped")}, + {"label": "iterations", "value": len(result.get("iterations") or [])}, + {"label": "files", "value": len(code_result.get("files_modified") or [])}, + {"label": "candidates", "value": len(locate.get("results") or [])}, + {"label": "affected nodes", "value": len(plan.get("affected_nodes") or [])}, + {"label": "suggestions", "value": len(result.get("suggestions") or [])}, + ] + + +def _review_timeline(result: Dict[str, Any], artifacts: Dict[str, Any]) -> List[Dict[str, Any]]: + validate = artifacts.get("validate") if isinstance(artifacts.get("validate"), dict) else None + locate = artifacts.get("locate") if isinstance(artifacts.get("locate"), dict) else None + plan = artifacts.get("plan") if isinstance(artifacts.get("plan"), dict) else None + impact = artifacts.get("impact") if isinstance(artifacts.get("impact"), dict) else None + code_result = artifacts.get("code_result") if isinstance(artifacts.get("code_result"), dict) else None + return [ + {"name": "validate", "status": validate.get("type") if validate else "missing", "reason": validate.get("message", "") if validate else "artifact not found"}, + {"name": "locate", "status": locate.get("type") if locate else "missing", "reason": f"{len(locate.get('results') or [])} candidates" if locate else "artifact not found"}, + {"name": "plan", "status": "available" if plan else "missing", "reason": f"{len(plan.get('code_changes') or [])} code changes" if plan else "artifact not found"}, + {"name": "impact", "status": impact.get("type", "available") if impact else "missing", "reason": f"{len((impact.get('results') or {}))} impact result sets" if impact else "artifact not found"}, + {"name": "code", "status": code_result.get("last_status") if code_result else "missing", "reason": code_result.get("last_error") or f"success={code_result.get('success')}" if code_result else "artifact not found"}, + {"name": "review", "status": result.get("type"), "reason": result.get("reason") or f"success={result.get('success')}"}, + ] + + +def _review_verification(result: Dict[str, Any], artifacts: Dict[str, Any]) -> List[Dict[str, Any]]: + checks: List[Dict[str, Any]] = [] + validate = artifacts.get("validate") if isinstance(artifacts.get("validate"), dict) else None + code_result = artifacts.get("code_result") if isinstance(artifacts.get("code_result"), dict) else None + if validate: + checks.append({"name": "validate", "status": validate.get("type"), "detail": validate.get("message", "")}) + if code_result: + checks.append({"name": "code", "status": code_result.get("success"), "detail": code_result.get("last_error") or code_result.get("last_status")}) + checks.append({"name": "review", "status": result.get("success", result.get("type") == "skipped"), "detail": result.get("reason") or result.get("type")}) + for iteration in result.get("iterations") or []: + checks.append({ + "name": f"review iteration {iteration.get('iteration')}", + "status": iteration.get("post_pytest_passed"), + "detail": iteration.get("agent_detail", ""), + }) + return checks + + +def _status_from_bool(value: Any) -> Optional[str]: + if value is True: + return "passed" + if value is False: + return "failed" + return None + + +def _apply_status(apply_result: Dict[str, Any]) -> Any: + return apply_result.get("type") or apply_result.get("status") or "missing" + + +def _test_status(result: Dict[str, Any], code_result: Dict[str, Any], apply_result: Dict[str, Any]) -> Any: + test_result = apply_result.get("test_result") if isinstance(apply_result.get("test_result"), dict) else {} + status = _status_from_bool(test_result.get("passed")) + if status: + return status + for iteration in reversed(result.get("iterations") or []): + if isinstance(iteration, dict): + status = _status_from_bool(iteration.get("post_pytest_passed")) + if status: + return status + if code_result.get("last_status"): + return code_result.get("last_status") + return _status_from_bool(result.get("success")) + + +def _rollback_path(apply_result: Dict[str, Any]) -> Any: + if apply_result.get("rollback_path"): + return apply_result.get("rollback_path") + backups = apply_result.get("backups") if isinstance(apply_result.get("backups"), dict) else {} + return backups.get("rpg") or backups.get("dep_graph") or apply_result.get("rollback_command") + + +def _user_decision(result: Dict[str, Any], artifacts: Dict[str, Any]) -> UserDecisionEvent: + apply_result = artifacts.get("apply_result") if isinstance(artifacts.get("apply_result"), dict) else {} + code_result = artifacts.get("code_result") if isinstance(artifacts.get("code_result"), dict) else {} + current_head = read_head(REPO_DIR) + before_state = apply_result.get("before_state") if apply_result.get("before_state") else current_head + branch = before_state.get("head_branch") if isinstance(before_state, dict) else None + if not branch and isinstance(current_head, dict): + branch = current_head.get("head_branch") + confirmed = apply_result.get("confirmed") if "confirmed" in apply_result else None + return UserDecisionEvent( + decision="apply", + branch=branch, + before_state=before_state, + rollback_path=_rollback_path(apply_result), + confirmed=confirmed, + apply_status=_apply_status(apply_result), + test_status=_test_status(result, code_result, apply_result), + ) + + +class _ReportPayload: + def __init__(self, run: CommandRun, focused_graph: Dict[str, Any]): + self.run = run + self.focused_graph = focused_graph + + def to_dict(self) -> Dict[str, Any]: + data = self.run.to_dict() + if self.focused_graph: + data["focused_graph"] = self.focused_graph + return data + + +def _publish_review_report(result: Dict[str, Any], plan_path: Path, impact_path: Optional[Path]) -> Dict[str, Any]: + _write_review_result(result) + artifacts = _load_review_artifacts(plan_path, impact_path) + candidates = _selected_candidate_rows(artifacts) + code_deltas = _code_delta_rows(artifacts) + focused_graph = _focused_graph_artifact(candidates, artifacts) + artifact_rows = _artifact_links(plan_path, impact_path) + if focused_graph.get("path"): + artifact_rows.append({"label": "focused_graph", "path": focused_graph["path"], "status": focused_graph.get("status")}) + evidence = {"artifacts": artifacts, "review_result": result} + if focused_graph: + evidence["focused_graph"] = focused_graph + try: + report_run = CommandRun( + command="rpg_edit", + title="CoderMind rpg_edit Explain View", + status=str(result.get("type", "review")), + summary=_review_summary_cards(result, artifacts), + steps=[ + StepEvent(name=row.get("name", "stage"), status=row.get("status"), reason=row.get("reason", "")) + for row in _review_timeline(result, artifacts) + ], + rpg_deltas=[ + RPGDeltaEvent( + node_id=row.get("node_id"), + name=row.get("name"), + type=row.get("type"), + path=row.get("path") or row.get("meta_path"), + score=row.get("score"), + ) + for row in candidates + ], + dep_graph_deltas=[ + DepGraphDeltaEvent( + dep_node_id=row.get("node_id"), + path=row.get("path"), + source_feature=row.get("source_feature"), + change=row.get("change"), + ) + for row in _dep_node_rows(candidates) + ], + retrievals=[ + RetrievalEvent(query=row.get("query"), tool=row.get("tool"), hits=row.get("hits"), reason=row.get("reason")) + for row in _retrieval_rows(artifacts, candidates) + ], + artifacts=[ + ArtifactEvent(label=row["label"], path=row["path"], status=row.get("status")) + for row in artifact_rows + ], + code_deltas=[ + CodeDeltaEvent( + file=row.get("file"), + change_type=row.get("change_type"), + before=row.get("before"), + after=row.get("after"), + diff=row.get("diff"), + ) + for row in code_deltas + ], + verification=[ + VerificationEvent(name=row.get("name", "verification"), status=row.get("status"), detail=row.get("detail")) + for row in _review_verification(result, artifacts) + ], + user_decisions=[_user_decision(result, artifacts)], + evidence=evidence, + ) + report_path = write_command_report(_ReportPayload(report_run, focused_graph)) + result["report_path"] = str(report_path) + except Exception as exc: + result["report_error"] = str(exc) + _write_review_result(result) + return result + + # --------------------------------------------------------------------------- # Review prompt template # --------------------------------------------------------------------------- @@ -546,7 +991,7 @@ def impact_review( all_suggestions.append(s) if all_suggestions: results["suggestions"] = all_suggestions - return results + return _publish_review_report(results, plan_path, impact_path) # --------------------------------------------------------------------------- @@ -583,6 +1028,7 @@ def main(): if not args.plan.exists(): result = {"type": "error", "message": f"Plan not found: {args.plan}"} + result = _publish_review_report(result, args.plan, args.impact) print(json.dumps(result) if args.json else f"Error: {result['message']}") return 1 @@ -599,12 +1045,14 @@ def main(): if total_callers == 0 and affected_files <= 1: result = { "type": "skipped", + "success": True, "reason": f"Impact too small for sub-agent review " f"(callers={total_callers}, files={affected_files}). " f"Agent self-review is sufficient.", } + result = _publish_review_report(result, args.plan, args.impact) print(json.dumps(result, indent=2) if args.json else - f"Skipped: {result['reason']}") + f"Skipped: {result['reason']}\nReport: {result.get('report_path', '')}") return 0 result = impact_review( @@ -615,10 +1063,16 @@ def main(): timeout=args.timeout, ) - print(json.dumps(result, indent=2) if args.json else - f"Review {'PASSED' if result['success'] else 'FAILED'} " - f"({len(result['iterations'])} iterations, " - f"{result['total_duration']:.1f}s)") + if args.json: + print(json.dumps(result, indent=2)) + else: + print( + f"Review {'PASSED' if result['success'] else 'FAILED'} " + f"({len(result['iterations'])} iterations, " + f"{result['total_duration']:.1f}s)" + ) + if result.get("report_path"): + print(f"Report: {result['report_path']}") return 0 if result["success"] else 1 diff --git a/CoderMind/scripts/rpg_edit/validate.py b/CoderMind/scripts/rpg_edit/validate.py index e52f7b0..d098fee 100644 --- a/CoderMind/scripts/rpg_edit/validate.py +++ b/CoderMind/scripts/rpg_edit/validate.py @@ -12,7 +12,15 @@ if str(SCRIPTS_DIR) not in sys.path: sys.path.insert(0, str(SCRIPTS_DIR)) -from common.paths import REPO_RPG_FILE, DEP_GRAPH_FILE # noqa: E402 +from common.paths import REPO_RPG_FILE, DEP_GRAPH_FILE, RPG_EDIT_VALIDATE_FILE # noqa: E402 + + +def _write_validate_result(result: dict) -> None: + RPG_EDIT_VALIDATE_FILE.parent.mkdir(parents=True, exist_ok=True) + RPG_EDIT_VALIDATE_FILE.write_text( + json.dumps(result, indent=2, ensure_ascii=False), + encoding="utf-8", + ) def main(): @@ -31,6 +39,7 @@ def main(): if not args.rpg.exists(): result = {"type": "error", "error_code": "rpg_not_found", "message": f"RPG file not found: {args.rpg}"} + _write_validate_result(result) print(json.dumps(result) if args.json else f"Error: {result['message']}") return 1 @@ -40,6 +49,7 @@ def main(): except Exception as e: result = {"type": "error", "error_code": "rpg_load_failed", "message": f"Failed to load RPG: {e}"} + _write_validate_result(result) print(json.dumps(result) if args.json else f"Error: {result['message']}") return 1 @@ -52,6 +62,7 @@ def main(): "Run /cmind.encode to (re)build it; the embedded " "dep_graph rides inside rpg.json." )} + _write_validate_result(result) print(json.dumps(result) if args.json else f"Error: {result['message']}") return 1 @@ -64,6 +75,7 @@ def main(): "dep_to_rpg": len(svc.rpg._dep_to_rpg_map), "feature_to_dep": len(svc.rpg._feature_to_dep_map), } + _write_validate_result(result) print(json.dumps(result, indent=2) if args.json else f"Ready: {result['nodes']} nodes, dep_graph={'yes' if has_dep_graph else 'no'}") return 0 diff --git a/CoderMind/scripts/rpg_encoder/rpg_evolution.py b/CoderMind/scripts/rpg_encoder/rpg_evolution.py index 5b9507e..18c483b 100644 --- a/CoderMind/scripts/rpg_encoder/rpg_evolution.py +++ b/CoderMind/scripts/rpg_encoder/rpg_evolution.py @@ -758,6 +758,18 @@ def process_diff( save_path=dep_graph_save_path, ) + last_rpg._last_diff_summary = { + "added": 0, + "deleted": 0, + "modified": 0, + "renamed": 0, + } + last_rpg._last_diff_files = { + "added": [], + "deleted": [], + "modified": [], + "renamed": [], + } total_time = time.time() - global_start logger.info( "\nNo changes detected for [%s]. RPG remains unchanged.\n" @@ -795,6 +807,20 @@ def process_diff( save_path=dep_graph_save_path, ) + diff_summary = { + "added": len(add_files), + "deleted": len(deleted_files), + "modified": len(modified_result), + "renamed": 0, + } + ctx["last_rpg"]._last_diff_summary = diff_summary + ctx["last_rpg"]._last_diff_files = { + "added": add_files, + "deleted": deleted_files, + "modified": list(modified_result.keys()), + "renamed": [], + } + # Save results result = { "repo_name": repo_name, @@ -804,11 +830,7 @@ def process_diff( "structure": ctx["last_rpg"].to_dict(), "feature_tree": ctx["last_rpg"].get_functionality_graph(), }, - "diff_summary": { - "added": len(add_files), - "deleted": len(deleted_files), - "modified": len(modified_result), - }, + "diff_summary": diff_summary, } if save_path: diff --git a/CoderMind/scripts/rpg_encoder/run_encode.py b/CoderMind/scripts/rpg_encoder/run_encode.py index d9ed2a0..6c299a1 100644 --- a/CoderMind/scripts/rpg_encoder/run_encode.py +++ b/CoderMind/scripts/rpg_encoder/run_encode.py @@ -27,9 +27,53 @@ from common.paths import RPG_FILE, RPG_HTML_FILE, WORKSPACE_ROOT, ensure_cmind_dir # noqa: E402 from common.rpg_io import atomic_write_rpg # noqa: E402 +from common.run_events import ArtifactEvent, CommandRun, StepEvent, VerificationEvent # noqa: E402 +from common.run_report import write_command_report # noqa: E402 from common.trajectory import Trajectory # noqa: E402 +def _attach_encode_report(result: dict) -> dict: + try: + dep_summary = [] + if result.get("dep_nodes") is not None: + dep_summary.append(f"nodes={result.get('dep_nodes')}") + if result.get("dep_edges") is not None: + dep_summary.append(f"edges={result.get('dep_edges')}") + if result.get("dep_to_rpg_map_size") is not None: + dep_summary.append(f"mapped={result.get('dep_to_rpg_map_size')}") + report_path = write_command_report(CommandRun( + command="encode", + title="CoderMind encode Explain View", + status=result.get("status"), + summary=[ + {"label": "repo", "value": result.get("repo_name", "unknown")}, + {"label": "RPG nodes", "value": result.get("node_count", 0)}, + {"label": "RPG edges", "value": result.get("edge_count", 0)}, + {"label": "dep graph", "value": ", ".join(dep_summary) or "not recorded"}, + {"label": "output", "value": result.get("output_path", "")}, + {"label": "visualization", "value": result.get("viz_path", result.get("viz_error", ""))}, + {"label": "trajectory", "value": result.get("trajectory", "")}, + ], + steps=[ + StepEvent(name="parse_rpg", status="recorded", reason=f"nodes={result.get('node_count', 0)}, edges={result.get('edge_count', 0)}"), + StepEvent(name="dep_graph", status="recorded" if dep_summary else "not recorded", reason=", ".join(dep_summary)), + StepEvent(name="save_rpg", status="recorded" if result.get("output_path") else "not recorded", reason=result.get("output_path", "")), + StepEvent(name="visualize", status="recorded" if result.get("viz_path") else "not recorded", reason=result.get("viz_path") or result.get("viz_error", "")), + ], + artifacts=[ + ArtifactEvent(label="rpg_json", path=result.get("output_path")), + ArtifactEvent(label="rpg_html", path=result.get("viz_path")), + ArtifactEvent(label="trajectory", path=result.get("trajectory")), + ], + verification=[VerificationEvent(name="encode", status=result.get("status"))], + evidence=result, + )) + result["report_path"] = str(report_path) + except Exception as exc: + result["report_error"] = str(exc) + return result + + def run_encode( repo_dir: str | None = None, repo_name: str | None = None, @@ -54,7 +98,12 @@ def run_encode( repo_dir = os.path.abspath(repo_dir) if not os.path.isdir(repo_dir): - return {"status": "error", "error": f"Repository directory not found: {repo_dir}"} + return _attach_encode_report({ + "status": "error", + "error": f"Repository directory not found: {repo_dir}", + "repo_name": repo_name or os.path.basename(repo_dir) or "unknown", + "output_path": output or str(RPG_FILE), + }) if repo_name is None: repo_name = os.path.basename(repo_dir) or "unknown" @@ -188,12 +237,18 @@ def run_encode( traj.complete(stats) stats["trajectory"] = str(traj.trajectory_file) - return {"status": "success", **stats} + return _attach_encode_report({"status": "success", **stats}) except Exception as exc: logger.exception("Encoding failed: %s", exc) traj.fail(str(exc)) - return {"status": "error", "error": str(exc), "trajectory": str(traj.trajectory_file)} + return _attach_encode_report({ + "status": "error", + "error": str(exc), + "repo_name": repo_name, + "output_path": output, + "trajectory": str(traj.trajectory_file), + }) def main(): diff --git a/CoderMind/scripts/rpg_encoder/run_update_rpg.py b/CoderMind/scripts/rpg_encoder/run_update_rpg.py index c368ef4..d62f7e0 100644 --- a/CoderMind/scripts/rpg_encoder/run_update_rpg.py +++ b/CoderMind/scripts/rpg_encoder/run_update_rpg.py @@ -228,8 +228,7 @@ def run_update_rpg( # to a full rebuild (rebase / diverged path). meta_git_advanced = False try: - ws_root = WORKSPACE_ROOT - current = read_head(ws_root) + current = read_head(cur_repo_dir) if current: updated_rpg.set_git_meta( head_commit=current["head_commit"], @@ -254,6 +253,8 @@ def run_update_rpg( post_nodes = len(updated_rpg.nodes) post_edges = _serialized_feature_edges(result_data) post_dep_stats = _serialized_dep_stats(result_data) + diff_summary = getattr(updated_rpg, "_last_diff_summary", None) + diff_files = getattr(updated_rpg, "_last_diff_files", None) stats = { "repo_name": repo_name, @@ -275,6 +276,10 @@ def run_update_rpg( "previous_commit": pre_commit, "new_commit": (updated_rpg.git_meta or {}).get("head_commit"), } + if isinstance(diff_summary, dict): + stats["diff_summary"] = diff_summary + if isinstance(diff_files, dict): + stats["diff_files"] = diff_files try: stats["functional_areas"] = len(updated_rpg.get_functional_areas()) except Exception: diff --git a/CoderMind/scripts/rpg_visualize.py b/CoderMind/scripts/rpg_visualize.py index 9f1b8de..e318073 100644 --- a/CoderMind/scripts/rpg_visualize.py +++ b/CoderMind/scripts/rpg_visualize.py @@ -227,6 +227,190 @@ def to_tree(nid): "children": [to_tree(r) for r in sorted(roots)]} +def _id_set(values) -> set: + if values is None: + return set() + if isinstance(values, (str, bytes)): + values = [values] + return {str(value) for value in values if value not in (None, "")} + + +def _collect_tree_ids(node: dict) -> set: + ids = set() + node_id = node.get("id") + if node_id not in (None, ""): + ids.add(str(node_id)) + for child in node.get("children", []) or []: + ids.update(_collect_tree_ids(child)) + return ids + + +def _filter_tree(node: dict, keep_ids: set, *, keep_root: bool = True) -> dict | None: + children = [ + child + for child in (_filter_tree(child, keep_ids, keep_root=False) for child in node.get("children", []) or []) + if child is not None + ] + node_id = str(node.get("id", "")) + if keep_root or node_id in keep_ids or children: + filtered = dict(node) + filtered["children"] = children + return filtered + return None + + +def _expand_rpg_focus(data: dict, rpg_ids: set, dep_ids: set, include_neighbors: bool) -> set: + focus = set(rpg_ids) + dep_to_rpg = data.get("_dep_to_rpg_map", {}) if isinstance(data.get("_dep_to_rpg_map"), dict) else {} + for dep_id, mapped_rpg_ids in dep_to_rpg.items(): + mapped = _id_set(mapped_rpg_ids) + if dep_id in dep_ids: + focus.update(mapped) + if focus.intersection(mapped): + dep_ids.add(str(dep_id)) + + if include_neighbors: + for edge in get_semantic_edges(data): + src = str(edge.get("src", "")) + dst = str(edge.get("dst", "")) + if src in focus or dst in focus: + focus.update([src, dst]) + return focus + + +def _dep_parent_map(dep_graph: dict) -> Dict[str, str]: + parent: Dict[str, str] = {} + for edge in dep_graph.get("edges", []) or []: + if edge.get("attrs", {}).get("type", "") in ("contains", "CONTAINS"): + parent[str(edge.get("dst", ""))] = str(edge.get("src", "")) + return {child: par for child, par in parent.items() if child and par} + + +def _expand_dep_focus(data: dict, rpg_ids: set, dep_ids: set, include_neighbors: bool) -> set: + dep_graph = data.get("dep_graph", {}) if isinstance(data.get("dep_graph"), dict) else {} + raw_nodes = dep_graph.get("nodes", {}) if isinstance(dep_graph.get("nodes"), dict) else {} + focus = set(dep_ids) + dep_to_rpg = data.get("_dep_to_rpg_map", {}) if isinstance(data.get("_dep_to_rpg_map"), dict) else {} + for dep_id, mapped_rpg_ids in dep_to_rpg.items(): + if rpg_ids.intersection(_id_set(mapped_rpg_ids)): + focus.add(str(dep_id)) + for dep_id, attrs in raw_nodes.items(): + if not isinstance(attrs, dict): + continue + mapped = _id_set(attrs.get("rpg_nodes")) + if mapped.intersection(rpg_ids): + focus.add(str(dep_id)) + if str(dep_id) in focus: + rpg_ids.update(mapped) + + if include_neighbors: + for edge in dep_graph.get("edges", []) or []: + edge_type = edge.get("attrs", {}).get("type", "") + if edge_type in ("contains", "CONTAINS"): + continue + src = str(edge.get("src", "")) + dst = str(edge.get("dst", "")) + if src in focus or dst in focus: + focus.update([src, dst]) + + parent = _dep_parent_map(dep_graph) + for dep_id in list(focus): + cur = dep_id + while cur in parent: + cur = parent[cur] + focus.add(cur) + for dep_id in list(focus): + attrs = raw_nodes.get(dep_id) + if isinstance(attrs, dict): + rpg_ids.update(_id_set(attrs.get("rpg_nodes"))) + rpg_ids.update(_id_set(dep_to_rpg.get(dep_id))) + return focus + + +def build_focused_graph_data( + data: dict, + *, + rpg_node_ids: List[str] | None = None, + dep_node_ids: List[str] | None = None, + include_neighbors: bool = True, +) -> dict: + selected_rpg = _id_set(rpg_node_ids) + selected_dep = _id_set(dep_node_ids) + if not selected_rpg and not selected_dep: + return dict(data) + + rpg_focus = _expand_rpg_focus(data, set(selected_rpg), set(selected_dep), include_neighbors) + dep_focus = _expand_dep_focus(data, rpg_focus, set(selected_dep), include_neighbors) + rpg_focus = _expand_rpg_focus(data, rpg_focus, dep_focus, include_neighbors=False) + + tree = normalize_to_tree(data) + filtered_tree = _filter_tree(tree, rpg_focus) or {"id": "__root__", "name": "Focused graph", "children": []} + tree_ids = _collect_tree_ids(filtered_tree) + matched_rpg = sorted(selected_rpg.intersection(tree_ids)) + + semantic_edges = [ + edge for edge in get_semantic_edges(data) + if str(edge.get("src", "")) in tree_ids and str(edge.get("dst", "")) in tree_ids + ] + + dep_graph = data.get("dep_graph", {}) if isinstance(data.get("dep_graph"), dict) else {} + raw_nodes = dep_graph.get("nodes", {}) if isinstance(dep_graph.get("nodes"), dict) else {} + raw_edges = dep_graph.get("edges", []) if isinstance(dep_graph.get("edges"), list) else [] + filtered_dep_nodes = {dep_id: attrs for dep_id, attrs in raw_nodes.items() if str(dep_id) in dep_focus} + filtered_dep_edges = [ + edge for edge in raw_edges + if str(edge.get("src", "")) in dep_focus and str(edge.get("dst", "")) in dep_focus + ] + dep_ids = {str(dep_id) for dep_id in filtered_dep_nodes} + matched_dep = sorted(selected_dep.intersection(dep_ids)) + + dep_to_rpg = data.get("_dep_to_rpg_map", {}) if isinstance(data.get("_dep_to_rpg_map"), dict) else {} + filtered_map = {} + for dep_id, mapped_rpg_ids in dep_to_rpg.items(): + if str(dep_id) not in dep_ids: + continue + mapped = [str(rpg_id) for rpg_id in mapped_rpg_ids if str(rpg_id) in tree_ids] + if mapped: + filtered_map[str(dep_id)] = mapped + + focused = dict(data) + if isinstance(data.get("nodes"), list): + focused["nodes"] = [ + dict(node) for node in data["nodes"] + if isinstance(node, dict) and str(node.get("id", "")) in tree_ids + ] + focused["root"] = filtered_tree + focused["edges"] = semantic_edges + focused["dep_graph"] = {**dep_graph, "nodes": filtered_dep_nodes, "edges": filtered_dep_edges} + focused["_dep_to_rpg_map"] = filtered_map + focused["_focused_graph"] = { + "selected_rpg_nodes": sorted(selected_rpg), + "selected_dep_nodes": sorted(selected_dep), + "matched_rpg_nodes": matched_rpg, + "matched_dep_nodes": matched_dep, + "rpg_node_count": len(tree_ids), + "dep_node_count": len(dep_ids), + "semantic_edge_count": len(semantic_edges), + "dep_edge_count": len(filtered_dep_edges), + } + return focused + + +def generate_focused_html( + data: dict, + *, + rpg_node_ids: List[str] | None = None, + dep_node_ids: List[str] | None = None, + include_neighbors: bool = True, +) -> str: + return generate_html(build_focused_graph_data( + data, + rpg_node_ids=rpg_node_ids, + dep_node_ids=dep_node_ids, + include_neighbors=include_neighbors, + )) + + def generate_html(data: dict) -> str: tree = normalize_to_tree(data) semantic_edges = get_semantic_edges(data) diff --git a/CoderMind/scripts/run_batch.py b/CoderMind/scripts/run_batch.py index d2ed385..02ce272 100644 --- a/CoderMind/scripts/run_batch.py +++ b/CoderMind/scripts/run_batch.py @@ -65,6 +65,8 @@ cmd_for, REPO_DIR, ) +from common.run_events import ArtifactEvent, CommandRun, StepEvent, VerificationEvent +from common.run_report import write_command_report from code_gen.context_collector import build_dependency_context from code_gen.prompts import ( build_test_prompt_from_batch, @@ -923,8 +925,53 @@ def run_batch( # CLI # ============================================================================ +def _write_batch_report(result: Dict[str, Any]) -> Optional[str]: + if result.get("type") not in {"batch_complete", "batch_failed", "final_test", "complete"}: + return None + try: + stats = result.get("stats") or {} + report_path = write_command_report(CommandRun( + command="code_gen", + title="CoderMind code_gen Batch View", + status=result.get("type"), + summary=[ + {"label": "result", "value": result.get("type", "")}, + {"label": "success", "value": result.get("success", "")}, + {"label": "batch", "value": result.get("batch_id", "")}, + {"label": "attempts", "value": result.get("attempts_used", "")}, + {"label": "duration", "value": f"{result.get('total_duration', 0):.1f}s" if "total_duration" in result else ""}, + {"label": "completed", "value": stats.get("completed", "")}, + {"label": "failed", "value": stats.get("failed", result.get("failed", ""))}, + ], + steps=[ + StepEvent(name="batch", status=result.get("type"), reason=result.get("failure_reason") or result.get("message", "")), + StepEvent(name="verification", status=result.get("success"), reason=f"passed={result.get('passed', '')} failed={result.get('failed', '')} errors={result.get('errors', '')}"), + StepEvent(name="next_action", status="available" if result.get("next_action") else "missing", reason=result.get("next_action", "")), + ], + artifacts=[ + ArtifactEvent(label="feature_spec", path=FEATURE_SPEC_FILE), + ArtifactEvent(label="tasks", path=TASKS_FILE), + ArtifactEvent(label="code_gen_state", path=STATE_FILE), + ArtifactEvent(label="rpg_json", path=REPO_RPG_FILE), + ], + verification=[ + VerificationEvent(name="result", status=result.get("success", result.get("type"))), + VerificationEvent(name="pytest", status=result.get("passed", ""), detail=f"failed={result.get('failed', '')}, errors={result.get('errors', '')}"), + ], + evidence={"result": result}, + )) + return str(report_path) + except Exception as exc: + result["report_error"] = str(exc) + return None + + def print_result(result: Dict[str, Any], json_output: bool = False) -> None: """Print result to stdout and log it.""" + report_path = _write_batch_report(result) + if report_path: + result["report_path"] = report_path + # Always log the result as JSON for the file log logger.info("Batch result: %s", json.dumps(result, indent=2)) @@ -962,6 +1009,9 @@ def print_result(result: Dict[str, Any], json_output: bool = False) -> None: if "next_action" in result: print(f"\n -> {result['next_action']}") + if result.get("report_path"): + print(f"\n Report: {result['report_path']}") + def main() -> int: # Convert SIGTERM → SystemExit so "except BaseException" in Popen calls diff --git a/CoderMind/scripts/skeleton/skeleton_models.py b/CoderMind/scripts/skeleton/skeleton_models.py index 0d34aee..db6d622 100644 --- a/CoderMind/scripts/skeleton/skeleton_models.py +++ b/CoderMind/scripts/skeleton/skeleton_models.py @@ -15,7 +15,7 @@ import json import logging from typing import List, Optional, Dict, Any -from abc import abstractmethod +from abc import ABC, abstractmethod from pathlib import PurePosixPath from rpg.code_unit import CodeUnit, ParsedFile @@ -36,7 +36,7 @@ def normalize_path(path: str) -> str: return posix_path -class RepoNode: +class RepoNode(ABC): """Base class for repository nodes.""" def __init__(self, name: str, path: str): diff --git a/CoderMind/scripts/smoke_test.py b/CoderMind/scripts/smoke_test.py index 71e9c4d..2f91e05 100644 --- a/CoderMind/scripts/smoke_test.py +++ b/CoderMind/scripts/smoke_test.py @@ -160,7 +160,13 @@ def _find_source_files(repo_path: Path) -> List[Path]: def _run_in_repo(repo_path: Path, cmd: List[str], timeout: int = 30) -> subprocess.CompletedProcess: """Run a command in the repo directory with the dev venv.""" env = os.environ.copy() - env["PYTHONPATH"] = str(repo_path) + python_path = [str(repo_path)] + scripts_dir = repo_path / "scripts" + if scripts_dir.is_dir(): + python_path.append(str(scripts_dir)) + if env.get("PYTHONPATH"): + python_path.append(env["PYTHONPATH"]) + env["PYTHONPATH"] = os.pathsep.join(python_path) # Suppress interactive prompts env["PYTHONDONTWRITEBYTECODE"] = "1" return subprocess.run( diff --git a/CoderMind/scripts/update_graphs.py b/CoderMind/scripts/update_graphs.py index b9964b3..0f48fb2 100644 --- a/CoderMind/scripts/update_graphs.py +++ b/CoderMind/scripts/update_graphs.py @@ -34,6 +34,8 @@ from common.paths import REPO_RPG_FILE, DEP_GRAPH_FILE, RPG_HTML_FILE, HOOK_CALLS_LOG # noqa: E402 from common.rpg_io import atomic_write_rpg, safe_load_rpg # noqa: E402 +from common.run_events import ArtifactEvent, CommandRun, StepEvent, VerificationEvent # noqa: E402 +from common.run_report import write_command_report # noqa: E402 # Shared message used by every subcommand that requires an existing @@ -79,6 +81,166 @@ def _log_hook_call(hook_type: str, result: dict) -> None: pass +def _change_count(value: object) -> int: + if isinstance(value, int): + return value + if isinstance(value, (list, tuple, set, dict)): + return len(value) + return 0 + + +def _diff_summary(result: dict) -> dict: + summary = result.get("diff_summary") + if isinstance(summary, dict): + return { + "added": _change_count(summary.get("added")), + "deleted": _change_count(summary.get("deleted")), + "modified": _change_count(summary.get("modified")), + "renamed": _change_count(summary.get("renamed")), + } + return { + key: _change_count(result.get(key)) + for key in ("added", "deleted", "modified", "renamed") + } + + +def _format_count_delta(value: object, delta: object) -> object: + if value in (None, ""): + return "" + if isinstance(delta, int): + return f"{value} (delta: {delta:+d})" + return value + + +def _format_diff_summary(summary: dict) -> str: + total = sum(summary.values()) + parts = [f"{total} semantic files"] + for key in ("added", "deleted", "modified", "renamed"): + count = summary.get(key, 0) + if count: + parts.append(f"{key}={count}") + return ", ".join(parts) + + +def _git_delta_files(prev_ref: str, workspace_root: str) -> list[dict[str, str]]: + import subprocess + + try: + output = subprocess.check_output( + ["git", "diff", "--relative", "--name-status", f"{prev_ref}..HEAD", "--", "."], + cwd=workspace_root, + stderr=subprocess.DEVNULL, + text=True, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + return [] + files: list[dict[str, str]] = [] + for line in output.splitlines(): + parts = line.split("\t") + if len(parts) >= 2: + files.append({"status": parts[0], "path": parts[-1]}) + return files + + +def _attach_update_report(result: dict) -> dict: + try: + semantic_summary = _diff_summary(result) + semantic_total = sum(semantic_summary.values()) + git_delta = result.get("git_delta") + git_total = _change_count(git_delta) if git_delta is not None else "" + node_count = result.get("node_count", result.get("rpg_nodes", "")) + rpg_path = result.get("output_path") or result.get("rpg_path") + dep_summary = "" + if result.get("dep_nodes") not in (None, ""): + dep_summary = "nodes={}".format( + _format_count_delta( + result.get("dep_nodes"), + result.get("dep_nodes_delta"), + ) + ) + if result.get("dep_edges") not in (None, ""): + dep_summary += ", edges={}".format( + _format_count_delta( + result.get("dep_edges"), + result.get("dep_edges_delta"), + ) + ) + viz_status = result.get("viz_error") or ( + "ok" if result.get("viz_path") else "not recorded" + ) + report_path = write_command_report(CommandRun( + command="update_rpg", + title="CoderMind update_rpg Explain View", + status=result.get("status") or ("error" if result.get("error") else result.get("mode")), + summary=[ + {"label": "mode", "value": result.get("mode", "")}, + { + "label": "reason", + "value": result.get("reason") or result.get("error", ""), + }, + {"label": "git files", "value": git_total}, + {"label": "semantic files", "value": semantic_total}, + { + "label": "RPG nodes", + "value": _format_count_delta( + node_count, + result.get("nodes_delta"), + ), + }, + {"label": "dep graph", "value": dep_summary}, + { + "label": "visualization", + "value": result.get("viz_path") or result.get("viz_error", ""), + }, + ], + steps=[ + StepEvent( + name="git delta", + status=result.get("mode", ""), + reason=( + f"{git_total} changed files" + if git_total != "" + else "not recorded" + ), + ), + StepEvent( + name="semantic delta", + status=result.get("mode", ""), + reason=_format_diff_summary(semantic_summary), + ), + StepEvent( + name="sync graph", + status=result.get("status") or ("error" if result.get("error") else result.get("mode", "")), + reason=result.get("reason", ""), + ), + StepEvent( + name="visualize", + status=( + "ok" + if result.get("viz_path") + else "error" + if result.get("viz_error") + else "skipped" + ), + reason=result.get("viz_path") or result.get("viz_error", ""), + ), + ], + artifacts=[ + ArtifactEvent(label="rpg_json", path=rpg_path), + ArtifactEvent(label="rpg_html", path=result.get("viz_path")), + ], + verification=[ + VerificationEvent(name="update_rpg", status=result.get("status") or ("error" if result.get("error") else result.get("mode"))), + VerificationEvent(name="viz", status=viz_status), + ], + evidence=result, + )) + result["report_path"] = str(report_path) + except Exception as exc: + result["report_error"] = str(exc) + return result + + def _refresh_rpg_html(rpg_path: Path) -> dict: """Regenerate ``rpg.html`` next to ``rpg.json`` after a hook update. @@ -364,12 +526,12 @@ def cmd_sync( # instead so the hook log shows exactly what's wrong and how to fix # it. if not rpg_path.is_file(): - return { + return _attach_update_report({ "mode": "sync", "error": _RPG_MISSING_MSG.format(rpg_path=rpg_path), "rpg_path": str(rpg_path), "duration": round(time.time() - t0, 3), - } + }) svc = RPGService.load(str(rpg_path)) @@ -427,6 +589,7 @@ def cmd_sync( "viz_error": viz_result.get("viz_error"), "duration": round(time.time() - t0, 3), } + _attach_update_report(sync_out) _log_hook_call("sync", sync_out) return sync_out @@ -458,11 +621,11 @@ def cmd_update_rpg( t0 = time.time() if not rpg_path.is_file(): - return { + return _attach_update_report({ "mode": "update-rpg", "error": _RPG_MISSING_MSG.format(rpg_path=rpg_path), "rpg_path": str(rpg_path), - } + }) # Check git has enough history try: @@ -472,10 +635,11 @@ def cmd_update_rpg( stderr=subprocess.DEVNULL, ).decode().strip() except (subprocess.CalledProcessError, FileNotFoundError): - return { + return _attach_update_report({ "mode": "update-rpg", "error": "Need at least 2 commits for incremental update (no HEAD~1)", - } + "rpg_path": str(rpg_path), + }) # Prune orphaned worktrees from previous runs that were killed. subprocess.call( @@ -495,22 +659,34 @@ def cmd_update_rpg( text=True, ) if wt_proc.returncode != 0: - return { + return _attach_update_report({ "mode": "update-rpg", "error": f"git worktree add failed for {prev_ref}: {wt_proc.stderr.strip()}", - } + "rpg_path": str(rpg_path), + "prev_ref": prev_ref, + }) + from common.git_utils import git_workspace_prefix from rpg_encoder.run_update_rpg import run_update_rpg + git_prefix = git_workspace_prefix(workspace_root) + last_repo_dir = ( + os.path.join(worktree_dir, git_prefix) + if git_prefix + else worktree_dir + ) + git_delta = _git_delta_files(prev_ref, workspace_root) + result = run_update_rpg( rpg_file=str(rpg_path), - last_repo_dir=worktree_dir, + last_repo_dir=last_repo_dir, cur_repo_dir=workspace_root, dep_graph_path=str(dep_graph_path), ) result["mode"] = "update-rpg" result["prev_ref"] = prev_ref + result["git_delta"] = git_delta # Refresh ``rpg.html`` whenever the JSON was actually rewritten. # ``run_update_rpg`` returns ``status="success"`` on a normal @@ -524,6 +700,7 @@ def cmd_update_rpg( result["viz_error"] = viz_result["viz_error"] result["duration"] = round(time.time() - t0, 3) + _attach_update_report(result) _log_hook_call("update-rpg", result) return result diff --git a/CoderMind/templates/commands/rpg_edit.md b/CoderMind/templates/commands/rpg_edit.md index bb8550f..f467237 100644 --- a/CoderMind/templates/commands/rpg_edit.md +++ b/CoderMind/templates/commands/rpg_edit.md @@ -241,9 +241,16 @@ replies above before proceeding. Treat any other free-form reply as ### Step 5: Apply Changes (RPG-First, on a dedicated branch) All work in this step happens on a fresh `rpg-edit/` branch -in the project repo (workspace root), never directly on `main`. The -branch is merged into `main` only after Step 5e tests pass, so a -failed run leaves `main` clean and the branch preserved for inspection. +in the project repo (workspace root), never directly on the user's +working branch. The branch is merged back into `` — the +branch the user was on when the command started — only after Step 5e +tests pass, so a failed run leaves `` clean and the branch +preserved for inspection. + +The user may be developing on any branch (a feature branch, not +necessarily `main`). Never assume `main`: capture the base branch in +Step 5a and restore it on both the success and failure paths so the +user's git environment is left exactly where they started. `` should be derived from the plan filename or the first affected node id (e.g. last 8 chars of `feature_changes[0].node_id`). @@ -259,9 +266,19 @@ test -z "$(git status --porcelain)" || { echo "Error: working tree has uncommitted changes. Commit or stash first."; exit 1; } +# Capture the branch the user is currently on so we can merge back into +# it (and restore it on failure). Do NOT assume `main` — the user may +# be developing on a feature branch. +BASE_BRANCH="$(git rev-parse --abbrev-ref HEAD)" +echo "base branch: $BASE_BRANCH" + git checkout -b rpg-edit/ ``` +Remember `` = the captured `$BASE_BRANCH` value — every +later step that references `` means this branch, not +`main`. + If the precondition fails, surface the error to the user and stop — do **not** silently `git stash`, as that would hide their work. @@ -339,22 +356,23 @@ If any test step fails: fix the code on the branch, re-run dep-refresh (Step 5c command), `git commit --amend --no-edit` to fold the fix into the same branch commit, then re-test. -**Step 5e — Merge into `main` (only after Step 5d is green):** +**Step 5e — Merge into `` (only after Step 5d is green):** ```bash -git checkout main +git checkout "$BASE_BRANCH" git merge --no-ff rpg-edit/ -m "rpg_edit: merge " git branch -d rpg-edit/ ``` `--no-ff` preserves the merge commit so the rpg_edit boundary is -visible in `git log --graph`. +visible in `git log --graph`. The user is returned to `` +— the same branch they started on. ### Step 6: Report Results - **Success path** (Step 5e completed): - > Merged `rpg-edit/` into `main` (commit ``). + > Merged `rpg-edit/` into `` (commit ``). > To revert later: > - Code: `git revert -m 1 ` > - Graphs: `cmind script rpg_edit/apply.py --rollback --json` @@ -369,17 +387,17 @@ visible in `git log --graph`. - **Failure path** (Step 5d failed, Step 5e skipped): - Restore `main` and preserve the branch for the user to inspect: + Restore `` and preserve the branch for the user to inspect: ```bash - git checkout main + git checkout "$BASE_BRANCH" ``` Report to the user: > Tests failed. Branch `rpg-edit/` preserved for inspection. - > `main` is clean. Choose one of: - > - Inspect: `git diff main rpg-edit/` + > `` is clean. Choose one of: + > - Inspect: `git diff rpg-edit/` > - Discard code + graphs together: > `cmind script rpg_edit/apply.py --rollback --rollback-branch rpg-edit/ --json` > - Discard code only: `git branch -D rpg-edit/` @@ -390,6 +408,6 @@ visible in `git log --graph`. 1. **RPG is the anchor** — all modifications start from RPG feature graph nodes, not from files. 2. **Three-way sync** — code, RPG, and dep_graph must stay consistent after every edit. 3. **User confirmation** — always confirm the plan before applying changes. Never auto-apply. -4. **Branch isolation** — `main` is touched only after tests pass. Failed runs leave the work on a `rpg-edit/` branch for inspection. +4. **Branch isolation** — `` (the branch the user started on) is touched only after tests pass, and the user is always returned to it. Never assume `main`. Failed runs leave the work on a `rpg-edit/` branch for inspection. 5. **Coordinated rollback** — `--rollback --rollback-branch ` reverts RPG, dep_graph, and the dedicated branch in one step. 6. **Independent command** — does not depend on or invoke any other `/cmind.*` command. diff --git a/CoderMind/tests/test_encode_commands.py b/CoderMind/tests/test_encode_commands.py index 1e8b5db..391cce0 100644 --- a/CoderMind/tests/test_encode_commands.py +++ b/CoderMind/tests/test_encode_commands.py @@ -273,6 +273,103 @@ def test_missing_rpg_file(self, tmp_repo): assert result["status"] == "error" assert "not found" in result["error"] + def test_meta_git_reads_explicit_cur_repo_dir(self, tmp_path): + from rpg_encoder.run_update_rpg import run_update_rpg + + last_repo = tmp_path / "last" + cur_repo = tmp_path / "current" + last_repo.mkdir() + cur_repo.mkdir() + rpg_file = tmp_path / "rpg.json" + rpg_file.write_text(json.dumps({ + "repo_name": "test_repo", + "repo_info": "", + "root": { + "id": "test_repo_L0", + "name": "test_repo", + "node_type": "repo", + "level": 0, + "meta": {"type_name": "directory", "path": "."}, + "children": [], + }, + "edges": [], + })) + + calls = [] + + def fake_read_head(repo_dir): + calls.append(Path(repo_dir)) + return { + "head_commit": "a" * 40, + "head_short": "aaaaaaa", + "head_branch": "main", + "head_timestamp": "2026-06-30T00:00:00+00:00", + } + + with patch( + "rpg_encoder.rpg_evolution.RPGEvolution.process_diff", + side_effect=lambda **kwargs: kwargs["last_rpg"], + ), patch( + "rpg.service.RPGService.enrich_from_code", + return_value={}, + ), patch("common.git_utils.read_head", side_effect=fake_read_head): + result = run_update_rpg( + rpg_file=str(rpg_file), + last_repo_dir=str(last_repo), + cur_repo_dir=str(cur_repo), + ) + + assert result["status"] == "success" + assert result["meta_git_advanced"] is True + assert result["new_commit"] == "a" * 40 + assert calls == [cur_repo.resolve()] + + def test_diff_summary_is_returned(self, tmp_rpg_file, tmp_path): + from rpg_encoder.run_update_rpg import run_update_rpg + + last_repo = tmp_path / "last" + cur_repo = tmp_path / "current" + last_repo.mkdir() + cur_repo.mkdir() + + def fake_process_diff(**kwargs): + rpg = kwargs["last_rpg"] + rpg._last_diff_summary = { + "added": 1, + "deleted": 0, + "modified": 2, + "renamed": 0, + } + rpg._last_diff_files = { + "added": ["a.py"], + "deleted": [], + "modified": ["b.py", "c.py"], + "renamed": [], + } + return rpg + + with patch( + "rpg_encoder.rpg_evolution.RPGEvolution.process_diff", + side_effect=fake_process_diff, + ), patch( + "rpg.service.RPGService.enrich_from_code", + return_value={}, + ), patch("common.git_utils.read_head", return_value=None): + result = run_update_rpg( + rpg_file=tmp_rpg_file, + last_repo_dir=str(last_repo), + cur_repo_dir=str(cur_repo), + ) + + assert result["status"] == "success" + assert result["diff_summary"] == { + "added": 1, + "deleted": 0, + "modified": 2, + "renamed": 0, + } + assert result["diff_files"]["modified"] == ["b.py", "c.py"] + def test_missing_last_repo_dir(self, tmp_rpg_file): """Should return error when last repo dir doesn't exist.""" from rpg_encoder.run_update_rpg import run_update_rpg @@ -295,6 +392,58 @@ def test_missing_cur_repo_dir(self, tmp_rpg_file): assert "not found" in result["error"] +# ============================================================================ +# Test: update_graphs report wiring +# ============================================================================ + + +def test_attach_update_report_uses_update_rpg_result_fields(tmp_path, monkeypatch): + import update_graphs + + captured = {} + + def fake_write_command_report(run): + captured.update(run.to_dict()) + return tmp_path / "report.html" + + monkeypatch.setattr(update_graphs, "write_command_report", fake_write_command_report) + + result = update_graphs._attach_update_report({ + "mode": "update-rpg", + "status": "success", + "output_path": "/tmp/rpg.json", + "node_count": 4504, + "nodes_delta": 2, + "dep_nodes": 2708, + "dep_nodes_delta": 46, + "dep_edges": 5498, + "dep_edges_delta": 103, + "diff_summary": { + "added": 0, + "deleted": 0, + "modified": 3, + "renamed": 0, + }, + "git_delta": [ + {"status": "M", "path": "scripts/a.py"}, + {"status": "M", "path": "tests/test_a.py"}, + ], + "viz_path": "/tmp/rpg.html", + }) + + cards = {card["label"]: card["value"] for card in captured["summary"]} + artifacts = {artifact["label"]: artifact["path"] for artifact in captured["artifacts"]} + assert result["report_path"] == str(tmp_path / "report.html") + assert cards["git files"] == 2 + assert cards["semantic files"] == 3 + assert cards["RPG nodes"] == "4504 (delta: +2)" + assert cards["dep graph"] == "nodes=2708 (delta: +46), edges=5498 (delta: +103)" + assert artifacts["rpg_json"] == "/tmp/rpg.json" + assert captured["status"] == "success" + assert captured["steps"][0]["reason"] == "2 changed files" + assert captured["steps"][1]["reason"] == "3 semantic files, modified=3" + + # ============================================================================ # Test: Template validation # ============================================================================ @@ -385,7 +534,10 @@ def _make_rpg_with_dep_graph(tmp_path): "name": "Core Logic", "node_type": "functional_area", "level": 1, - "meta": {"type_name": "directory", "path": "."}, + "meta": { + "type_name": "directory", + "path": ["src/core.py", "src/extra.py"], + }, "children": [ { "id": "feat_1", @@ -457,6 +609,17 @@ def test_graph_query_engine_search_feature(self, tmp_rpg_with_dep_graph): assert len(results) >= 1 assert any(r["id"] == "area_1" for r in results) + def test_graph_query_engine_search_list_path(self, tmp_rpg_with_dep_graph): + from rpg.graph_query import GraphQueryEngine + engine = GraphQueryEngine.from_rpg_file(tmp_rpg_with_dep_graph) + feature_results = engine.search("src/core.py", scope="feature") + all_results = engine.search("src/extra.py", scope="all") + assert any( + r["id"] == "area_1" and r["path"] == ["src/core.py", "src/extra.py"] + for r in feature_results + ) + assert any(r["id"] == "area_1" for r in all_results) + def test_graph_query_engine_explore(self, tmp_rpg_with_dep_graph): """explore() should traverse edges from a node.""" from rpg.graph_query import GraphQueryEngine diff --git a/CoderMind/tests/test_encoder_workspace_layout.py b/CoderMind/tests/test_encoder_workspace_layout.py index 71a3ea7..7184cfe 100644 --- a/CoderMind/tests/test_encoder_workspace_layout.py +++ b/CoderMind/tests/test_encoder_workspace_layout.py @@ -18,6 +18,7 @@ import importlib import os +import subprocess import sys from pathlib import Path @@ -31,6 +32,10 @@ # Fixtures — reload ``common.paths`` against an arbitrary workspace # --------------------------------------------------------------------------- +def _git(cwd: Path, *args: str) -> None: + subprocess.run(["git", *args], cwd=cwd, check=True, capture_output=True) + + def _reload_paths_against(workspace: Path): """Import / reload ``common.paths`` with ``workspace`` as cwd. @@ -77,6 +82,30 @@ def workspace_with_repo_subdir(tmp_path, monkeypatch): return ws +@pytest.fixture +def workspace_inside_git_repo(tmp_path, monkeypatch): + git_root = tmp_path / "gitroot" + workspace = git_root / "subproject" + src = workspace / "src" + src.mkdir(parents=True) + (workspace / ".cmind").mkdir() + (src / "module.py").write_text("def value():\n return 1\n") + + _git(git_root, "init", "-q", "-b", "main") + _git(git_root, "config", "user.email", "test@example.com") + _git(git_root, "config", "user.name", "Test User") + _git(git_root, "add", ".") + _git(git_root, "commit", "-q", "-m", "initial") + + (src / "module.py").write_text("def value():\n return 2\n") + _git(git_root, "add", ".") + _git(git_root, "commit", "-q", "-m", "update module") + + monkeypatch.chdir(workspace) + monkeypatch.delenv("CMIND_WORKSPACE", raising=False) + return git_root, workspace + + # --------------------------------------------------------------------------- # Encoder entry point defaults # --------------------------------------------------------------------------- @@ -229,3 +258,65 @@ def test_update_graphs_auto_detect_ignores_present_repo_subdir( assert result == str(workspace_with_repo_subdir) # Critically NOT the repo/ subdir assert result != str(workspace_with_repo_subdir / "repo") + + +def test_git_workspace_prefix_handles_workspace_inside_git_repo( + workspace_inside_git_repo, +): + git_root, workspace = workspace_inside_git_repo + _reload_paths_against(workspace) + from common.git_utils import git_workspace_prefix + + assert git_workspace_prefix(workspace) == "subproject" + assert git_workspace_prefix(git_root) == "" + + +def test_cmd_update_rpg_passes_subdir_adjusted_last_repo_dir( + workspace_inside_git_repo, + monkeypatch, +): + _, workspace = workspace_inside_git_repo + _reload_paths_against(workspace) + + import update_graphs + importlib.reload(update_graphs) + import rpg_encoder.run_update_rpg as run_update_mod + + data_dir = workspace / ".cmind" / "data" + data_dir.mkdir(parents=True) + rpg_path = data_dir / "rpg.json" + dep_graph_path = data_dir / "dep_graph.json" + rpg_path.write_text("{}") + + captured = {} + + def fake_run_update_rpg(*, rpg_file, last_repo_dir, cur_repo_dir, dep_graph_path): + last_repo = Path(last_repo_dir) + captured["rpg_file"] = rpg_file + captured["last_repo_dir"] = last_repo.as_posix() + captured["cur_repo_dir"] = cur_repo_dir + captured["dep_graph_path"] = dep_graph_path + captured["old_file_at_workspace_root"] = ( + last_repo / "src" / "module.py" + ).is_file() + captured["nested_subproject_exists"] = ( + last_repo / "subproject" + ).exists() + return {"status": "success", "repo_name": "subproject"} + + monkeypatch.setattr(run_update_mod, "run_update_rpg", fake_run_update_rpg) + monkeypatch.setattr(update_graphs, "_refresh_rpg_html", lambda _rpg_path: {}) + monkeypatch.setattr(update_graphs, "_attach_update_report", lambda result: result) + monkeypatch.setattr(update_graphs, "_log_hook_call", lambda *_args, **_kwargs: None) + + result = update_graphs.cmd_update_rpg( + rpg_path=rpg_path, + dep_graph_path=dep_graph_path, + workspace_root=str(workspace), + ) + + assert result["status"] == "success" + assert captured["last_repo_dir"].endswith("/subproject") + assert captured["cur_repo_dir"] == str(workspace) + assert captured["old_file_at_workspace_root"] is True + assert captured["nested_subproject_exists"] is False diff --git a/CoderMind/tests/test_rpg_edit_run_report.py b/CoderMind/tests/test_rpg_edit_run_report.py new file mode 100644 index 0000000..b0dfad7 --- /dev/null +++ b/CoderMind/tests/test_rpg_edit_run_report.py @@ -0,0 +1,320 @@ +from __future__ import annotations + +import importlib.util +import json +import sys +import types +from pathlib import Path + +_REPO = Path(__file__).resolve().parents[1] +_SCRIPTS = _REPO / "scripts" +if str(_SCRIPTS) not in sys.path: + sys.path.insert(0, str(_SCRIPTS)) + + +def _load_script(name: str, path: Path): + spec = importlib.util.spec_from_file_location(name, path) + assert spec is not None and spec.loader is not None + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + return module + + +def test_validate_and_locate_results_are_persisted(tmp_path: Path, monkeypatch) -> None: + validate = _load_script("rpg_edit_validate_test", _SCRIPTS / "rpg_edit" / "validate.py") + locate = _load_script("rpg_edit_locate_test", _SCRIPTS / "rpg_edit" / "locate.py") + + validate_path = tmp_path / "rpg_edit_validate.json" + locate_path = tmp_path / "rpg_edit_locate.json" + monkeypatch.setattr(validate, "RPG_EDIT_VALIDATE_FILE", validate_path) + monkeypatch.setattr(locate, "RPG_EDIT_LOCATE_FILE", locate_path) + + validate._write_validate_result({"type": "ready", "nodes": 2}) + locate._write_locate_result({"type": "candidates", "results": [{"node_id": "n1"}]}) + + assert json.loads(validate_path.read_text(encoding="utf-8"))["type"] == "ready" + assert json.loads(locate_path.read_text(encoding="utf-8"))["results"][0]["node_id"] == "n1" + + +def test_code_result_is_persisted(tmp_path: Path, monkeypatch) -> None: + code = _load_script("rpg_edit_code_test", _SCRIPTS / "rpg_edit" / "code.py") + + result_path = tmp_path / "rpg_edit_code_result.json" + monkeypatch.setattr(code, "RPG_EDIT_CODE_RESULT_FILE", result_path) + + code._write_code_result({"success": True, "commit_sha": "abc123"}) + + data = json.loads(result_path.read_text(encoding="utf-8")) + assert data == {"success": True, "commit_sha": "abc123"} + + +def test_code_result_clears_recovered_subagent_error(tmp_path: Path, monkeypatch) -> None: + code = _load_script("rpg_edit_code_retry_test", _SCRIPTS / "rpg_edit" / "code.py") + + plan_path = tmp_path / "plan.json" + plan_path.write_text( + json.dumps({"code_changes": [{"file_path": "a.py", "description": "update report"}]}), + encoding="utf-8", + ) + calls = iter([ + (None, "Sub-agent failed after 900.1s: timeout"), + ("CODE_STATUS: COMPLETE", None), + ]) + fake_run_batch = types.ModuleType("run_batch") + fake_run_batch.dispatch_sub_agent = lambda *args, **kwargs: next(calls) + monkeypatch.setitem(sys.modules, "run_batch", fake_run_batch) + monkeypatch.setattr(code, "_format_rpg_target_nodes", lambda plan, rpg_path: "nodes") + monkeypatch.setattr(code, "_format_impact_context", lambda plan: "impact") + monkeypatch.setattr(code, "_commit_changes", lambda repo_path, summary, status: "abc123") + + result = code.apply_code_changes(plan_path, tmp_path / "rpg.json", tmp_path, max_iterations=2, timeout=1) + + assert result["success"] is True + assert result["last_status"] == "complete" + assert result["last_error"] is None + assert result["commit_sha"] == "abc123" + assert result["iterations"][0]["parsed_status"] == "llm_error" + assert result["iterations"][1]["parsed_status"] == "complete" + + +def test_review_publish_report_returns_report_path(tmp_path: Path, monkeypatch) -> None: + review = _load_script("rpg_edit_review_test", _SCRIPTS / "rpg_edit" / "review.py") + + validate_path = tmp_path / "validate.json" + locate_path = tmp_path / "locate.json" + plan_path = tmp_path / "plan.json" + impact_path = tmp_path / "impact.json" + code_path = tmp_path / "code.json" + apply_path = tmp_path / "apply.json" + review_path = tmp_path / "review.json" + report_path = tmp_path / "report.html" + rpg_path = tmp_path / "rpg.json" + + validate_path.write_text(json.dumps({"type": "ready"}), encoding="utf-8") + locate_path.write_text( + json.dumps({"type": "candidates", "query": "a.py", "results": [{"node_id": "n1", "name": "Node", "score": 1.0, "dep_nodes": ["a.py:f"]}]}), + encoding="utf-8", + ) + plan_path.write_text(json.dumps({"affected_nodes": ["n1"], "code_changes": [{"file_path": "a.py"}]}), encoding="utf-8") + impact_path.write_text( + json.dumps({ + "type": "impact", + "results": { + "n1": { + "name": "Node", + "dep_nodes": ["a.py:f"], + "affected_files": ["a.py"], + "impact_summary": {"total_callers": 1, "affected_file_count": 1}, + } + }, + }), + encoding="utf-8", + ) + code_path.write_text(json.dumps({"success": True, "commit_sha": "abc123", "files_modified": ["a.py"], "last_status": "complete"}), encoding="utf-8") + backup_path = tmp_path / "rpg.before-edit-123.json" + apply_path.write_text( + json.dumps({ + "type": "success", + "backup_timestamp": "123", + "backups": {"rpg": str(backup_path)}, + "applied_features": [{"node_id": "n1", "action": "modified"}], + "dep_graph_refreshed": True, + "rollback_path": str(backup_path), + "rollback_command": "cmind script rpg_edit/apply.py --rollback 123", + "before_state": { + "head_commit": "before123", + "head_short": "before1", + "head_branch": "rpg-edit/test", + "head_timestamp": "2026-06-30T12:00:00+00:00", + }, + "confirmed": True, + "test_result": {"passed": True, "output": ""}, + }), + encoding="utf-8", + ) + rpg_path.write_text( + json.dumps({ + "repo_name": "test", + "root": {"id": "n1", "name": "Node", "node_type": "feature", "meta": {"path": "a.py"}, "children": []}, + "edges": [], + "dep_graph": { + "nodes": {"a.py:f": {"name": "f", "type": "function", "module": "a.py", "rpg_nodes": ["n1"]}}, + "edges": [], + }, + "_dep_to_rpg_map": {"a.py:f": ["n1"]}, + }), + encoding="utf-8", + ) + + monkeypatch.setattr(review, "RPG_EDIT_VALIDATE_FILE", validate_path) + monkeypatch.setattr(review, "RPG_EDIT_LOCATE_FILE", locate_path) + monkeypatch.setattr(review, "RPG_EDIT_IMPACT_FILE", impact_path) + monkeypatch.setattr(review, "RPG_EDIT_CODE_RESULT_FILE", code_path) + monkeypatch.setattr(review, "RPG_EDIT_APPLY_RESULT_FILE", apply_path) + monkeypatch.setattr(review, "RPG_EDIT_REVIEW_RESULT_FILE", review_path) + monkeypatch.setattr(review, "REPO_RPG_FILE", rpg_path) + monkeypatch.setattr(review, "REPORTS_DIR", tmp_path) + + def fake_file_diffs_between(repo_dir, from_commit=None, to_commit="HEAD", *, files=None, py_only=False): + assert to_commit == "abc123" + assert files == ["a.py"] + assert py_only is False + return [{"file": "a.py", "change_type": "modify", "diff": "+new "}] + + monkeypatch.setattr(review, "file_diffs_between", fake_file_diffs_between) + monkeypatch.setattr( + review, + "read_head", + lambda repo_dir: { + "head_commit": "current123", + "head_short": "current", + "head_branch": "current-branch", + "head_timestamp": "2026-06-30T12:30:00+00:00", + }, + ) + + def fake_write_command_report(run): + data = run.to_dict() + review_artifact = next( + item for item in data["artifacts"] if item["label"] == "review_result" + ) + apply_artifact = next( + item for item in data["artifacts"] if item["label"] == "apply_result" + ) + assert review_artifact["status"] == "available" + assert apply_artifact["status"] == "available" + decision = data["user_decisions"][0] + assert decision["decision"] == "apply" + assert decision["branch"] == "rpg-edit/test" + assert decision["before_state"]["head_commit"] == "before123" + assert decision["rollback_path"] == str(backup_path) + assert decision["confirmed"] is True + assert decision["apply_status"] == "success" + assert decision["test_status"] == "passed" + assert data["retrievals"][0]["tool"] == str(locate_path) + assert data["retrievals"][0]["query"] == "a.py" + assert "locate score=1.0" in data["retrievals"][0]["hits"][0]["reason"] + assert "impact callers=1, affected_files=1" in data["retrievals"][0]["hits"][0]["reason"] + assert data["retrievals"][1]["tool"] == str(impact_path) + assert data["code_deltas"] == [{"file": "a.py", "change_type": "modify", "diff": "+new "}] + assert data["focused_graph"]["status"] == "available" + assert data["focused_graph"]["selected_rpg_nodes"] == ["n1"] + assert data["focused_graph"]["selected_dep_nodes"] == ["a.py:f"] + assert Path(data["focused_graph"]["path"]).exists() + assert any(item["label"] == "focused_graph" for item in data["artifacts"]) + return report_path + + monkeypatch.setattr(review, "write_command_report", fake_write_command_report) + + result = review._publish_review_report({"type": "skipped", "success": True}, plan_path, impact_path) + + assert result["report_path"] == str(report_path) + persisted = json.loads(review_path.read_text(encoding="utf-8")) + assert persisted["report_path"] == str(report_path) + + +def test_rollback_command_only_uses_rpg_edit_branch(tmp_path: Path, monkeypatch) -> None: + apply_module = _load_script("rpg_edit_apply_test", _SCRIPTS / "rpg_edit" / "apply.py") + + result = apply_module._rollback_command( + "123", + { + "head_commit": "before123", + "head_short": "before1", + "head_branch": "feature/my-branch", + "head_timestamp": "2026-06-30T12:00:00+00:00", + }, + ) + assert result == "cmind script rpg_edit/apply.py --rollback 123" + + result = apply_module._rollback_command( + "123", + { + "head_commit": "before123", + "head_short": "before1", + "head_branch": "rpg-edit/test", + "head_timestamp": "2026-06-30T12:00:00+00:00", + }, + ) + assert result == "cmind script rpg_edit/apply.py --rollback 123 --rollback-branch rpg-edit/test" + + +def test_review_report_reconstructs_affected_node_evidence_from_impact(tmp_path: Path, monkeypatch) -> None: + review = _load_script("rpg_edit_review_impact_test", _SCRIPTS / "rpg_edit" / "review.py") + + validate_path = tmp_path / "validate.json" + locate_path = tmp_path / "locate.json" + plan_path = tmp_path / "plan.json" + impact_path = tmp_path / "impact.json" + code_path = tmp_path / "code.json" + apply_path = tmp_path / "apply.json" + review_path = tmp_path / "review.json" + report_path = tmp_path / "report.html" + dep_id = "scripts/common/run_report.py:_render_artifacts" + + validate_path.write_text(json.dumps({"type": "ready"}), encoding="utf-8") + locate_path.write_text(json.dumps({"type": "candidates", "results": [{"node_id": "other", "name": "Other"}]}), encoding="utf-8") + plan_path.write_text(json.dumps({"affected_nodes": ["planned"], "code_changes": [{"file_path": "scripts/common/run_report.py"}]}), encoding="utf-8") + impact_path.write_text( + json.dumps({"type": "impact", "results": {"planned": {"name": "Planned Node", "dep_nodes": [dep_id], "affected_files": ["scripts/common/run_report.py"]}}}), + encoding="utf-8", + ) + code_path.write_text(json.dumps({"success": True, "files_modified": ["scripts/common/run_report.py"], "last_status": "complete"}), encoding="utf-8") + apply_path.write_text( + json.dumps({ + "type": "dep_refreshed", + "backup_timestamp": "456", + "backups": {"dep_graph": str(tmp_path / "dep_graph.before-edit-456.json")}, + "applied_features": [], + "dep_graph_refreshed": True, + "rollback_command": "cmind script rpg_edit/apply.py --rollback 456", + "test_result": {"passed": False, "output": "failing test"}, + }), + encoding="utf-8", + ) + + monkeypatch.setattr(review, "RPG_EDIT_VALIDATE_FILE", validate_path) + monkeypatch.setattr(review, "RPG_EDIT_LOCATE_FILE", locate_path) + monkeypatch.setattr(review, "RPG_EDIT_IMPACT_FILE", impact_path) + monkeypatch.setattr(review, "RPG_EDIT_CODE_RESULT_FILE", code_path) + monkeypatch.setattr(review, "RPG_EDIT_APPLY_RESULT_FILE", apply_path) + monkeypatch.setattr(review, "RPG_EDIT_REVIEW_RESULT_FILE", review_path) + monkeypatch.setattr(review, "_focused_graph_artifact", lambda candidates, artifacts: {}) + monkeypatch.setattr( + review, + "read_head", + lambda repo_dir: { + "head_commit": "current456", + "head_short": "current", + "head_branch": "fallback-branch", + "head_timestamp": "2026-06-30T13:00:00+00:00", + }, + ) + + def fake_write_command_report(run): + data = run.to_dict() + decision = data["user_decisions"][0] + assert decision["decision"] == "apply" + assert decision["branch"] == "fallback-branch" + assert decision["before_state"]["head_commit"] == "current456" + assert "confirmed" not in decision + assert decision["apply_status"] == "dep_refreshed" + assert decision["test_status"] == "failed" + assert decision["rollback_path"].endswith("dep_graph.before-edit-456.json") + assert any(item["label"] == "apply_result" for item in data["artifacts"]) + assert data["rpg_deltas"] == [{"node_id": "planned", "name": "Planned Node"}] + assert data["dep_graph_deltas"] == [ + {"dep_node_id": dep_id, "path": "scripts/common/run_report.py", "source_feature": "planned"} + ] + assert data["retrievals"][0]["hits"][0]["node_id"] == "planned" + assert "1 dep nodes" in data["retrievals"][0]["hits"][0]["reason"] + assert "impact callers=0, affected_files=1" in data["retrievals"][0]["hits"][0]["reason"] + assert data["retrievals"][1]["tool"] == str(impact_path) + return report_path + + monkeypatch.setattr(review, "write_command_report", fake_write_command_report) + + result = review._publish_review_report({"type": "skipped", "success": True}, plan_path, impact_path) + + assert result["report_path"] == str(report_path) diff --git a/CoderMind/tests/test_rpg_evolution.py b/CoderMind/tests/test_rpg_evolution.py index 4f72b4b..e9bb59c 100644 --- a/CoderMind/tests/test_rpg_evolution.py +++ b/CoderMind/tests/test_rpg_evolution.py @@ -745,7 +745,7 @@ def test_save_path_creates_file(self, simple_rpg): return_value=diff_result, ): - RPGEvolution.process_diff( + updated_rpg = RPGEvolution.process_diff( repo_name="test", repo_info="Test repo", save_path=save_path, @@ -756,6 +756,13 @@ def test_save_path_creates_file(self, simple_rpg): update_dep_graph=False, ) + assert updated_rpg._last_diff_summary == { + "added": 0, + "deleted": 1, + "modified": 0, + "renamed": 0, + } + assert updated_rpg._last_diff_files["deleted"] == ["src/module_a.py"] assert os.path.isfile(save_path) with open(save_path, "r") as f: data = json.load(f) diff --git a/CoderMind/tests/test_run_report.py b/CoderMind/tests/test_run_report.py new file mode 100644 index 0000000..f27ccd0 --- /dev/null +++ b/CoderMind/tests/test_run_report.py @@ -0,0 +1,261 @@ +from __future__ import annotations + +import sys +from pathlib import Path + +_REPO = Path(__file__).resolve().parents[1] +_SCRIPTS = _REPO / "scripts" +if str(_SCRIPTS) not in sys.path: + sys.path.insert(0, str(_SCRIPTS)) + +from common.run_events import ( + ArtifactEvent, + CodeDeltaEvent, + CommandRun, + DepGraphDeltaEvent, + RPGDeltaEvent, + RetrievalEvent, + StepEvent, + UserDecisionEvent, + VerificationEvent, +) +from common.run_report import write_command_report + + +def test_write_command_report_escapes_content_and_writes_sections(tmp_path: Path) -> None: + report = write_command_report( + CommandRun( + command="rpg/edit ", + title="Title ", + status="ok ", + summary=[ + {"label": "node", "value": ""}, + {"label": "count", "value": 3}, + ], + steps=[StepEvent(name="locate ", status="done", reason="score > 1")], + rpg_deltas=[RPGDeltaEvent(node_id="feature"}, + timestamp="2026-06-30T12:34:56Z", + ), + report_dir=tmp_path, + ) + + assert report.parent == tmp_path + assert report.name.startswith("cmind_run_rpg_edit_script_alert_1_script_") + html = report.read_text(encoding="utf-8") + assert "Summary" in html + assert "Stage timeline" in html + assert "Safety boundary" in html + assert "Artifact links" in html + assert "Evidence JSON" in html + assert "Before state" in html + assert "Confirmation" in html + assert "Branch" in html + assert "Apply status" in html + assert "Test status" in html + assert "Rollback path" in html + assert "apply <unsafe>" in html + assert "rpg-edit/<branch>" in html + assert "abc<script>" in html + assert "backup/<script>" in html + assert "success <ok>" in html + assert "passed <ok>" in html + assert "<script>evil()</script>" in html + assert "" not in html + assert "backup/" for i in range(40) + ) + + report = write_command_report( + { + "command": "rpg_edit", + "retrievals": [ + RetrievalEvent( + query="a.py", + tool="RPG_EDIT_LOCATE_FILE", + hits=[{"node_id": "n1" not in html + assert "
    View diff" in html + assert "
    Focused graph evidence") == 1 + assert "Graph artifact" in html + assert "Inspector metadata" in html + + +def test_write_command_report_limits_summary_cards(tmp_path: Path) -> None: + report = write_command_report( + CommandRun( + command="encode", + summary=[{"label": f"card-{i}", "value": i} for i in range(9)], + timestamp="fixed", + ), + report_dir=tmp_path, + ) + + html = report.read_text(encoding="utf-8") + assert html.count('class="card-label"') == 7 + visible_cards = html.split("
    ", 1)[0] + assert "card-0" in visible_cards + assert "card-6" in visible_cards + assert "card-7" not in visible_cards + assert "card-8" not in visible_cards + + +def test_write_command_report_preserves_same_timestamp_runs(tmp_path: Path) -> None: + first = write_command_report(CommandRun("update_rpg", timestamp="fixed"), report_dir=tmp_path) + second = write_command_report(CommandRun("update_rpg", timestamp="fixed"), report_dir=tmp_path) + + assert first != second + assert first.name == "cmind_run_update_rpg_fixed.html" + assert second.name == "cmind_run_update_rpg_fixed_2.html" + assert first.exists() + assert second.exists() + + +def test_write_command_report_does_not_invent_node_rows_from_counts(tmp_path: Path) -> None: + report = write_command_report( + CommandRun( + command="encode", + evidence={"dep_nodes": 4, "rpg_nodes": 6}, + timestamp="fixed", + ), + report_dir=tmp_path, + ) + + html = report.read_text(encoding="utf-8") + assert html.count("No node evidence recorded.") == 2 + assert '"dep_nodes": 4' in html + assert '4' not in html + + +def test_write_command_report_infers_artifact_status_and_preserves_verification_detail(tmp_path: Path) -> None: + available = tmp_path / "available.json" + available.write_text("{}", encoding="utf-8") + missing = tmp_path / "missing.json" + + report = write_command_report( + CommandRun( + command="encode", + artifacts=[ + ArtifactEvent(label="rpg_json", path=available), + ArtifactEvent(label="missing_json", path=missing), + ], + verification=[ + VerificationEvent(name="message", status="ok", detail="from message"), + VerificationEvent(name="reason", status="warn", detail="from reason"), + ], + timestamp="fixed", + ), + report_dir=tmp_path, + ) + + html = report.read_text(encoding="utf-8") + assert "rpg_json" in html + assert "missing_json" in html + assert html.count("available") == 1 + assert html.count("missing") == 1 + assert "from message" in html + assert "from reason" in html + + +def test_all_event_types_serialize_with_optional_fields(tmp_path: Path) -> None: + available = tmp_path / "artifact.txt" + available.write_text("ok", encoding="utf-8") + + events = [ + StepEvent(), + RetrievalEvent(query="grep", tool="grep", hits=[{"path": "a.py"}], reason="matched"), + RPGDeltaEvent(node_id="feature", name="Feature", type="function", path="a.py", change="modified", score=1.0), + DepGraphDeltaEvent(dep_node_id="a.py:f", path="a.py", source_feature="feature", change="modified"), + CodeDeltaEvent(file="a.py", change_type="modify", before="old", after="new", diff="@@"), + VerificationEvent(), + UserDecisionEvent( + decision="apply", + branch="rpg-edit/x", + before_state={"clean": True}, + rollback_path="backup", + confirmed=True, + apply_status="success", + test_status="passed", + ), + ArtifactEvent(label="artifact", path=available), + ] + + for event in events: + assert isinstance(event.to_dict(), dict) + + run = CommandRun( + command="events", + retrievals=[events[1]], + rpg_deltas=[events[2]], + dep_graph_deltas=[events[3]], + code_deltas=[events[4]], + user_decisions=[events[6]], + artifacts=[events[7]], + ).to_dict() + + assert run["retrievals"][0]["tool"] == "grep" + assert run["code_deltas"][0]["file"] == "a.py" + assert run["user_decisions"][0]["confirmed"] is True + assert run["user_decisions"][0]["apply_status"] == "success" + assert run["user_decisions"][0]["test_status"] == "passed" + assert run["artifacts"][0]["status"] == "available" + + +def test_write_command_report_accepts_command_run_mapping(tmp_path: Path) -> None: + run = CommandRun( + command="mapping", + summary=[{"label": "safe", "value": ""}], + timestamp="fixed", + ).to_dict() + + report = write_command_report(run, report_dir=tmp_path) + + html = report.read_text(encoding="utf-8") + assert "<ok>" in html diff --git a/CoderMind/tests/test_sync_from_commit_diff.py b/CoderMind/tests/test_sync_from_commit_diff.py index 023d18c..fcef438 100644 --- a/CoderMind/tests/test_sync_from_commit_diff.py +++ b/CoderMind/tests/test_sync_from_commit_diff.py @@ -129,6 +129,49 @@ def _node_edge_snapshot(g: DependencyGraph) -> Tuple[Dict[str, dict], Set[Tuple[ return nodes, edges +def test_git_diff_helpers_return_workspace_relative_paths_in_subdir(tmp_path): + from common.git_utils import ( + changed_files_between, + staged_changes, + working_tree_changes, + ) + + git_root = tmp_path / "gitroot" + workspace = git_root / "subproject" + src = workspace / "src" + src.mkdir(parents=True) + (src / "a.py").write_text("def a():\n return 1\n") + + _sh(git_root, "init", "-q", "-b", "main") + _sh(git_root, "config", "user.email", "test@example.com") + _sh(git_root, "config", "user.name", "Test") + _sh(git_root, "add", ".") + _sh(git_root, "commit", "-q", "-m", "initial") + first = _head_sha(git_root) + + (src / "a.py").write_text("def a():\n return 2\n") + _sh(git_root, "add", ".") + _sh(git_root, "commit", "-q", "-m", "update a") + second = _head_sha(git_root) + + changed, renames = changed_files_between(workspace, first, second) + assert changed == ["src/a.py"] + assert renames == {} + + (src / "a.py").write_text("def a():\n return 3\n") + _sh(git_root, "add", "subproject/src/a.py") + changed, renames = staged_changes(workspace) + assert changed == ["src/a.py"] + assert renames == {} + + (src / "new.py").write_text("def new():\n return 1\n") + changed, renames = working_tree_changes(workspace) + assert "src/a.py" in changed + assert "src/new.py" in changed + assert all(not path.startswith("subproject/") for path in changed) + assert renames == {} + + # --------------------------------------------------------------------------- # Decision tree # --------------------------------------------------------------------------- diff --git a/CoderMind/utils/claude/llm_usage_count_coarse.py b/CoderMind/utils/claude/llm_usage_count_coarse.py index 49fe89a..ec02829 100644 --- a/CoderMind/utils/claude/llm_usage_count_coarse.py +++ b/CoderMind/utils/claude/llm_usage_count_coarse.py @@ -22,11 +22,21 @@ from typing import Dict, List, Optional # Import from the detailed counter -from llm_usage_count import ( - parse_file, - FileSummary, - _match_pricing, -) +if __package__: + from .llm_usage_count import ( + parse_file, + FileSummary, + _match_pricing, + ) +else: + module_dir = str(Path(__file__).resolve().parent) + if module_dir not in sys.path: + sys.path.insert(0, module_dir) + from llm_usage_count import ( + parse_file, + FileSummary, + _match_pricing, + ) # ── Stage definitions ───────────────────────────────────────────────────────── diff --git a/zerorepo/rpg_gen/base/node/node.py b/zerorepo/rpg_gen/base/node/node.py index 04990b2..00b3e07 100644 --- a/zerorepo/rpg_gen/base/node/node.py +++ b/zerorepo/rpg_gen/base/node/node.py @@ -1,11 +1,11 @@ from typing import List, Optional, Dict -from abc import abstractmethod +from abc import ABC, abstractmethod import logging from ..unit import CodeUnit, ParsedFile from zerorepo.utils.file import normalize_path -class RepoNode: +class RepoNode(ABC): def __init__(self, name: str, path: str): self.name = name self.path = normalize_path(path) # 相对路径