From eb2ccc92a96c639ce44e3b8aaa47ac4653509e54 Mon Sep 17 00:00:00 2001 From: Devil Date: Tue, 30 Jun 2026 21:04:53 -0400 Subject: [PATCH] Rebrand plugin and planning metadata --- CLODEX.md | 12 + README.md | 34 ++- clodex/agents.py | 3 +- clodex/artifacts.py | 2 + clodex/cli.py | 133 +++++++- clodex/commands.py | 30 +- clodex/config.py | 42 +++ clodex/evals.py | 18 ++ clodex/hooks.py | 55 ++++ clodex/mcp_server.py | 74 ++++- clodex/prompts.py | 8 +- clodex/state.py | 163 +++++++++- clodex/tasks.py | 101 ++++++ clodex/trace.py | 28 ++ clodex/workflow.py | 306 +++++++++++++++---- clodex/workspace.py | 68 +++++ dist/.claude/commands/clodex/clodex-build.md | 2 + skills/clodex-workflow/SKILL.md | 10 +- tests/test_clodex.py | 217 ++++++++++++- 19 files changed, 1212 insertions(+), 94 deletions(-) create mode 100644 clodex/evals.py create mode 100644 clodex/hooks.py create mode 100644 clodex/tasks.py create mode 100644 clodex/trace.py create mode 100644 clodex/workspace.py diff --git a/CLODEX.md b/CLODEX.md index 1aa8bc9..994f1ff 100644 --- a/CLODEX.md +++ b/CLODEX.md @@ -5,6 +5,9 @@ max_fix_loops: 2 workspace_root: .clodex/workspaces runs_root: .clodex/runs state_path: .clodex/state.sqlite3 +workspace: + backend: git-worktree + apply_mode: manual claude: model: opus effort: max @@ -13,9 +16,15 @@ codex: model: gpt-5.5 reasoning_effort: xhigh sandbox: workspace-write + approval_profile: ci audit: quorum: unanimous personas: [security, performance, portability, test-gap] + reviewers: [{"id": "claude-plan", "backend": "claude", "persona": "plan-adherence", "required": true, "timeout": 600}, {"id": "codex-architecture", "backend": "codex", "persona": "architecture", "required": true, "timeout": 600}, {"id": "security", "backend": "codex", "persona": "security", "required": false, "timeout": 600}, {"id": "performance", "backend": "codex", "persona": "performance", "required": false, "timeout": 600}, {"id": "portability", "backend": "codex", "persona": "portability", "required": false, "timeout": 600}, {"id": "test-gap", "backend": "claude", "persona": "test-gap", "required": false, "timeout": 600}] +mcp: + async_tasks: true +tracing: + enabled: true --- # Clodex Workflow Contract @@ -31,6 +40,9 @@ remains unresolved. Both agents then perform adversarial audit against the same final diff hash. Clodex advances only when both agents return explicit approval. +Default builds run in isolated git worktrees and leave source checkouts +unchanged until a run is explicitly applied. + Default behavior is local-first. Subscription CLI authentication is preferred for both Claude Code and Codex. API keys are optional fallback mechanisms for CI or headless environments, not the default local workflow. diff --git a/README.md b/README.md index f51b3c0..3da74bb 100644 --- a/README.md +++ b/README.md @@ -39,6 +39,7 @@ API keys or long-lived tokens are fallback-only for CI/headless automation. clodex doctor clodex plan --dry-run "Add a small feature" clodex build "Add a small feature" +clodex apply ``` From this checkout without installing: @@ -61,9 +62,14 @@ PowerShell: | --- | --- | | `clodex doctor` | Check Python, git, Claude Code, Codex, and `CLODEX.md` | | `clodex plan ""` | Run Claude planning only | -| `clodex build ""` | Run plan, implementation, and dual audit loop | +| `clodex build ""` | Run plan, implementation, and dual audit loop in an isolated worktree | | `clodex audit --diff` | Audit current uncommitted changes | | `clodex run ""` | Alias for `build` | +| `clodex apply ` | Apply an approved worktree patch back to the source checkout | +| `clodex task start/get/cancel/list` | Manage durable async runs | +| `clodex trace export ` | Print a run trace as JSONL | +| `clodex hooks print/install/ingest` | Generate or ingest Claude Code hook events | +| `clodex eval run` | Run local harness smoke evals | | `clodex queue add/list/update` | Manage the local task ledger | | `clodex status` | Show recent tasks and runs | | `clodex mcp-server` | Run the stdio MCP server | @@ -82,6 +88,10 @@ codex: model: gpt-5.5 reasoning_effort: xhigh sandbox: workspace-write + approval_profile: ci +workspace: + backend: git-worktree + apply_mode: manual max_fix_loops: 2 ``` @@ -93,9 +103,17 @@ Run artifacts are written to `.clodex/runs//`: - `04-codex-audit.json` - `05-agreement.json` - `changes.diff` +- `apply.patch` +- `trace.jsonl` +- `workspace.json` +- `reviewers/*.json` Local task/run state is stored in `.clodex/state.sqlite3`. +By default, `clodex build` executes inside `.clodex/workspaces//`. +The source checkout is not modified until `clodex apply ` succeeds. +Use `--workspace local` for compatibility with the earlier in-place behavior. + ## MCP Tools The MCP server exposes: @@ -106,6 +124,12 @@ The MCP server exposes: - `clodex_status` - `clodex_task_create` - `clodex_task_update` +- `clodex_task_start` +- `clodex_task_get` +- `clodex_task_cancel` + +The server also handles MCP-style `tasks/get`, `tasks/update`, and +`tasks/cancel` JSON-RPC methods using the Clodex `run_id` as the task id. Start it with: @@ -115,7 +139,7 @@ python -m clodex mcp-server ## Safety -Clodex defaults to Codex `workspace-write` sandboxing and Claude plan mode. -Dangerous full-access workflows are intentionally not the default. A run is -complete only when `05-agreement.json` has `approved: true` for the final diff -hash. +Clodex defaults to git worktree isolation, Codex `workspace-write` sandboxing, +and Claude plan mode. Dangerous full-access workflows are intentionally not the +default. A run is complete only when `05-agreement.json` has `approved: true` +for the final diff hash. diff --git a/clodex/agents.py b/clodex/agents.py index 3928ada..f15a6b1 100644 --- a/clodex/agents.py +++ b/clodex/agents.py @@ -24,7 +24,7 @@ class AgentRunner: def __init__(self, repo_root: Path): self.repo_root = repo_root - def run(self, command: AgentCommand, prompt: str) -> AgentResult: + def run(self, command: AgentCommand, prompt: str, timeout: int | None = None) -> AgentResult: argv = list(command.argv) resolved = shutil.which(argv[0]) if resolved: @@ -38,6 +38,7 @@ def run(self, command: AgentCommand, prompt: str) -> AgentResult: encoding="utf-8", errors="replace", check=False, + timeout=timeout, ) return AgentResult( command=command, diff --git a/clodex/artifacts.py b/clodex/artifacts.py index e14db7b..699bf44 100644 --- a/clodex/artifacts.py +++ b/clodex/artifacts.py @@ -36,11 +36,13 @@ def __init__(self, config: ClodexConfig, run_id: str): def write_json(self, name: str, data: dict[str, Any]) -> Path: path = self.path / name + path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(data, indent=2, sort_keys=True) + "\n", encoding="utf-8") return path def write_text(self, name: str, text: str) -> Path: path = self.path / name + path.parent.mkdir(parents=True, exist_ok=True) path.write_text(text, encoding="utf-8") return path diff --git a/clodex/cli.py b/clodex/cli.py index 97f3f46..8c65fb2 100644 --- a/clodex/cli.py +++ b/clodex/cli.py @@ -2,11 +2,15 @@ import argparse import json +import sys from pathlib import Path from typing import Any from .doctor import run_doctor +from .evals import run_local_evals +from .hooks import hook_config, ingest_hook_event from .mcp_server import main as mcp_main +from .tasks import TaskManager from .workflow import ClodexWorkflow @@ -22,16 +26,31 @@ def main(argv: list[str] | None = None) -> int: plan.add_argument("--dry-run", action="store_true") build = sub.add_parser("build", help="Run plan, implementation, and dual audit") - build.add_argument("task", nargs="+") - build.add_argument("--dry-run", action="store_true") + add_build_args(build) audit = sub.add_parser("audit", help="Audit current uncommitted changes") audit.add_argument("--dry-run", action="store_true") audit.add_argument("--diff", action="store_true", help="Accepted for compatibility; audit always uses git diff") run = sub.add_parser("run", help="Alias for build") - run.add_argument("task", nargs="+") - run.add_argument("--dry-run", action="store_true") + add_build_args(run) + + task = sub.add_parser("task", help="Manage durable Clodex runs") + task_sub = task.add_subparsers(dest="task_command", required=True) + task_start = task_sub.add_parser("start") + task_start.add_argument("task", nargs="+") + task_start.add_argument("--workspace", choices=["git-worktree", "local"]) + task_start.add_argument("--approval-profile", choices=["ci", "local", "auto_review"]) + task_start.add_argument("--dry-run", action="store_true") + task_get = task_sub.add_parser("get") + task_get.add_argument("run_id") + task_cancel = task_sub.add_parser("cancel") + task_cancel.add_argument("run_id") + task_sub.add_parser("list") + task_worker = task_sub.add_parser("worker") + task_worker.add_argument("run_id") + task_worker.add_argument("--workspace", choices=["git-worktree", "local"]) + task_worker.add_argument("--approval-profile", choices=["ci", "local", "auto_review"]) queue = sub.add_parser("queue", help="Manage the local task ledger") queue_sub = queue.add_subparsers(dest="queue_command", required=True) @@ -43,6 +62,28 @@ def main(argv: list[str] | None = None) -> int: queue_update.add_argument("id") queue_update.add_argument("status") + apply_cmd = sub.add_parser("apply", help="Apply an approved worktree run patch") + apply_cmd.add_argument("run_id") + apply_cmd.add_argument("--check", action="store_true") + + trace = sub.add_parser("trace", help="Inspect run traces") + trace_sub = trace.add_subparsers(dest="trace_command", required=True) + trace_export = trace_sub.add_parser("export") + trace_export.add_argument("run_id") + trace_export.add_argument("--format", choices=["jsonl"], default="jsonl") + + eval_cmd = sub.add_parser("eval", help="Run local Clodex harness evals") + eval_sub = eval_cmd.add_subparsers(dest="eval_command", required=True) + eval_sub.add_parser("run") + + hooks = sub.add_parser("hooks", help="Print or ingest Claude Code hook events") + hooks_sub = hooks.add_subparsers(dest="hooks_command", required=True) + hooks_sub.add_parser("print") + hooks_install = hooks_sub.add_parser("install") + hooks_install.add_argument("--dry-run", action="store_true", required=True) + hooks_ingest = hooks_sub.add_parser("ingest") + hooks_ingest.add_argument("--run-id", default="manual-hook-event") + sub.add_parser("status", help="Show recent tasks and runs") sub.add_parser("mcp-server", help="Run the Clodex MCP stdio server") @@ -62,13 +103,33 @@ def main(argv: list[str] | None = None) -> int: print_result(result.__dict__, args.json) return 0 if args.command in {"build", "run"}: - result = workflow.build(" ".join(args.task), dry_run=args.dry_run) + result = workflow.build( + " ".join(args.task), + dry_run=args.dry_run, + workspace_backend=args.workspace, + approval_profile=args.approval_profile, + apply_changes=args.apply_changes, + ) print_result(result.__dict__, args.json) return 0 if result.status not in {"blocked"} else 1 if args.command == "audit": result = workflow.audit(dry_run=args.dry_run) print_result(result.__dict__, args.json) return 0 if result.status not in {"blocked"} else 1 + if args.command == "task": + return handle_task(args, workflow, args.json) + if args.command == "apply": + result = workflow.apply_run(args.run_id, check=args.check) + print_result(result.__dict__, args.json) + return 0 if result.status not in {"apply-failed", "apply-check-failed"} else 1 + if args.command == "trace": + return handle_trace(args, workflow, args.json) + if args.command == "eval": + data = run_local_evals(Path.cwd()) + print_output(data, args.json) + return 0 if data["passed"] else 1 + if args.command == "hooks": + return handle_hooks(args, args.json) if args.command == "status": print_output({"tasks": workflow.state.list_tasks(), "runs": workflow.state.list_runs()}, args.json) return 0 @@ -88,6 +149,68 @@ def main(argv: list[str] | None = None) -> int: return 2 +def add_build_args(parser: argparse.ArgumentParser) -> None: + parser.add_argument("task", nargs="+") + parser.add_argument("--dry-run", action="store_true") + parser.add_argument("--workspace", choices=["git-worktree", "local"]) + parser.add_argument("--approval-profile", choices=["ci", "local", "auto_review"]) + parser.add_argument("--apply", action="store_true", dest="apply_changes") + + +def handle_task(args: argparse.Namespace, workflow: ClodexWorkflow, as_json: bool) -> int: + manager = TaskManager(Path.cwd()) + if args.task_command == "start": + result = manager.start(" ".join(args.task), workspace_backend=args.workspace, approval_profile=args.approval_profile, dry_run=args.dry_run) + print_result(result.__dict__, as_json) + return 0 + if args.task_command == "get": + data = manager.get(args.run_id) + if data is None: + print_output({"error": f"unknown run: {args.run_id}"}, as_json) + return 1 + print_output(data, as_json) + return 0 + if args.task_command == "cancel": + result = manager.cancel(args.run_id) + print_result(result.__dict__, as_json) + return 0 + if args.task_command == "list": + print_output(manager.list(), as_json) + return 0 + if args.task_command == "worker": + result = workflow.run_existing(args.run_id, workspace_backend=args.workspace, approval_profile=args.approval_profile) + print_result(result.__dict__, as_json) + return 0 if result.status not in {"blocked"} else 1 + return 2 + + +def handle_trace(args: argparse.Namespace, workflow: ClodexWorkflow, as_json: bool) -> int: + if args.trace_command == "export": + run = workflow.state.get_run(args.run_id) + artifacts_dir = Path(str((run or {}).get("artifacts_dir") or workflow.config.runs_root / args.run_id)) + trace_path = artifacts_dir / "trace.jsonl" + if not trace_path.exists(): + print_output({"error": f"trace not found: {trace_path}"}, as_json) + return 1 + print(trace_path.read_text(encoding="utf-8"), end="") + return 0 + return 2 + + +def handle_hooks(args: argparse.Namespace, as_json: bool) -> int: + if args.hooks_command == "print": + print_output(hook_config(Path.cwd()), as_json) + return 0 + if args.hooks_command == "install": + print_output({"dry_run": True, "config": hook_config(Path.cwd())}, as_json) + return 0 + if args.hooks_command == "ingest": + payload = json.loads(sys.stdin.read() or "{}") + print_output(ingest_hook_event(Path.cwd(), args.run_id, payload), as_json) + return 0 + return 2 + + def print_result(data: dict[str, Any], as_json: bool) -> None: if as_json: print(json.dumps(data, indent=2, sort_keys=True)) diff --git a/clodex/commands.py b/clodex/commands.py index b1c68a6..c3ec707 100644 --- a/clodex/commands.py +++ b/clodex/commands.py @@ -46,26 +46,32 @@ def claude_audit_command(config: ClodexConfig) -> AgentCommand: return claude_plan_command(config) -def codex_exec_command(config: ClodexConfig, repo_root: Path) -> AgentCommand: +def codex_exec_command(config: ClodexConfig, repo_root: Path, approval_profile: str | None = None) -> AgentCommand: codex = config.codex - return AgentCommand( - name="codex-build", - argv=[ - "codex", - "exec", - "-m", - str(codex["model"]), - "-c", - f'model_reasoning_effort="{codex["reasoning_effort"]}"', + profile = approval_profile or str(codex.get("approval_profile", "ci")) + approval = "never" if profile == "ci" else "on-request" + argv = [ + "codex", + "exec", + "-m", + str(codex["model"]), + "-c", + f'model_reasoning_effort="{codex["reasoning_effort"]}"', + ] + if profile == "auto_review": + argv.extend(["-c", 'approvals_reviewer="auto_review"']) + argv.extend( + [ "--sandbox", str(codex["sandbox"]), "--ask-for-approval", - "never", + approval, "-C", str(repo_root), "-", - ], + ] ) + return AgentCommand(name="codex-build", argv=argv) def codex_review_command(config: ClodexConfig) -> AgentCommand: diff --git a/clodex/config.py b/clodex/config.py index fe06dad..564051d 100644 --- a/clodex/config.py +++ b/clodex/config.py @@ -1,5 +1,6 @@ from __future__ import annotations +import json from dataclasses import dataclass, field from pathlib import Path from typing import Any @@ -12,6 +13,10 @@ "workspace_root": ".clodex/workspaces", "runs_root": ".clodex/runs", "state_path": ".clodex/state.sqlite3", + "workspace": { + "backend": "git-worktree", + "apply_mode": "manual", + }, "claude": { "model": "opus", "effort": "max", @@ -21,10 +26,25 @@ "model": "gpt-5.5", "reasoning_effort": "xhigh", "sandbox": "workspace-write", + "approval_profile": "ci", }, "audit": { "quorum": "unanimous", "personas": ["security", "performance", "portability", "test-gap"], + "reviewers": [ + {"id": "claude-plan", "backend": "claude", "persona": "plan-adherence", "required": True, "timeout": 600}, + {"id": "codex-architecture", "backend": "codex", "persona": "architecture", "required": True, "timeout": 600}, + {"id": "security", "backend": "codex", "persona": "security", "required": False, "timeout": 600}, + {"id": "performance", "backend": "codex", "persona": "performance", "required": False, "timeout": 600}, + {"id": "portability", "backend": "codex", "persona": "portability", "required": False, "timeout": 600}, + {"id": "test-gap", "backend": "claude", "persona": "test-gap", "required": False, "timeout": 600}, + ], + }, + "mcp": { + "async_tasks": True, + }, + "tracing": { + "enabled": True, }, } @@ -51,6 +71,10 @@ def runs_root(self) -> Path: def state_path(self) -> Path: return self.repo_root / str(self.raw.get("state_path", DEFAULT_CONFIG["state_path"])) + @property + def workspace(self) -> dict[str, Any]: + return dict(DEFAULT_CONFIG["workspace"] | self.raw.get("workspace", {})) + @property def claude(self) -> dict[str, Any]: return dict(DEFAULT_CONFIG["claude"] | self.raw.get("claude", {})) @@ -63,6 +87,19 @@ def codex(self) -> dict[str, Any]: def audit(self) -> dict[str, Any]: return dict(DEFAULT_CONFIG["audit"] | self.raw.get("audit", {})) + @property + def reviewers(self) -> list[dict[str, Any]]: + reviewers = self.audit.get("reviewers", DEFAULT_CONFIG["audit"]["reviewers"]) + return [dict(item) for item in reviewers] + + @property + def mcp(self) -> dict[str, Any]: + return dict(DEFAULT_CONFIG["mcp"] | self.raw.get("mcp", {})) + + @property + def tracing(self) -> dict[str, Any]: + return dict(DEFAULT_CONFIG["tracing"] | self.raw.get("tracing", {})) + def find_repo_root(start: Path | None = None) -> Path: current = (start or Path.cwd()).resolve() @@ -123,6 +160,11 @@ def parse_minimal_yaml(text: str) -> dict[str, Any]: def parse_scalar(value: str) -> Any: if value == "": return "" + if value.startswith(("[", "{")) and value.endswith(("]", "}")): + try: + return json.loads(value) + except json.JSONDecodeError: + pass lowered = value.lower() if lowered == "true": return True diff --git a/clodex/evals.py b/clodex/evals.py new file mode 100644 index 0000000..96d3384 --- /dev/null +++ b/clodex/evals.py @@ -0,0 +1,18 @@ +from __future__ import annotations + +from pathlib import Path +from typing import Any + +from .config import load_config +from .state import StateStore + + +def run_local_evals(repo_root: Path | None = None) -> dict[str, Any]: + config = load_config(repo_root) + store = StateStore(config.state_path) + scenarios = [ + {"name": "config-loads", "passed": config.workspace["backend"] in {"git-worktree", "local"}}, + {"name": "state-schema-v2", "passed": store.schema_version() >= 2}, + {"name": "reviewers-configured", "passed": len(config.reviewers) >= 2}, + ] + return {"passed": all(item["passed"] for item in scenarios), "scenarios": scenarios} diff --git a/clodex/hooks.py b/clodex/hooks.py new file mode 100644 index 0000000..abb785a --- /dev/null +++ b/clodex/hooks.py @@ -0,0 +1,55 @@ +from __future__ import annotations + +import json +import sys +from pathlib import Path +from typing import Any + +from .config import load_config +from .state import StateStore +from .trace import TraceWriter + + +HOOK_EVENTS = [ + "SessionStart", + "SessionEnd", + "UserPromptSubmit", + "PreToolUse", + "PostToolUse", + "SubagentStart", + "SubagentStop", + "TaskCreated", + "TaskCompleted", + "WorktreeCreated", + "WorktreeRemoved", + "FileChanged", + "Stop", +] + + +def hook_config(repo_root: Path | None = None) -> dict[str, Any]: + config = load_config(repo_root) + command = f"{sys.executable} -m clodex hooks ingest" + return { + "hooks": { + event: [{"type": "command", "command": command}] + for event in HOOK_EVENTS + }, + "clodex": { + "repo_root": str(config.repo_root), + "events": HOOK_EVENTS, + }, + } + + +def ingest_hook_event(repo_root: Path | None, run_id: str, payload: dict[str, Any]) -> dict[str, Any]: + config = load_config(repo_root) + state = StateStore(config.state_path) + run_dir = config.runs_root / run_id + events_dir = run_dir / "events" + events_dir.mkdir(parents=True, exist_ok=True) + hook_file = events_dir / "claude-hooks.jsonl" + with hook_file.open("a", encoding="utf-8", newline="\n") as handle: + handle.write(json.dumps(payload, sort_keys=True) + "\n") + TraceWriter(run_dir, run_id, state).event("hook.ingest", payload) + return {"run_id": run_id, "event_file": str(hook_file)} diff --git a/clodex/mcp_server.py b/clodex/mcp_server.py index 91a84ea..8090103 100644 --- a/clodex/mcp_server.py +++ b/clodex/mcp_server.py @@ -4,6 +4,7 @@ import sys from typing import Any +from .tasks import TaskManager from .workflow import ClodexWorkflow @@ -44,6 +45,28 @@ "description": "Update a local Clodex task status.", "inputSchema": {"type": "object", "properties": {"id": {"type": "string"}, "status": {"type": "string"}}, "required": ["id", "status"]}, }, + { + "name": "clodex_task_start", + "title": "Start async Clodex task", + "description": "Start a durable Clodex build and return a task handle.", + "inputSchema": { + "type": "object", + "properties": {"task": {"type": "string"}, "workspace": {"type": "string"}, "approval_profile": {"type": "string"}, "dry_run": {"type": "boolean"}}, + "required": ["task"], + }, + }, + { + "name": "clodex_task_get", + "title": "Get async Clodex task", + "description": "Get a durable Clodex run by run id.", + "inputSchema": {"type": "object", "properties": {"run_id": {"type": "string"}}, "required": ["run_id"]}, + }, + { + "name": "clodex_task_cancel", + "title": "Cancel async Clodex task", + "description": "Request cancellation of a durable Clodex run.", + "inputSchema": {"type": "object", "properties": {"run_id": {"type": "string"}}, "required": ["run_id"]}, + }, ] @@ -70,7 +93,7 @@ def handle_request(request: dict[str, Any]) -> dict[str, Any] | None: "id": request_id, "result": { "protocolVersion": "2025-11-25", - "capabilities": {"tools": {"listChanged": False}}, + "capabilities": {"tools": {"listChanged": False}, "tasks": {}}, "serverInfo": {"name": "clodex-mcp-server", "version": "0.1.0"}, }, } @@ -83,6 +106,28 @@ def handle_request(request: dict[str, Any]) -> dict[str, Any] | None: name = params.get("name") arguments = params.get("arguments") or {} return {"jsonrpc": "2.0", "id": request_id, "result": tool_call(name, arguments)} + if method == "tasks/get": + params = request.get("params") or {} + run_id = str(params.get("id") or params.get("taskId") or "") + data = TaskManager().get(run_id) + if data is None: + return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32004, "message": f"Unknown task: {run_id}"}} + return {"jsonrpc": "2.0", "id": request_id, "result": task_result(data)} + if method == "tasks/cancel": + params = request.get("params") or {} + run_id = str(params.get("id") or params.get("taskId") or "") + try: + result = TaskManager().cancel(run_id) + except ValueError as exc: + return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32004, "message": str(exc)}} + return {"jsonrpc": "2.0", "id": request_id, "result": task_result({"run": result.__dict__})} + if method == "tasks/update": + params = request.get("params") or {} + run_id = str(params.get("id") or params.get("taskId") or "") + data = TaskManager().get(run_id) + if data is None: + return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32004, "message": f"Unknown task: {run_id}"}} + return {"jsonrpc": "2.0", "id": request_id, "result": task_result(data)} return {"jsonrpc": "2.0", "id": request_id, "error": {"code": -32601, "message": f"Method not found: {method}"}} @@ -103,10 +148,37 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: elif name == "clodex_task_update": workflow.state.update_task(str(arguments["id"]), str(arguments["status"])) return {"content": [{"type": "text", "text": "task updated"}], "isError": False} + elif name == "clodex_task_start": + result = TaskManager().start( + str(arguments["task"]), + workspace_backend=arguments.get("workspace"), + approval_profile=arguments.get("approval_profile"), + dry_run=bool(arguments.get("dry_run", False)), + ) + return {"content": [{"type": "text", "text": json.dumps(task_result({"run": result.__dict__}), indent=2)}], "isError": False} + elif name == "clodex_task_get": + data = TaskManager().get(str(arguments["run_id"])) + if data is None: + return {"content": [{"type": "text", "text": f"Unknown run: {arguments['run_id']}"}], "isError": True} + return {"content": [{"type": "text", "text": json.dumps(data, indent=2)}], "isError": False} + elif name == "clodex_task_cancel": + result = TaskManager().cancel(str(arguments["run_id"])) + return {"content": [{"type": "text", "text": json.dumps(result.__dict__, indent=2)}], "isError": False} else: return {"content": [{"type": "text", "text": f"Unknown tool: {name}"}], "isError": True} return {"content": [{"type": "text", "text": json.dumps(result.__dict__, indent=2)}], "isError": result.status == "blocked"} +def task_result(data: dict[str, Any]) -> dict[str, Any]: + run = data.get("run") or {} + run_id = run.get("id") or run.get("run_id") + status = run.get("status", "unknown") + return { + "id": run_id, + "status": status, + "result": data, + } + + if __name__ == "__main__": raise SystemExit(main()) diff --git a/clodex/prompts.py b/clodex/prompts.py index f49052f..3c5e3b2 100644 --- a/clodex/prompts.py +++ b/clodex/prompts.py @@ -46,11 +46,15 @@ def implementation_prompt(plan: dict[str, Any], task: str) -> str: """ -def audit_prompt(agent_name: str, plan: dict[str, Any], diff: str, diff_hash: str) -> str: +def audit_prompt(agent_name: str, plan: dict[str, Any], diff: str, diff_hash: str, reviewer_id: str | None = None, persona: str | None = None) -> str: + reviewer = reviewer_id or agent_name.lower().replace(" ", "-") + selected_persona = persona or agent_name return f"""You are the {agent_name} adversarial auditor in Clodex. Audit the diff against the accepted plan. Be strict: reject correctness bugs, missing tests, unsafe behavior, scope creep, broken CLI contracts, or unverified claims. +Reviewer ID: {reviewer} +Persona: {selected_persona} Diff hash: {diff_hash} Accepted plan: @@ -65,6 +69,8 @@ def audit_prompt(agent_name: str, plan: dict[str, Any], diff: str, diff_hash: st {{ "approved": true, "diff_hash": "{diff_hash}", + "reviewer_id": "{reviewer}", + "persona": "{selected_persona}", "summary": "short verdict", "findings": [ {{"severity": "high|medium|low", "file": "path", "line": 1, "message": "finding"}} diff --git a/clodex/state.py b/clodex/state.py index a2585c5..206f649 100644 --- a/clodex/state.py +++ b/clodex/state.py @@ -35,6 +35,9 @@ def _init(self) -> None: with self.session() as con: con.executescript( """ + create table if not exists schema_version ( + version integer not null + ); create table if not exists tasks ( id text primary key, title text not null, @@ -67,8 +70,67 @@ def _init(self) -> None: body text not null, created_at text not null ); + create table if not exists run_events ( + id integer primary key autoincrement, + run_id text not null, + event text not null, + data_json text not null, + created_at text not null + ); + create table if not exists artifacts ( + id integer primary key autoincrement, + run_id text not null, + name text not null, + path text not null, + kind text not null, + created_at text not null + ); + create table if not exists workspace_locks ( + run_id text primary key, + source_path text not null, + workspace_path text not null, + backend text not null, + created_at text not null, + released_at text + ); + create table if not exists cancellations ( + run_id text primary key, + requested integer not null, + requested_at text not null, + completed_at text + ); """ ) + self._ensure_run_columns(con) + current = con.execute("select version from schema_version order by version desc limit 1").fetchone() + if current is None: + con.execute("insert into schema_version(version) values (2)") + elif int(current["version"]) < 2: + con.execute("delete from schema_version") + con.execute("insert into schema_version(version) values (2)") + + def _ensure_run_columns(self, con: sqlite3.Connection) -> None: + existing = {row["name"] for row in con.execute("pragma table_info(runs)")} + columns = { + "workspace_path": "text", + "artifacts_dir": "text", + "pid": "integer", + "error": "text", + "started_at": "text", + "completed_at": "text", + } + for name, kind in columns.items(): + if name not in existing: + con.execute(f"alter table runs add column {name} {kind}") + + def table_names(self) -> set[str]: + with self.session() as con: + return {str(row["name"]) for row in con.execute("select name from sqlite_master where type='table'")} + + def schema_version(self) -> int: + with self.session() as con: + row = con.execute("select version from schema_version order by version desc limit 1").fetchone() + return int(row["version"]) if row else 0 def upsert_task(self, task_id: str, title: str, status: str = "todo") -> None: timestamp = now_iso() @@ -92,24 +154,59 @@ def update_task(self, task_id: str, status: str) -> None: (status, now_iso(), task_id), ) - def create_run(self, run_id: str, task_id: str, prompt: str, status: str) -> None: + def create_run( + self, + run_id: str, + task_id: str, + prompt: str, + status: str, + workspace_path: str | None = None, + artifacts_dir: str | None = None, + pid: int | None = None, + ) -> None: timestamp = now_iso() with self.session() as con: con.execute( """ - insert into runs(id, task_id, status, prompt, created_at, updated_at) - values (?, ?, ?, ?, ?, ?) + insert into runs(id, task_id, status, prompt, workspace_path, artifacts_dir, pid, created_at, updated_at, started_at) + values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) """, - (run_id, task_id, status, prompt, timestamp, timestamp), + (run_id, task_id, status, prompt, workspace_path, artifacts_dir, pid, timestamp, timestamp, timestamp), ) - def update_run(self, run_id: str, status: str, diff_hash: str | None = None) -> None: + def update_run( + self, + run_id: str, + status: str, + diff_hash: str | None = None, + workspace_path: str | None = None, + artifacts_dir: str | None = None, + pid: int | None = None, + error: str | None = None, + ) -> None: + completed = now_iso() if status in {"approved", "blocked", "failed", "cancelled", "applied"} else None with self.session() as con: con.execute( - "update runs set status=?, diff_hash=coalesce(?, diff_hash), updated_at=? where id=?", - (status, diff_hash, now_iso(), run_id), + """ + update runs set + status=?, + diff_hash=coalesce(?, diff_hash), + workspace_path=coalesce(?, workspace_path), + artifacts_dir=coalesce(?, artifacts_dir), + pid=coalesce(?, pid), + error=coalesce(?, error), + completed_at=coalesce(?, completed_at), + updated_at=? + where id=? + """, + (status, diff_hash, workspace_path, artifacts_dir, pid, error, completed, now_iso(), run_id), ) + def get_run(self, run_id: str) -> dict[str, Any] | None: + with self.session() as con: + row = con.execute("select * from runs where id=?", (run_id,)).fetchone() + return dict(row) if row else None + def add_audit(self, run_id: str, agent: str, approved: bool, diff_hash: str | None, verdict_json: str) -> None: with self.session() as con: con.execute( @@ -127,6 +224,58 @@ def add_message(self, task_id: str | None, topic: str, body: str) -> None: (task_id, topic, body, now_iso()), ) + def add_event(self, run_id: str, event: str, data: dict[str, Any]) -> None: + import json + + with self.session() as con: + con.execute( + "insert into run_events(run_id, event, data_json, created_at) values (?, ?, ?, ?)", + (run_id, event, json.dumps(data, sort_keys=True), now_iso()), + ) + + def add_artifact(self, run_id: str, name: str, path: str, kind: str) -> None: + with self.session() as con: + con.execute( + "insert into artifacts(run_id, name, path, kind, created_at) values (?, ?, ?, ?, ?)", + (run_id, name, path, kind, now_iso()), + ) + + def add_workspace_lock(self, run_id: str, source_path: str, workspace_path: str, backend: str) -> None: + with self.session() as con: + con.execute( + """ + insert into workspace_locks(run_id, source_path, workspace_path, backend, created_at) + values (?, ?, ?, ?, ?) + on conflict(run_id) do update set + source_path=excluded.source_path, + workspace_path=excluded.workspace_path, + backend=excluded.backend + """, + (run_id, source_path, workspace_path, backend, now_iso()), + ) + + def request_cancel(self, run_id: str) -> None: + with self.session() as con: + con.execute( + """ + insert into cancellations(run_id, requested, requested_at) + values (?, 1, ?) + on conflict(run_id) do update set requested=1, requested_at=excluded.requested_at + """, + (run_id, now_iso()), + ) + con.execute("update runs set status=?, updated_at=? where id=? and status not in ('approved','blocked','failed','cancelled')", ("cancel_requested", now_iso(), run_id)) + + def cancellation_requested(self, run_id: str) -> bool: + with self.session() as con: + row = con.execute("select requested from cancellations where run_id=?", (run_id,)).fetchone() + return bool(row and row["requested"]) + + def complete_cancel(self, run_id: str) -> None: + with self.session() as con: + con.execute("update cancellations set completed_at=? where run_id=?", (now_iso(), run_id)) + con.execute("update runs set status=?, completed_at=?, updated_at=? where id=?", ("cancelled", now_iso(), now_iso(), run_id)) + def list_tasks(self) -> list[dict[str, Any]]: with self.session() as con: return [dict(row) for row in con.execute("select * from tasks order by updated_at desc")] diff --git a/clodex/tasks.py b/clodex/tasks.py new file mode 100644 index 0000000..9358cbc --- /dev/null +++ b/clodex/tasks.py @@ -0,0 +1,101 @@ +from __future__ import annotations + +import os +import signal +import subprocess +import sys +from pathlib import Path +from typing import Any + +from .artifacts import ArtifactStore, make_run_id, make_task_id +from .config import ClodexConfig, load_config +from .state import StateStore +from .workflow import WorkflowResult + + +class TaskManager: + def __init__(self, repo_root: Path | None = None): + self.config: ClodexConfig = load_config(repo_root) + self.repo_root = self.config.repo_root + self._state: StateStore | None = None + + @property + def state(self) -> StateStore: + if self._state is None: + self._state = StateStore(self.config.state_path) + return self._state + + def start( + self, + task: str, + workspace_backend: str | None = None, + approval_profile: str | None = None, + dry_run: bool = False, + ) -> WorkflowResult: + task_id = make_task_id(task) + run_id = make_run_id(task_id) + selected_workspace = workspace_backend or self.config.workspace["backend"] + selected_profile = approval_profile or self.config.codex["approval_profile"] + if dry_run: + artifacts_path = self.config.runs_root / run_id + return WorkflowResult( + "dry-run", + run_id, + task_id, + str(artifacts_path), + {"workspace": selected_workspace, "approval_profile": selected_profile}, + ) + + artifacts = ArtifactStore(self.config, run_id) + self.state.upsert_task(task_id, task, "queued") + self.state.create_run(run_id, task_id, task, "queued", artifacts_dir=str(artifacts.path)) + stdout = (artifacts.path / "worker.stdout.log").open("w", encoding="utf-8") + stderr = (artifacts.path / "worker.stderr.log").open("w", encoding="utf-8") + argv = [ + sys.executable, + "-m", + "clodex", + "task", + "worker", + run_id, + "--workspace", + selected_workspace, + "--approval-profile", + selected_profile, + ] + env = os.environ.copy() + root = str(Path(__file__).resolve().parents[1]) + env["PYTHONPATH"] = root + os.pathsep + env.get("PYTHONPATH", "") + process = subprocess.Popen(argv, cwd=self.repo_root, stdout=stdout, stderr=stderr, env=env) + stdout.close() + stderr.close() + self.state.update_run(run_id, "queued", pid=process.pid, artifacts_dir=str(artifacts.path)) + return WorkflowResult("queued", run_id, task_id, str(artifacts.path), {"pid": process.pid, "workspace": selected_workspace}) + + def get(self, run_id: str) -> dict[str, Any] | None: + run = self.state.get_run(run_id) + if run is None: + return None + return {"run": run} + + def list(self) -> dict[str, Any]: + return {"runs": self.state.list_runs(), "tasks": self.state.list_tasks()} + + def cancel(self, run_id: str) -> WorkflowResult: + run = self.state.get_run(run_id) + if run is None: + raise ValueError(f"Unknown run: {run_id}") + self.state.request_cancel(run_id) + pid = run.get("pid") + if pid: + self._terminate(int(pid)) + self.state.complete_cancel(run_id) + updated = self.state.get_run(run_id) or run + return WorkflowResult(str(updated["status"]), run_id, updated.get("task_id"), updated.get("artifacts_dir"), {"cancel_requested": True}) + + @staticmethod + def _terminate(pid: int) -> None: + try: + os.kill(pid, signal.SIGTERM) + except OSError: + pass diff --git a/clodex/trace.py b/clodex/trace.py new file mode 100644 index 0000000..97b1aa9 --- /dev/null +++ b/clodex/trace.py @@ -0,0 +1,28 @@ +from __future__ import annotations + +import json +from pathlib import Path +from typing import Any + +from .state import StateStore, now_iso + + +class TraceWriter: + def __init__(self, run_dir: Path, run_id: str, state: StateStore | None = None): + self.run_dir = run_dir + self.run_id = run_id + self.state = state + self.path = run_dir / "trace.jsonl" + self.run_dir.mkdir(parents=True, exist_ok=True) + + def event(self, event: str, data: dict[str, Any] | None = None) -> None: + payload = { + "ts": now_iso(), + "run_id": self.run_id, + "event": event, + "data": data or {}, + } + with self.path.open("a", encoding="utf-8", newline="\n") as handle: + handle.write(json.dumps(payload, sort_keys=True) + "\n") + if self.state is not None: + self.state.add_event(self.run_id, event, data or {}) diff --git a/clodex/workflow.py b/clodex/workflow.py index 50bee09..86f0a09 100644 --- a/clodex/workflow.py +++ b/clodex/workflow.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import subprocess from dataclasses import dataclass from pathlib import Path from typing import Any @@ -12,6 +13,8 @@ from .jsonutil import extract_json_object from .prompts import audit_prompt, fix_prompt, implementation_prompt, plan_prompt from .state import StateStore +from .trace import TraceWriter +from .workspace import DirtyWorkspaceError, WorkspaceManager, WorkspaceRef @dataclass @@ -28,7 +31,6 @@ def __init__(self, repo_root: Path | None = None): self.config: ClodexConfig = load_config(repo_root) self.repo_root = self.config.repo_root self._state: StateStore | None = None - self.runner = AgentRunner(self.repo_root) @property def state(self) -> StateStore: @@ -36,10 +38,11 @@ def state(self) -> StateStore: self._state = StateStore(self.config.state_path) return self._state - def dry_run_commands(self) -> dict[str, str]: + def dry_run_commands(self, workspace_root: Path | None = None, approval_profile: str | None = None) -> dict[str, str]: + root = workspace_root or self.repo_root return { "claude_plan": claude_plan_command(self.config).display(), - "codex_build": codex_exec_command(self.config, self.repo_root).display(), + "codex_build": codex_exec_command(self.config, root, approval_profile=approval_profile).display(), "claude_audit": claude_audit_command(self.config).display(), "codex_audit": codex_review_command(self.config).display(), } @@ -51,38 +54,60 @@ def plan(self, task: str, dry_run: bool = False) -> WorkflowResult: task_id = make_task_id(task) run_id = make_run_id(task_id) artifacts = ArtifactStore(self.config, run_id) + trace = TraceWriter(artifacts.path, run_id, self.state if self.config.tracing.get("enabled") else None) self.state.upsert_task(task_id, task, "planning") - self.state.create_run(run_id, task_id, task, "planning") + self.state.create_run(run_id, task_id, task, "planning", artifacts_dir=str(artifacts.path)) + trace.event("run.start", {"command": "plan", "task": task}) prompt = plan_prompt(self.config, task) - plan_json = self._run_json_with_retry(claude_plan_command(self.config), prompt, "Claude planning") + plan_json = self._run_json_with_retry(AgentRunner(self.repo_root), claude_plan_command(self.config), prompt, "Claude planning", trace) artifacts.write_json("01-claude-plan.json", plan_json) self.state.add_message(task_id, "planning", json.dumps(plan_json, indent=2)) self.state.update_run(run_id, "planned") self.state.update_task(task_id, "planned") + trace.event("run.complete", {"status": "planned"}) return WorkflowResult("planned", run_id, task_id, str(artifacts.path), plan_json) - def build(self, task: str, dry_run: bool = False) -> WorkflowResult: + def build( + self, + task: str, + dry_run: bool = False, + workspace_backend: str | None = None, + approval_profile: str | None = None, + apply_changes: bool = False, + ) -> WorkflowResult: if dry_run: - return WorkflowResult("dry-run", None, None, None, {"commands": self.dry_run_commands(), "task": task}) + return WorkflowResult( + "dry-run", + None, + None, + None, + { + "commands": self.dry_run_commands(approval_profile=approval_profile), + "task": task, + "workspace": workspace_backend or self.config.workspace["backend"], + "approval_profile": approval_profile or self.config.codex["approval_profile"], + }, + ) task_id = make_task_id(task) run_id = make_run_id(task_id) artifacts = ArtifactStore(self.config, run_id) self.state.upsert_task(task_id, task, "running") - self.state.create_run(run_id, task_id, task, "running") - - plan_json = self._run_json_with_retry(claude_plan_command(self.config), plan_prompt(self.config, task), "Claude planning") - artifacts.write_json("01-claude-plan.json", plan_json) - - implementation = self.runner.run(codex_exec_command(self.config, self.repo_root), implementation_prompt(plan_json, task)) - artifacts.write_text("02-codex-implementation.md", self._format_agent_report(implementation.stdout, implementation.stderr)) - if not implementation.ok: - self.state.update_run(run_id, "blocked") - self.state.update_task(task_id, "blocked") - return WorkflowResult("blocked", run_id, task_id, str(artifacts.path), {"error": "Codex implementation failed"}) + self.state.create_run(run_id, task_id, task, "running", artifacts_dir=str(artifacts.path)) + return self._execute_build(task, task_id, run_id, artifacts, workspace_backend, approval_profile, apply_changes) - result = self._audit_loop(task_id, run_id, artifacts, plan_json) - return result + def run_existing( + self, + run_id: str, + workspace_backend: str | None = None, + approval_profile: str | None = None, + apply_changes: bool = False, + ) -> WorkflowResult: + run = self.state.get_run(run_id) + if run is None: + raise ValueError(f"Unknown run: {run_id}") + artifacts = ArtifactStore(self.config, run_id) + return self._execute_build(str(run["prompt"]), str(run["task_id"]), run_id, artifacts, workspace_backend, approval_profile, apply_changes) def audit(self, dry_run: bool = False) -> WorkflowResult: if dry_run: @@ -91,62 +116,207 @@ def audit(self, dry_run: bool = False) -> WorkflowResult: task_id = make_task_id(task) run_id = make_run_id(task_id) artifacts = ArtifactStore(self.config, run_id) + trace = TraceWriter(artifacts.path, run_id, self.state if self.config.tracing.get("enabled") else None) self.state.upsert_task(task_id, task, "auditing") - self.state.create_run(run_id, task_id, task, "auditing") + self.state.create_run(run_id, task_id, task, "auditing", workspace_path=str(self.repo_root), artifacts_dir=str(artifacts.path)) plan_json = {"goal": task, "acceptance_criteria": ["Current diff is safe to ship"], "implementation_spec": []} artifacts.write_json("01-claude-plan.json", plan_json) - return self._audit_loop(task_id, run_id, artifacts, plan_json) + return self._audit_loop(task_id, run_id, artifacts, plan_json, AgentRunner(self.repo_root), self.repo_root, trace, None) - def _audit_loop(self, task_id: str, run_id: str, artifacts: ArtifactStore, plan_json: dict[str, Any]) -> WorkflowResult: + def apply_run(self, run_id: str | None, check: bool = False) -> WorkflowResult: + if not run_id: + raise ValueError("run_id is required") + run = self.state.get_run(run_id) + if run is None: + raise ValueError(f"Unknown run: {run_id}") + artifacts_dir = Path(str(run.get("artifacts_dir") or self.config.runs_root / run_id)) + patch = artifacts_dir / "apply.patch" + if not patch.exists(): + patch = artifacts_dir / "changes.diff" + if not patch.exists(): + raise ValueError(f"No patch artifact found for run: {run_id}") + argv = ["git", "apply", "--check" if check else "--whitespace=nowarn", str(patch)] + result = subprocess.run(argv, cwd=self.repo_root, capture_output=True, text=True, encoding="utf-8", errors="replace", check=False) + if result.returncode != 0: + return WorkflowResult("apply-check-failed" if check else "apply-failed", run_id, run.get("task_id"), str(artifacts_dir), {"stderr": result.stderr}) + if check: + return WorkflowResult("apply-check", run_id, run.get("task_id"), str(artifacts_dir), {"patch": str(patch)}) + self.state.update_run(run_id, "applied") + return WorkflowResult("applied", run_id, run.get("task_id"), str(artifacts_dir), {"patch": str(patch)}) + + def _execute_build( + self, + task: str, + task_id: str, + run_id: str, + artifacts: ArtifactStore, + workspace_backend: str | None, + approval_profile: str | None, + apply_changes: bool, + ) -> WorkflowResult: + trace = TraceWriter(artifacts.path, run_id, self.state if self.config.tracing.get("enabled") else None) + trace.event("run.start", {"command": "build", "task": task}) + try: + workspace = WorkspaceManager(self.repo_root, self.config).prepare(run_id, workspace_backend) + except DirtyWorkspaceError as exc: + self.state.update_run(run_id, "blocked", error=str(exc)) + self.state.update_task(task_id, "blocked") + trace.event("workspace.blocked", {"error": str(exc)}) + return WorkflowResult("blocked", run_id, task_id, str(artifacts.path), {"error": str(exc)}) + + WorkspaceManager(self.repo_root, self.config).write_metadata(artifacts.path, workspace) + self.state.add_workspace_lock(run_id, str(workspace.source_path), str(workspace.path), workspace.backend) + self.state.update_run(run_id, "running", workspace_path=str(workspace.path), artifacts_dir=str(artifacts.path)) + trace.event("workspace.ready", workspace.as_dict()) + + runner = AgentRunner(workspace.path) + if self.state.cancellation_requested(run_id): + self.state.complete_cancel(run_id) + trace.event("run.cancelled", {}) + return WorkflowResult("cancelled", run_id, task_id, str(artifacts.path), {"workspace": workspace.as_dict()}) + + plan_json = self._run_json_with_retry(runner, claude_plan_command(self.config), plan_prompt(self.config, task), "Claude planning", trace) + artifacts.write_json("01-claude-plan.json", plan_json) + if self._cancelled(run_id, trace): + return WorkflowResult("cancelled", run_id, task_id, str(artifacts.path), {"workspace": workspace.as_dict()}) + + implementation = runner.run(codex_exec_command(self.config, workspace.path, approval_profile=approval_profile), implementation_prompt(plan_json, task)) + trace.event("command.complete", {"name": implementation.command.name, "returncode": implementation.returncode}) + artifacts.write_text("02-codex-implementation.md", self._format_agent_report(implementation.stdout, implementation.stderr)) + if not implementation.ok: + self.state.update_run(run_id, "blocked", error="Codex implementation failed") + self.state.update_task(task_id, "blocked") + return WorkflowResult("blocked", run_id, task_id, str(artifacts.path), {"error": "Codex implementation failed", "workspace": workspace.as_dict()}) + if self._cancelled(run_id, trace): + return WorkflowResult("cancelled", run_id, task_id, str(artifacts.path), {"workspace": workspace.as_dict()}) + + result = self._audit_loop(task_id, run_id, artifacts, plan_json, runner, workspace.path, trace, approval_profile, include_untracked=workspace.is_worktree) + result.data["workspace"] = workspace.as_dict() + if result.status == "approved" and workspace.is_worktree: + self._include_untracked(workspace.path) + patch = artifacts.write_text("apply.patch", current_diff(workspace.path)) + result.data["apply_patch"] = str(patch) + if apply_changes: + return self.apply_run(run_id) + return result + + def _audit_loop( + self, + task_id: str, + run_id: str, + artifacts: ArtifactStore, + plan_json: dict[str, Any], + runner: AgentRunner, + diff_root: Path, + trace: TraceWriter, + approval_profile: str | None, + include_untracked: bool = False, + ) -> WorkflowResult: agreement: dict[str, Any] = {} for attempt in range(self.config.max_fix_loops + 1): - diff = current_diff(self.repo_root) + if include_untracked: + self._include_untracked(diff_root) + diff = current_diff(diff_root) diff_hash = hash_text(diff) artifacts.write_text("changes.diff", diff) - claude_audit = self._run_json_with_retry( - claude_audit_command(self.config), - audit_prompt("Claude Code", plan_json, diff, diff_hash), - "Claude audit", - ) - codex_audit = self._run_json_with_retry( - codex_review_command(self.config), - audit_prompt("Codex", plan_json, diff, diff_hash), - "Codex audit", - ) - artifacts.write_json(f"audit-attempt-{attempt}-claude.json", claude_audit) - artifacts.write_json(f"audit-attempt-{attempt}-codex.json", codex_audit) - artifacts.write_json("03-claude-audit.json", claude_audit) - artifacts.write_json("04-codex-audit.json", codex_audit) - self.state.add_audit(run_id, "claude", self._approved(claude_audit, diff_hash), claude_audit.get("diff_hash"), json.dumps(claude_audit)) - self.state.add_audit(run_id, "codex", self._approved(codex_audit, diff_hash), codex_audit.get("diff_hash"), json.dumps(codex_audit)) - agreement = { - "approved": self._approved(claude_audit, diff_hash) and self._approved(codex_audit, diff_hash), - "attempt": attempt, - "diff_hash": diff_hash, - "claude_approved": self._approved(claude_audit, diff_hash), - "codex_approved": self._approved(codex_audit, diff_hash), - } + verdicts = self._run_reviewers(artifacts, plan_json, diff, diff_hash, runner, trace, attempt) + agreement = self._agreement(verdicts, diff_hash, attempt) artifacts.write_json("05-agreement.json", agreement) self.state.update_run(run_id, "approved" if agreement["approved"] else "needs-fix", diff_hash) + trace.event("audit.agreement", agreement) if agreement["approved"]: self.state.update_task(task_id, "done") + trace.event("run.complete", {"status": "approved"}) return WorkflowResult("approved", run_id, task_id, str(artifacts.path), agreement) if attempt >= self.config.max_fix_loops: break - findings = self._required_fixes(claude_audit, codex_audit) - fix = self.runner.run(codex_exec_command(self.config, self.repo_root), fix_prompt(plan_json, findings)) + findings = self._required_fixes(*verdicts) + fix = runner.run(codex_exec_command(self.config, diff_root, approval_profile=approval_profile), fix_prompt(plan_json, findings)) artifacts.write_text(f"fix-attempt-{attempt}.md", self._format_agent_report(fix.stdout, fix.stderr)) - if not fix.ok: + trace.event("fix.complete", {"attempt": attempt, "returncode": fix.returncode}) + if not fix.ok or self._cancelled(run_id, trace): break self.state.update_run(run_id, "blocked", agreement.get("diff_hash")) self.state.update_task(task_id, "blocked") + trace.event("run.complete", {"status": "blocked"}) return WorkflowResult("blocked", run_id, task_id, str(artifacts.path), agreement) - def _run_json_with_retry(self, command, prompt: str, label: str) -> dict[str, Any]: + @staticmethod + def _include_untracked(repo_root: Path) -> None: + result = subprocess.run( + ["git", "ls-files", "--others", "--exclude-standard", "-z"], + cwd=repo_root, + capture_output=True, + check=False, + ) + if result.returncode != 0 or not result.stdout: + return + files = [item.decode("utf-8", errors="replace") for item in result.stdout.split(b"\0") if item] + if files: + subprocess.run(["git", "add", "-N", "--", *files], cwd=repo_root, check=False, capture_output=True) + + def _run_reviewers( + self, + artifacts: ArtifactStore, + plan_json: dict[str, Any], + diff: str, + diff_hash: str, + runner: AgentRunner, + trace: TraceWriter, + attempt: int, + ) -> list[dict[str, Any]]: + verdicts: list[dict[str, Any]] = [] + first_claude: dict[str, Any] | None = None + first_codex: dict[str, Any] | None = None + for reviewer in self.config.reviewers: + backend = str(reviewer.get("backend", "codex")) + reviewer_id = str(reviewer.get("id", backend)) + persona = str(reviewer.get("persona", reviewer_id)) + timeout = int(reviewer.get("timeout", 600)) + command = claude_audit_command(self.config) if backend == "claude" else codex_review_command(self.config) + verdict = self._run_json_with_retry( + runner, + command, + audit_prompt(backend.title(), plan_json, diff, diff_hash, reviewer_id, persona), + f"{reviewer_id} audit", + trace, + timeout=timeout, + ) + verdict.setdefault("reviewer_id", reviewer_id) + verdict.setdefault("persona", persona) + verdict["_required"] = bool(reviewer.get("required", True)) + artifacts.write_json(f"reviewers/{reviewer_id}.json", verdict) + artifacts.write_json(f"audit-attempt-{attempt}-{reviewer_id}.json", verdict) + self.state.add_audit(artifacts.run_id, reviewer_id, self._approved(verdict, diff_hash), verdict.get("diff_hash"), json.dumps(verdict)) + trace.event("audit.verdict", {"reviewer_id": reviewer_id, "approved": self._approved(verdict, diff_hash), "required": verdict["_required"]}) + verdicts.append(verdict) + if backend == "claude" and first_claude is None: + first_claude = verdict + if backend == "codex" and first_codex is None: + first_codex = verdict + if first_claude is not None: + artifacts.write_json("03-claude-audit.json", first_claude) + if first_codex is not None: + artifacts.write_json("04-codex-audit.json", first_codex) + return verdicts + + def _run_json_with_retry( + self, + runner: AgentRunner, + command, + prompt: str, + label: str, + trace: TraceWriter | None = None, + timeout: int | None = None, + ) -> dict[str, Any]: last_error = "" current_prompt = prompt - for attempt in range(2): - result = self.runner.run(command, current_prompt) + for _attempt in range(2): + if trace: + trace.event("command.start", {"name": command.name}) + result = runner.run(command, current_prompt, timeout=timeout) + if trace: + trace.event("command.complete", {"name": command.name, "returncode": result.returncode}) if not result.ok: raise RuntimeError(f"{label} failed with exit code {result.returncode}: {result.stderr.strip()}") try: @@ -160,14 +330,44 @@ def _run_json_with_retry(self, command, prompt: str, label: str) -> dict[str, An ) raise RuntimeError(f"{label} did not return parseable JSON: {last_error}") + def _cancelled(self, run_id: str, trace: TraceWriter) -> bool: + if self.state.cancellation_requested(run_id): + self.state.complete_cancel(run_id) + trace.event("run.cancelled", {}) + return True + return False + @staticmethod def _approved(audit: dict[str, Any], diff_hash: str) -> bool: return bool(audit.get("approved")) and audit.get("diff_hash") == diff_hash + @staticmethod + def _agreement(verdicts: list[dict[str, Any]], diff_hash: str, attempt: int) -> dict[str, Any]: + required = [verdict for verdict in verdicts if verdict.get("_required", True)] + reviewer_status = { + str(verdict.get("reviewer_id")): { + "approved": ClodexWorkflow._approved(verdict, diff_hash), + "required": bool(verdict.get("_required", True)), + "diff_hash": verdict.get("diff_hash"), + "persona": verdict.get("persona"), + } + for verdict in verdicts + } + return { + "approved": all(ClodexWorkflow._approved(verdict, diff_hash) for verdict in required), + "attempt": attempt, + "diff_hash": diff_hash, + "reviewers": reviewer_status, + "claude_approved": reviewer_status.get("claude-plan", {}).get("approved", False), + "codex_approved": reviewer_status.get("codex-architecture", {}).get("approved", False), + } + @staticmethod def _required_fixes(*audits: dict[str, Any]) -> list[str]: fixes: list[str] = [] for audit in audits: + if not audit.get("_required", True) and audit.get("approved"): + continue for fix in audit.get("required_fixes", []) or []: fixes.append(str(fix)) for finding in audit.get("findings", []) or []: diff --git a/clodex/workspace.py b/clodex/workspace.py new file mode 100644 index 0000000..fdc273a --- /dev/null +++ b/clodex/workspace.py @@ -0,0 +1,68 @@ +from __future__ import annotations + +import json +import subprocess +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from .config import ClodexConfig + + +class DirtyWorkspaceError(RuntimeError): + pass + + +@dataclass(frozen=True) +class WorkspaceRef: + backend: str + source_path: Path + path: Path + is_worktree: bool + + def as_dict(self) -> dict[str, Any]: + return { + "backend": self.backend, + "source_path": str(self.source_path), + "path": str(self.path), + "is_worktree": self.is_worktree, + } + + +class WorkspaceManager: + def __init__(self, repo_root: Path, config: ClodexConfig): + self.repo_root = repo_root.resolve() + self.config = config + + def prepare(self, run_id: str, backend: str | None = None) -> WorkspaceRef: + selected = backend or str(self.config.workspace.get("backend", "git-worktree")) + if selected == "local": + return WorkspaceRef("local", self.repo_root, self.repo_root, False) + if selected != "git-worktree": + raise ValueError(f"Unsupported workspace backend: {selected}") + self._ensure_clean_tracked() + path = (self.config.workspace_root / run_id).resolve() + path.parent.mkdir(parents=True, exist_ok=True) + if not path.exists(): + subprocess.run(["git", "worktree", "add", "--detach", str(path), "HEAD"], cwd=self.repo_root, check=True, capture_output=True, text=True) + return WorkspaceRef("git-worktree", self.repo_root, path, True) + + def write_metadata(self, run_dir: Path, workspace: WorkspaceRef) -> Path: + path = run_dir / "workspace.json" + path.write_text(json.dumps(workspace.as_dict(), indent=2, sort_keys=True) + "\n", encoding="utf-8") + return path + + def _ensure_clean_tracked(self) -> None: + result = subprocess.run( + ["git", "status", "--porcelain", "--untracked-files=no"], + cwd=self.repo_root, + capture_output=True, + text=True, + encoding="utf-8", + errors="replace", + check=False, + ) + if result.returncode != 0: + raise DirtyWorkspaceError(result.stderr.strip() or "Could not inspect git status") + if result.stdout.strip(): + raise DirtyWorkspaceError("Tracked changes are present; use --workspace local or commit/stash before git-worktree builds.") diff --git a/dist/.claude/commands/clodex/clodex-build.md b/dist/.claude/commands/clodex/clodex-build.md index 3cdfb34..3796435 100644 --- a/dist/.claude/commands/clodex/clodex-build.md +++ b/dist/.claude/commands/clodex/clodex-build.md @@ -7,3 +7,5 @@ clodex build "$ARGUMENTS" ``` Use this for the full Claude plan, Codex implementation, and dual audit loop. +By default, Clodex builds in `.clodex/workspaces//`; apply an approved +run with `clodex apply `. diff --git a/skills/clodex-workflow/SKILL.md b/skills/clodex-workflow/SKILL.md index 720a76b..55f66ca 100644 --- a/skills/clodex-workflow/SKILL.md +++ b/skills/clodex-workflow/SKILL.md @@ -8,14 +8,18 @@ description: Use when coordinating Claude Code CLI planning with Codex CLI imple Use `clodex` for local-first dual-agent work: 1. `clodex plan ""` asks Claude Code CLI to produce a structured plan. -2. `clodex build ""` runs Claude planning, Codex implementation, and dual audit. -3. `clodex audit --diff` audits the current uncommitted diff with both Claude and Codex. -4. `clodex status` shows recent local task and run state. +2. `clodex build ""` runs Claude planning, Codex implementation, and multi-reviewer audit in an isolated worktree. +3. `clodex apply ` applies an approved worktree patch back to the source checkout. +4. `clodex task start/get/cancel/list` manages durable async runs. +5. `clodex audit --diff` audits the current uncommitted diff with both Claude and Codex. +6. `clodex status` shows recent local task and run state. Defaults: - Claude planner/auditor: `claude -p --model opus --effort max --permission-mode plan`. - Codex engineer/auditor: `codex exec` or `codex review` with `gpt-5.5` and `model_reasoning_effort="xhigh"`. +- Builds default to `.clodex/workspaces//`; use `--workspace local` only when in-place changes are intentional. +- Approval profiles are `ci`, `local`, and `auto_review`; `ci` is the deterministic default. - Subscription CLI auth is preferred. API keys are fallback-only for CI/headless use. Do not mark a run complete unless `.clodex/runs//05-agreement.json` diff --git a/tests/test_clodex.py b/tests/test_clodex.py index e4a6535..f9ca733 100644 --- a/tests/test_clodex.py +++ b/tests/test_clodex.py @@ -7,13 +7,17 @@ import subprocess import sys import tempfile +import time import unittest from pathlib import Path from clodex.commands import claude_plan_command, codex_exec_command, codex_review_command from clodex.config import load_config from clodex.jsonutil import extract_json_object +from clodex.state import StateStore +from clodex.trace import TraceWriter from clodex.workflow import ClodexWorkflow +from clodex.workspace import DirtyWorkspaceError, WorkspaceManager ROOT = Path(__file__).resolve().parents[1] @@ -37,9 +41,10 @@ def __exit__(self, exc_type, exc, tb): class FakeCliPath: - def __init__(self, reject_once: bool = False, malformed_once: bool = False): + def __init__(self, reject_once: bool = False, malformed_once: bool = False, sleep_seconds: float = 0): self.reject_once = reject_once self.malformed_once = malformed_once + self.sleep_seconds = sleep_seconds def __enter__(self): self.tmp = tempfile.TemporaryDirectory() @@ -74,6 +79,10 @@ def _write_fake(self): args = sys.argv[2:] stdin = sys.stdin.read() +if {self.sleep_seconds!r}: + import time + time.sleep({self.sleep_seconds!r}) + if '--version' in args: print(name + ' fake 1.0.0') raise SystemExit(0) @@ -94,12 +103,20 @@ def requested_hash(): raise SystemExit(0) if 'adversarial auditor' in stdin: h = requested_hash() + reviewer = 'claude-plan' + persona = 'plan-adherence' + reviewer_match = re.search(r'Reviewer ID: ([^\\n]+)', stdin) + persona_match = re.search(r'Persona: ([^\\n]+)', stdin) + if reviewer_match: + reviewer = reviewer_match.group(1).strip() + if persona_match: + persona = persona_match.group(1).strip() reject_marker = Path('.fake-claude-reject') if {str(self.reject_once)!r} == 'True' and not reject_marker.exists(): reject_marker.write_text('seen') - print(json.dumps({{'approved': False, 'diff_hash': h, 'summary': 'reject once', 'findings': [], 'required_fixes': ['append fixed line']}})) + print(json.dumps({{'approved': False, 'diff_hash': h, 'reviewer_id': reviewer, 'persona': persona, 'summary': 'reject once', 'findings': [], 'required_fixes': ['append fixed line']}})) else: - print(json.dumps({{'approved': True, 'diff_hash': h, 'summary': 'ok', 'findings': [], 'required_fixes': []}})) + print(json.dumps({{'approved': True, 'diff_hash': h, 'reviewer_id': reviewer, 'persona': persona, 'summary': 'ok', 'findings': [], 'required_fixes': []}})) else: print(json.dumps({{'goal': 'test goal', 'scope': ['repo'], 'out_of_scope': [], 'implementation_spec': ['write implemented.txt'], 'acceptance_criteria': ['diff exists'], 'risks': [], 'test_commands': ['python -m unittest']}})) raise SystemExit(0) @@ -107,7 +124,15 @@ def requested_hash(): if name == 'codex': if args and args[0] == 'review': h = requested_hash() - print(json.dumps({{'approved': True, 'diff_hash': h, 'summary': 'ok', 'findings': [], 'required_fixes': []}})) + reviewer = 'codex-architecture' + persona = 'architecture' + reviewer_match = re.search(r'Reviewer ID: ([^\\n]+)', stdin) + persona_match = re.search(r'Persona: ([^\\n]+)', stdin) + if reviewer_match: + reviewer = reviewer_match.group(1).strip() + if persona_match: + persona = persona_match.group(1).strip() + print(json.dumps({{'approved': True, 'diff_hash': h, 'reviewer_id': reviewer, 'persona': persona, 'summary': 'ok', 'findings': [], 'required_fixes': []}})) raise SystemExit(0) if 'Required fixes' in stdin: Path('implemented.txt').write_text('implemented\\nfixed\\n', encoding='utf-8') @@ -142,10 +167,67 @@ def test_config_and_commands_use_requested_defaults(self): self.assertEqual(config.claude["effort"], "max") self.assertEqual(config.codex["model"], "gpt-5.5") self.assertEqual(config.codex["reasoning_effort"], "xhigh") + self.assertEqual(config.workspace["backend"], "git-worktree") + self.assertEqual(config.workspace["apply_mode"], "manual") + self.assertEqual(config.codex["approval_profile"], "ci") + self.assertTrue(config.mcp["async_tasks"]) + self.assertTrue(config.tracing["enabled"]) + self.assertGreaterEqual(len(config.reviewers), 2) self.assertIn("--permission-mode", claude_plan_command(config).argv) self.assertIn("model_reasoning_effort=\"xhigh\"", codex_exec_command(config, repo).argv) self.assertIn("--uncommitted", codex_review_command(config).argv) + def test_approval_profiles_change_codex_command(self): + with TempRepo() as repo: + config = load_config(repo) + ci = codex_exec_command(config, repo).argv + local = codex_exec_command(config, repo, approval_profile="local").argv + auto = codex_exec_command(config, repo, approval_profile="auto_review").argv + self.assertIn("never", ci) + self.assertIn("on-request", local) + self.assertIn("approvals_reviewer=\"auto_review\"", auto) + + def test_state_migrations_add_v2_tables(self): + with tempfile.TemporaryDirectory() as tmp: + db = Path(tmp) / "state.sqlite3" + store = StateStore(db) + tables = store.table_names() + self.assertIn("schema_version", tables) + self.assertIn("run_events", tables) + self.assertIn("artifacts", tables) + self.assertIn("workspace_locks", tables) + self.assertIn("cancellations", tables) + self.assertEqual(store.schema_version(), 2) + + def test_trace_writer_appends_jsonl_events(self): + with tempfile.TemporaryDirectory() as tmp: + trace = TraceWriter(Path(tmp), "run-1") + trace.event("phase.start", {"phase": "planning"}) + lines = (Path(tmp) / "trace.jsonl").read_text(encoding="utf-8").splitlines() + self.assertEqual(len(lines), 1) + event = json.loads(lines[0]) + self.assertEqual(event["run_id"], "run-1") + self.assertEqual(event["event"], "phase.start") + self.assertEqual(event["data"]["phase"], "planning") + + def test_workspace_manager_refuses_dirty_source(self): + with TempRepo() as repo: + (repo / "seed.txt").write_text("dirty\n", encoding="utf-8") + manager = WorkspaceManager(repo, load_config(repo)) + with self.assertRaises(DirtyWorkspaceError): + manager.prepare("run-dirty", backend="git-worktree") + + def test_worktree_build_isolated_until_apply(self): + with TempRepo() as repo, FakeCliPath(): + result = ClodexWorkflow(repo).build("implement fixture") + self.assertEqual(result.status, "approved") + self.assertFalse((repo / "implemented.txt").exists()) + apply_result = ClodexWorkflow(repo).apply_run(result.run_id) + self.assertEqual(apply_result.status, "applied") + self.assertTrue((repo / "implemented.txt").exists()) + workspace = Path(result.data["workspace"]["path"]) + self.assertTrue(workspace.exists()) + def test_json_extraction_handles_fenced_json(self): value = extract_json_object("```json\n{\"approved\": true}\n```") self.assertTrue(value["approved"]) @@ -168,18 +250,32 @@ def test_doctor_reports_fake_clis(self): def test_build_happy_path_creates_agreement(self): with TempRepo() as repo, FakeCliPath(): - result = ClodexWorkflow(repo).build("implement fixture") + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") self.assertEqual(result.status, "approved") agreement = json.loads((Path(result.artifacts_dir) / "05-agreement.json").read_text(encoding="utf-8")) self.assertTrue(agreement["approved"]) self.assertTrue((repo / "implemented.txt").exists()) + self.assertTrue((Path(result.artifacts_dir) / "trace.jsonl").exists()) + self.assertTrue((Path(result.artifacts_dir) / "reviewers" / "claude-plan.json").exists()) def test_rejection_triggers_fix_loop(self): with TempRepo() as repo, FakeCliPath(reject_once=True): - result = ClodexWorkflow(repo).build("implement fixture") + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") self.assertEqual(result.status, "approved") self.assertIn("fixed", (repo / "implemented.txt").read_text(encoding="utf-8")) + def test_required_reviewer_rejection_blocks(self): + with TempRepo() as repo: + config = repo / "CLODEX.md" + config.write_text( + config.read_text(encoding="utf-8").replace("max_fix_loops: 2", "max_fix_loops: 0"), + encoding="utf-8", + ) + with FakeCliPath(reject_once=True): + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "blocked") + self.assertFalse(result.data["approved"]) + def test_malformed_plan_retries_once(self): with TempRepo() as repo, FakeCliPath(malformed_once=True): result = ClodexWorkflow(repo).plan("plan fixture") @@ -202,6 +298,115 @@ def test_mcp_tools_list(self): names = {tool["name"] for tool in response["result"]["tools"]} self.assertIn("clodex_build", names) self.assertIn("clodex_task_update", names) + self.assertIn("clodex_task_start", names) + self.assertIn("clodex_task_get", names) + self.assertIn("clodex_task_cancel", names) + + def test_mcp_tasks_get_unknown_run(self): + request = {"jsonrpc": "2.0", "id": 1, "method": "tasks/get", "params": {"id": "missing"}} + result = subprocess.run( + [sys.executable, "-m", "clodex", "mcp-server"], + input=json.dumps(request) + "\n", + cwd=ROOT, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + self.assertEqual(result.returncode, 0, result.stderr) + response = json.loads(result.stdout.splitlines()[0]) + self.assertEqual(response["error"]["code"], -32004) + + def test_task_start_get_cancel_lifecycle(self): + with TempRepo() as repo, FakeCliPath(sleep_seconds=1): + start = subprocess.run( + [sys.executable, "-m", "clodex", "--json", "task", "start", "--workspace", "local", "slow fixture"], + cwd=repo, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + self.assertEqual(start.returncode, 0, start.stdout + start.stderr) + run_id = json.loads(start.stdout)["run_id"] + cancel = subprocess.run( + [sys.executable, "-m", "clodex", "--json", "task", "cancel", run_id], + cwd=repo, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + self.assertEqual(cancel.returncode, 0, cancel.stdout + cancel.stderr) + for _ in range(20): + status = subprocess.run( + [sys.executable, "-m", "clodex", "--json", "task", "get", run_id], + cwd=repo, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + data = json.loads(status.stdout) + if data["run"]["status"] in {"cancelled", "approved", "blocked"}: + break + time.sleep(0.1) + self.assertIn(data["run"]["status"], {"cancel_requested", "cancelled"}) + + def test_hooks_print_and_ingest(self): + with TempRepo() as repo: + printed = subprocess.run( + [sys.executable, "-m", "clodex", "--json", "hooks", "print"], + cwd=repo, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + self.assertEqual(printed.returncode, 0, printed.stderr) + config = json.loads(printed.stdout) + self.assertIn("hooks", config) + ingested = subprocess.run( + [sys.executable, "-m", "clodex", "--json", "hooks", "ingest", "--run-id", "run-hooks"], + cwd=repo, + input=json.dumps({"event": "SessionStart"}), + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + self.assertEqual(ingested.returncode, 0, ingested.stdout + ingested.stderr) + + def test_trace_export_and_eval_run(self): + with TempRepo() as repo, FakeCliPath(): + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") + exported = subprocess.run( + [sys.executable, "-m", "clodex", "--json", "trace", "export", result.run_id, "--format", "jsonl"], + cwd=repo, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + self.assertEqual(exported.returncode, 0, exported.stderr) + self.assertIn("run.start", exported.stdout) + evaluated = subprocess.run( + [sys.executable, "-m", "clodex", "--json", "eval", "run"], + cwd=repo, + env={**os.environ, "PYTHONPATH": str(ROOT)}, + capture_output=True, + text=True, + encoding="utf-8", + check=False, + ) + self.assertEqual(evaluated.returncode, 0, evaluated.stdout + evaluated.stderr) if __name__ == "__main__":