常规归档

langgraph-plan-review/scripts/agents.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import json
+import logging
+import subprocess
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from string import Template
+from typing import Mapping, Sequence
+
+
+@dataclass(frozen=True)
+class ReviewResult:
+    has_serious_issue: bool
+    comment: str = ""
+    codex_session_id: str = ""
+
+
+def _format_args(command: Sequence[str], values: Mapping[str, str]) -> list[str]:
+    return [Template(part).safe_substitute(values) for part in command]
+
+
+def _run_command(command: Sequence[str], values: Mapping[str, str], timeout: int) -> str:
+    args = _format_args(command, values)
+    logging.info("Running: %s", args[0] if args else "")
+    t0 = time.monotonic()
+    result = subprocess.run(
+        args,
+        capture_output=True,
+        check=False,
+        text=True,
+        timeout=timeout,
+    )
+    elapsed = time.monotonic() - t0
+    logging.info("Finished in %.1fs, exit=%d", elapsed, result.returncode)
+    if result.returncode != 0:
+        raise RuntimeError(
+            f"Command failed with exit code {result.returncode}: {args}\n"
+            f"STDOUT:\n{result.stdout}\nSTDERR:\n{result.stderr}"
+        )
+    return result.stdout.strip()
+
+
+class AgentRunner:
+    """Boundary around Codex CLI calls.
+
+    The Claude Code steps run in the current Claude Code session through the
+    skill instructions. This runner only invokes Codex review commands.
+    Commands are injected by passing argv lists with Template placeholders.
+    Supported placeholders: task_id, user_prompt, work_dir, task_path,
+    latest_discuss_path, discuss_path, notes_dir, codex_session_id.
+    """
+
+    def __init__(
+        self,
+        *,
+        codex_review_command: Sequence[str],
+        timeout: int = 600,
+    ) -> None:
+        self.codex_review_command = codex_review_command
+        self.timeout = timeout
+
+    def review(self, values: Mapping[str, str]) -> ReviewResult:
+        enriched = {**values, "timeout": str(self.timeout)}
+        output = _run_command(self.codex_review_command, enriched, self.timeout)
+        if not output:
+            return ReviewResult(has_serious_issue=False)
+
+        try:
+            payload = json.loads(output)
+        except json.JSONDecodeError:
+            return ReviewResult(has_serious_issue=True, comment=output)
+
+        return ReviewResult(
+            has_serious_issue=bool(payload.get("has_serious_issue")),
+            comment=str(payload.get("comment", "")).strip(),
+            codex_session_id=str(payload.get("codex_session_id", "")).strip(),
+        )