From 8c0aa2e998f126015b109e0b768d88b50df77115 Mon Sep 17 00:00:00 2001 From: Devil Date: Thu, 1 Oct 2026 00:25:23 -0400 Subject: [PATCH] feat: clodex_delegate - hand a native handoff's work to Codex Native handoffs only recorded state; nothing started Codex. New MCP tool `clodex_delegate` (run_id, mode, instructions, approval_profile, wait) starts a worker that runs Codex in the handoff's workspace and records the outcome on the handoff itself, since a non-interactive `codex exec` has no MCP access to report for itself: - implement: carries out the instructions; fix: applies the open findings of reviewers that have not approved the latest diff; audit: Codex reviews the diff (schema-constrained) and its verdict is recorded for the first required Codex reviewer, on the diff it actually saw. - Returns at once with a delegation to poll via clodex_handoff_get (which now lists `delegations`); `wait: true` blocks on a worker thread without stalling pings/cancels. One delegation per handoff, enforced atomically (new `delegations` table, schema v3, additive). - Success counts as one handoff against the budget; a failed job does not and is reported on the handoff. A worker that dies (or never started) is detected and its delegation failed. Cancelling the handoff stops the worker tree and releases the worktree; a pid without a recent heartbeat is never signalled. - A worktree is created on first use if the handoff has none; a retired Codex model is refused up front; the diff hash and artifacts are recorded like every other handoff update. - Native instruction templates describe the full loop (create with a worktree -> delegate implement -> verdicts -> delegate audit/fix -> decide -> apply). Worker liveness rules are shared with task workers (procs.worker_state). Co-Authored-By: Claude Sonnet 5.5 --- README.md | 30 +++ clodex/cli.py | 11 ++ clodex/delegate.py | 295 +++++++++++++++++++++++++++++ clodex/mcp_server.py | 70 ++++++- clodex/native.py | 16 +- clodex/procs.py | 31 ++++ clodex/prompts.py | 25 +++ clodex/state.py | 88 ++++++++- clodex/tasks.py | 43 +---- clodex/workflow.py | 4 + tests/support.py | 5 + tests/test_delegate.py | 409 +++++++++++++++++++++++++++++++++++++++++ tests/test_state.py | 2 +- 13 files changed, 981 insertions(+), 48 deletions(-) create mode 100644 clodex/delegate.py create mode 100644 tests/test_delegate.py diff --git a/README.md b/README.md index b9901b6..960f22c 100644 --- a/README.md +++ b/README.md @@ -82,6 +82,7 @@ Native coordination uses these MCP tools: - `clodex_handoff_update` - `clodex_handoff_get` - `clodex_handoff_decide` +- `clodex_delegate` If MCP is unavailable, use the CLI fallbacks: @@ -309,6 +310,35 @@ The server speaks MCP `2025-11-25` (and negotiates down to `2025-06-18`, `2025-0 ### Native handoffs +The loop an orchestrating agent follows: + +```text +clodex_handoff_create (workspace: git-worktree) + -> clodex_handoff_update record the plan + -> clodex_delegate implement Codex implements in the isolated worktree + -> clodex_handoff_update Claude's own verdict on the diff + -> clodex_delegate audit Codex's verdict (schema-constrained), recorded for you + -> clodex_delegate fix + audit again if anyone rejected + -> clodex_handoff_decide -> clodex apply +``` + +**`clodex_delegate`** (`run_id`, `mode`, `instructions`, `approval_profile`, `wait`) starts a +worker that runs Codex in the handoff's workspace and records the outcome on the handoff, because a +non-interactive `codex exec` has no MCP access to report for itself: + +- `implement` carries out `instructions` (required); `fix` applies the open findings of reviewers + that have not approved the latest diff, plus any `instructions`; `audit` has Codex review the + diff and records its verdict for the first required Codex reviewer, on the diff it actually saw. +- It returns immediately with a delegation to poll through `clodex_handoff_get` (which now shows + `delegations`); `wait: true` blocks until it finishes without stalling other requests. +- Only one delegation runs per handoff. A successful implement/fix/audit counts as one handoff + against the budget; a failed job does not, and is reported on the handoff so the orchestrator + can decide. A worker that dies is detected and its delegation marked failed. +- Cancelling the handoff (`clodex_task_cancel` / `clodex task cancel`) stops the worker and + everything it started, and removes the worktree. +- If the handoff has no workspace, an isolated worktree is created on first use. A retired Codex + model is refused up front. + - **Decisions follow your audit settings.** `clodex_handoff_decide` evaluates the reports recorded with `clodex_handoff_update` against the same `audit.reviewers` and `audit.quorum` as a classic run (`unanimous`, `majority` or a number; only `required` reviewers count). A report is a verdict diff --git a/clodex/cli.py b/clodex/cli.py index e76f5d0..82de9f2 100644 --- a/clodex/cli.py +++ b/clodex/cli.py @@ -8,6 +8,7 @@ from typing import Any from .config import ConfigError, resolve_repo_root +from .delegate import run_delegation from .doctor import run_doctor from .evals import run_local_evals from .hooks import derive_run_id, hook_config, ingest_hook_event, install_hooks, parse_hook_payload @@ -112,6 +113,10 @@ def _main(argv: list[str] | None = None) -> int: 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"]) + task_delegate = task_sub.add_parser("delegate-worker", help="Internal: run one delegated Codex job for a native handoff") + task_delegate.add_argument("run_id") + task_delegate.add_argument("delegation_id", type=int) + task_delegate.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) @@ -297,6 +302,12 @@ def handle_task(args: argparse.Namespace, workflow: ClodexWorkflow, as_json: boo if args.task_command == "list": print_output(manager.list(), as_json) return 0 + if args.task_command == "delegate-worker": + delegation_id = args.delegation_id + with Heartbeat(workflow.state, delegation_id, beat=lambda: workflow.state.touch_delegation(delegation_id)): + delegation = run_delegation(resolve_repo_root(), args.run_id, delegation_id) + print_output(delegation, as_json) + return 0 if delegation.get("status") == "completed" else 1 if args.task_command == "worker": with Heartbeat(workflow.state, args.run_id): result = workflow.run_existing(args.run_id, workspace_backend=args.workspace, approval_profile=args.approval_profile) diff --git a/clodex/delegate.py b/clodex/delegate.py new file mode 100644 index 0000000..0ba7549 --- /dev/null +++ b/clodex/delegate.py @@ -0,0 +1,295 @@ +"""Delegation: a native handoff asks Codex to implement, fix or audit, and a worker does the job. + +Claude orchestrates over MCP; Codex runs as a non-interactive subprocess without MCP access, so +the worker is the one that records Codex's result on the handoff (and, for audits, Codex's +verdict) using the same state as every other handoff update. +""" + +from __future__ import annotations + +import os +import subprocess +import sys +import time +import traceback +from collections.abc import Callable +from pathlib import Path +from typing import Any + +from .agents import AgentRunner +from .artifacts import ArtifactStore, current_diff, hash_text +from .commands import codex_exec_command, codex_review_command +from .config import ClodexConfig, load_config +from .models import ensure_usable +from .procs import START_GRACE, kill_tree, parse_time, popen_isolation_kwargs, worker_state +from .prompts import audit_diff_excerpt, audit_prompt, delegate_prompt +from .quorum import required_fixes, resolve_reviewer +from .state import DELEGATION_ACTIVE, DELEGATION_FINISHED, DELEGATION_MODES, StateStore, now_iso +from .workspace import WorkspaceManager + +POLL_SECONDS = 0.5 +SUMMARY_LIMIT = 4000 + + +class WaitAborted(Exception): + """The caller stopped waiting (cancelled request or closed connection).""" + + +def delegation_worker_state(delegation: dict[str, Any]) -> str: + return worker_state(delegation.get("pid"), delegation.get("heartbeat_at"), delegation.get("started_at") or delegation.get("created_at")) + + +def reconcile_delegation(state: StateStore, delegation: dict[str, Any]) -> dict[str, Any]: + """Notice a worker that died (or never started) without finishing, instead of leaving it running forever.""" + if str(delegation["status"]) not in DELEGATION_ACTIVE: + return delegation + health = delegation_worker_state(delegation) + if not delegation.get("pid"): + # Just created and not yet spawned is normal; a delegation that never got a worker is not. + parsed = parse_time(delegation.get("created_at")) + if parsed is not None and (time.time() - parsed.timestamp()) > START_GRACE: + return state.update_delegation(delegation["id"], status="failed", error="the worker was never started") + return delegation + if health == "dead": + return state.update_delegation(delegation["id"], status="failed", error="worker process exited before finishing the delegation") + return {**delegation, "worker_state": health} + + +def cancel_active_delegation(state: StateStore, run_id: str) -> bool: + """Stop the handoff's running delegation, if any. Returns True if a worker was stopped. + + The process is only signalled while its heartbeat is fresh, so a reused pid is never killed. + """ + delegation = state.active_delegation(run_id) + if delegation is None: + return False + stopped = False + if delegation_worker_state(delegation) == "alive": + kill_tree(int(delegation["pid"])) + stopped = True + state.update_delegation(delegation["id"], status="cancelled", error="cancelled with the handoff") + return stopped + + +def open_fixes(data: dict[str, Any], reviewers: list[dict[str, Any]]) -> list[str]: + """What reviewers have asked to be fixed on the latest diff and not yet approved.""" + latest_hash: str | None = None + verdicts: dict[str, dict[str, Any]] = {} + for event in data.get("events") or []: + if event.get("event") != "handoff.update" or not isinstance(event.get("data"), dict): + continue + event_data = event["data"] + report = event_data.get("report") if isinstance(event_data.get("report"), dict) else {} + diff_hash = (event_data.get("diff_hash") or report.get("diff_hash") or None) + if diff_hash and diff_hash != latest_hash: + latest_hash, verdicts = str(diff_hash), {} + reviewer = resolve_reviewer(report.get("reviewer_id") or event_data.get("actor"), reviewers) + if reviewer is None or not isinstance(report.get("approved"), bool): + continue + verdicts[str(reviewer["id"])] = {**report, "reviewer_id": str(reviewer["id"]), "approved": report["approved"], "diff_hash": latest_hash, "_required": bool(reviewer.get("required", True))} + rejected = [verdict for verdict in verdicts.values() if not verdict["approved"]] + return required_fixes(rejected, latest_hash) if latest_hash and rejected else [] + + +class DelegationManager: + 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, run_id: str, instructions: str | None = None, mode: str = "implement", approval_profile: str | None = None) -> dict[str, Any]: + if mode not in DELEGATION_MODES: + raise ValueError(f"unknown delegation mode: {mode} (use one of {sorted(DELEGATION_MODES)})") + data = self.state.get_handoff(run_id) + if data is None: + raise ValueError(f"Unknown run: {run_id}") + run = data["run"] + if run["status"] != "handoff": + raise ValueError(f"the handoff is {run['status']}; nothing can be delegated") + if data["budget_remaining"] < 1: + raise ValueError("handoff budget exhausted: nothing can be delegated") + if mode == "implement" and not (instructions and instructions.strip()): + raise ValueError("instructions are required to delegate an implementation") + ensure_usable(self.config) + active = self.state.active_delegation(run_id) + if active is not None: + reconcile_delegation(self.state, active) # a dead worker must not block the handoff forever + self._ensure_workspace(run) + + delegation = self.state.start_delegation(run_id, mode, instructions, approval_profile or str(self.config.codex.get("approval_profile", "ci"))) + try: + process = self._spawn(run_id, int(delegation["id"]), str(delegation["approval_profile"])) + except Exception as exc: # noqa: BLE001 + self.state.update_delegation(delegation["id"], status="failed", error=f"could not start the worker: {exc}") + raise + delegation = self.state.update_delegation(delegation["id"], pid=process.pid) + self.state.add_event(run_id, "handoff.delegate", {"delegation": delegation["id"], "mode": mode}) + return delegation + + def _ensure_workspace(self, run: dict[str, Any]) -> None: + if run.get("workspace_path"): + return + # The handoff was created without a workspace: give Codex an isolated worktree now. + workspace = WorkspaceManager(self.repo_root, self.config).prepare(str(run["id"]), None) + self.state.update_run(str(run["id"]), "handoff", workspace_path=str(workspace.path)) + self.state.add_workspace_lock(str(run["id"]), str(workspace.source_path), str(workspace.path), workspace.backend) + + def _spawn(self, run_id: str, delegation_id: int, approval_profile: str) -> subprocess.Popen: + artifacts = ArtifactStore(self.config, run_id, self.state) + out = (artifacts.path / f"delegation-{delegation_id}.stdout.log").open("w", encoding="utf-8") + err = (artifacts.path / f"delegation-{delegation_id}.stderr.log").open("w", encoding="utf-8") + env = os.environ.copy() + root = str(Path(__file__).resolve().parents[1]) + env["PYTHONPATH"] = root + os.pathsep + env.get("PYTHONPATH", "") + argv = [sys.executable, "-m", "clodex", "task", "delegate-worker", run_id, str(delegation_id), "--approval-profile", approval_profile] + try: + return subprocess.Popen(argv, cwd=self.repo_root, stdin=subprocess.DEVNULL, stdout=out, stderr=err, env=env, **popen_isolation_kwargs()) + finally: + out.close() + err.close() + + def get(self, delegation_id: int) -> dict[str, Any] | None: + delegation = self.state.get_delegation(delegation_id) + return reconcile_delegation(self.state, delegation) if delegation else None + + def reconcile_active(self, run_id: str) -> None: + active = self.state.active_delegation(run_id) + if active is not None: + reconcile_delegation(self.state, active) + + def wait(self, delegation_id: int, should_stop: Callable[[], bool] | None = None, timeout: float | None = None) -> dict[str, Any]: + deadline = None if timeout is None else time.monotonic() + timeout + while True: + delegation = self.get(delegation_id) + if delegation is None: + raise ValueError(f"unknown delegation: {delegation_id}") + if str(delegation["status"]) in DELEGATION_FINISHED: + return delegation + if should_stop is not None and should_stop(): + raise WaitAborted + if deadline is not None and time.monotonic() >= deadline: + return delegation + time.sleep(POLL_SECONDS) + + +# ---------------------------------------------------------------- the worker + + +def run_delegation(repo_root: Path | None, run_id: str, delegation_id: int) -> dict[str, Any]: + """Executed by `clodex task delegate-worker`. Always leaves the delegation in a finished state.""" + config = load_config(repo_root) + state = StateStore(config.state_path) + delegation = state.get_delegation(delegation_id) + if delegation is None or delegation["run_id"] != run_id: + raise ValueError(f"unknown delegation {delegation_id} for {run_id}") + state.update_delegation(delegation_id, status="running", started_at=now_iso()) + try: + outcome = _execute(config, state, run_id, delegation) + except Exception as exc: # noqa: BLE001 - record any failure on the delegation + error = f"{type(exc).__name__}: {exc}" + state.update_delegation(delegation_id, status="failed", error=error) + _record_failure(state, run_id, delegation, error) + (ArtifactStore(config, run_id, state).path / f"delegation-{delegation_id}.error.txt").write_text(traceback.format_exc(), encoding="utf-8") + return state.get_delegation(delegation_id) or {} + return state.update_delegation(delegation_id, status="completed", summary=outcome["summary"][:SUMMARY_LIMIT], diff_hash=outcome.get("diff_hash")) + + +def _record_failure(state: StateStore, run_id: str, delegation: dict[str, Any], error: str) -> None: + """Tell the orchestrator (via the handoff) that the job failed, without spending handoff budget.""" + try: + state.update_handoff(run_id, actor="codex", report={"delegation": delegation["id"], "mode": delegation["mode"], "error": error}) + except ValueError: + pass # the handoff finished or was cancelled meanwhile + + +def _prepare_diff(workspace: Path, repo_root: Path) -> tuple[str, str]: + if workspace.resolve() != repo_root.resolve(): # an isolated worktree: show new files in the diff too + names = subprocess.run(["git", "ls-files", "--others", "--exclude-standard", "-z"], cwd=workspace, capture_output=True, check=False).stdout + files = [item.decode("utf-8", errors="replace") for item in names.split(b"\0") if item] + if files: + subprocess.run(["git", "add", "-N", "--", *files], cwd=workspace, capture_output=True, check=False) + diff = current_diff(workspace) + return diff, hash_text(diff) + + +def _execute(config: ClodexConfig, state: StateStore, run_id: str, delegation: dict[str, Any]) -> dict[str, Any]: + data = state.get_handoff(run_id) + if data is None: + raise ValueError(f"Unknown run: {run_id}") + run = data["run"] + workspace = Path(str(run["workspace_path"])) + if not workspace.is_dir(): + raise ValueError(f"the handoff workspace is gone: {workspace}") + runner = AgentRunner(workspace) + artifacts = ArtifactStore(config, run_id, state) + mode = str(delegation["mode"]) + instructions = (delegation.get("instructions") or "").strip() or None + task = str(run["prompt"]) + + if mode == "audit": + return _audit(config, state, run_id, delegation, run, runner, artifacts, workspace, instructions, task) + + fixes = open_fixes(data, config.reviewers) if mode == "fix" else None + prompt = delegate_prompt(mode, task, instructions, fixes) + result = runner.run(codex_exec_command(config, workspace, approval_profile=delegation.get("approval_profile")), prompt) + report_path = artifacts.write_text(f"delegation-{delegation['id']}-codex.md", _report_text(result.stdout, result.stderr)) + if not result.ok: + raise RuntimeError(f"codex exited with code {result.returncode}: {(result.stderr or result.stdout).strip()[:500]}") + diff, diff_hash = _prepare_diff(workspace, config.repo_root) + diff_path = artifacts.write_text("changes.diff", diff, exact=True) + summary = result.stdout.strip() or "(codex printed no report)" + state.update_handoff( + run_id, + phase="implementation" if mode == "implement" else "fix", + actor="codex", + increment_handoff=True, + diff_hash=diff_hash, + report={ + "summary": summary[:SUMMARY_LIMIT], + "changed": bool(diff.strip()), + "delegation": delegation["id"], + "mode": mode, + "artifacts": [str(report_path), str(diff_path)], + }, + ) + return {"summary": summary, "diff_hash": diff_hash} + + +def _audit(config, state, run_id, delegation, run, runner, artifacts, workspace, instructions, task) -> dict[str, Any]: + from .workflow import ClodexWorkflow # deferred: workflow imports a lot + + diff, diff_hash = _prepare_diff(workspace, config.repo_root) + if not diff.strip(): + raise ValueError("there is nothing to audit: the workspace has no changes") + reviewer = resolve_reviewer("codex", config.reviewers) + reviewer_id = str(reviewer["id"]) if reviewer else "codex-audit" + persona = str(reviewer.get("persona", reviewer_id)) if reviewer else "audit" + timeout = int(reviewer.get("timeout", 600)) if reviewer else 600 + plan = {"goal": task, "implementation_spec": [instructions] if instructions else [], "acceptance_criteria": ["The diff is correct, scoped and safe to ship"]} + shown = audit_diff_excerpt(diff, int(config.audit.get("max_diff_bytes", 200_000))) + verdict = ClodexWorkflow(config.repo_root).run_agent_json( + runner, + codex_review_command(config, workspace), + audit_prompt("Codex", plan, shown, diff_hash, reviewer_id, persona), + f"{reviewer_id} audit", + timeout=timeout, + ) + verdict_path = artifacts.write_json(f"delegation-{delegation['id']}-audit.json", verdict) + diff_path = artifacts.write_text("changes.diff", diff, exact=True) + verdict.update({"reviewer_id": reviewer_id, "delegation": delegation["id"], "mode": "audit", "artifacts": [str(verdict_path), str(diff_path)]}) + # The verdict counts for the diff we actually audited, whatever hash the model echoed back. + state.update_handoff(run_id, phase="audit", actor="codex", increment_handoff=True, diff_hash=diff_hash, report=verdict) + return {"summary": str(verdict.get("summary") or ("approved" if verdict.get("approved") else "rejected")), "diff_hash": diff_hash} + + +def _report_text(stdout: str, stderr: str) -> str: + report = stdout.strip() + if stderr.strip(): + report += ("\n\n" if report else "") + "stderr:\n" + stderr.strip() + return report diff --git a/clodex/mcp_server.py b/clodex/mcp_server.py index 1133dc7..473ba1c 100644 --- a/clodex/mcp_server.py +++ b/clodex/mcp_server.py @@ -7,12 +7,14 @@ import sys import threading import uuid +from collections.abc import Callable from concurrent.futures import ThreadPoolExecutor from pathlib import Path from typing import Any from . import __version__ from .config import resolve_repo_root +from .delegate import DelegationManager, WaitAborted from .quorum import evaluate_handoff, resolve_reviewer from .schemas import validate as validate_schema from .tasks import TaskManager @@ -143,6 +145,27 @@ "description": "Read native handoff state, events, artifacts, budget, and next actor.", "inputSchema": {"type": "object", "properties": {"run_id": {"type": "string"}}, "required": ["run_id"]}, }, + { + "name": "clodex_delegate", + "title": "Delegate to Codex", + "description": ( + "Hand a native handoff's work to Codex, which runs in the handoff's isolated worktree and records its result on the handoff. " + "mode `implement` (needs `instructions`: the plan to carry out), `fix` (applies the open review findings plus any `instructions`) " + "or `audit` (Codex reviews the diff and records its verdict). Returns at once with a delegation to poll via clodex_handoff_get; " + "pass `wait: true` to block until it finishes." + ), + "inputSchema": { + "type": "object", + "properties": { + "run_id": {"type": "string"}, + "mode": {"type": "string", "enum": ["implement", "fix", "audit"]}, + "instructions": {"type": "string"}, + "approval_profile": {"type": "string", "enum": ["ci", "local", "auto_review"]}, + "wait": {"type": "boolean"}, + }, + "required": ["run_id"], + }, + }, { "name": "clodex_handoff_decide", "title": "Decide Clodex handoff", @@ -321,6 +344,9 @@ def dispatch_line(self, line: str, pool: ThreadPoolExecutor) -> None: def is_blocking(self, method: str, params: Any) -> bool: if method == "tasks/result": return True # waits for the task to finish + if method == "tools/call" and isinstance(params, dict) and params.get("name") == "clodex_delegate": + arguments = params.get("arguments") + return isinstance(arguments, dict) and arguments.get("wait") is True # only waiting blocks if method != "tools/call" or not isinstance(params, dict) or params.get("name") not in self.LONG_TOOLS: return False arguments = params.get("arguments") @@ -373,7 +399,7 @@ def handle(self, method: str, params: Any, cancelled: threading.Event) -> dict[s if method == "tools/list": return {"tools": TOOLS} if method == "tools/call": - return self.tools_call(params) + return self.tools_call(params, lambda: cancelled.is_set() or self._closing.is_set()) if self.tasks_enabled: if method == "tasks/get": return task_object(self.task_run(params)) @@ -402,7 +428,7 @@ def initialize(self, params: dict[str, Any]) -> dict[str, Any]: # ------------------------------------------------------------ tools - def tools_call(self, params: dict[str, Any]) -> dict[str, Any]: + def tools_call(self, params: dict[str, Any], should_stop: Callable[[], bool] | None = None) -> dict[str, Any]: name = params.get("name") if not isinstance(name, str): raise RpcError(INVALID_PARAMS, "tools/call requires a tool name") @@ -421,7 +447,10 @@ def tools_call(self, params: dict[str, Any]) -> dict[str, Any]: if name not in TASK_TOOLS: raise RpcError(METHOD_NOT_FOUND, f"Tool does not support task augmentation: {name}") return self.create_task(name, arguments) - return tool_call(name, arguments) + try: + return tool_call(name, arguments, should_stop) + except WaitAborted: + raise _Aborted from None def create_task(self, name: str, arguments: dict[str, Any]) -> dict[str, Any]: if arguments.get("dry_run"): @@ -494,7 +523,7 @@ def main() -> int: return McpServer().serve(sys.stdin) -def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: +def tool_call(name: str, arguments: dict[str, Any], should_stop: Callable[[], bool] | None = None) -> dict[str, Any]: root = resolve_repo_root() workflow = None if name in TASK_MANAGER_TOOLS else ClodexWorkflow(root) if name == "clodex_plan": @@ -589,9 +618,25 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: except (KeyError, TypeError, ValueError, sqlite3.IntegrityError) as exc: return call_text(expected_handoff_error(exc), is_error=True) return call_json(run, is_error=run.get("status") == "blocked") + elif name == "clodex_delegate": + manager = DelegationManager(root) + try: + delegation = manager.start( + str(arguments["run_id"]), + instructions=arguments.get("instructions"), + mode=str(arguments.get("mode") or "implement"), + approval_profile=arguments.get("approval_profile"), + ) + if arguments.get("wait") is True: + delegation = manager.wait(int(delegation["id"]), should_stop=should_stop) + except (ValueError, DirtyWorkspaceError, subprocess.CalledProcessError) as exc: + return call_text(expected_handoff_error(exc), is_error=True) + failed = delegation["status"] in {"failed", "cancelled"} + return call_json({"delegation": delegation, "handoff": brief_handoff(manager.state, str(arguments["run_id"]))}, is_error=failed) elif name == "clodex_handoff_get": try: run_id = str(required_argument(arguments, "run_id")) + DelegationManager(root).reconcile_active(run_id) data = workflow.state.get_handoff(run_id) except (KeyError, TypeError, ValueError, sqlite3.IntegrityError) as exc: return call_text(expected_handoff_error(exc), is_error=True) @@ -653,6 +698,23 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]: return {"content": [{"type": "text", "text": json.dumps(result.__dict__, indent=2)}], "isError": result.status == "blocked"} +def brief_handoff(state: Any, run_id: str) -> dict[str, Any]: + """The few handoff fields an orchestrator needs right after delegating.""" + data = state.get_handoff(run_id) + if data is None: + return {} + run = data["run"] + return { + "run_id": run_id, + "status": run.get("status"), + "phase": run.get("phase"), + "diff_hash": run.get("diff_hash"), + "workspace_path": run.get("workspace_path"), + "budget_remaining": data["budget_remaining"], + "next_expected_actor": data["next_expected_actor"], + } + + def call_text(text: str, is_error: bool = False) -> dict[str, Any]: return {"content": [{"type": "text", "text": text}], "isError": is_error} diff --git a/clodex/native.py b/clodex/native.py index 82ee7df..f8c0c6a 100644 --- a/clodex/native.py +++ b/clodex/native.py @@ -52,13 +52,14 @@ def build_claude_block() -> str: When a user asks for non-trivial design, implementation, debugging, refactoring, or review work: -1. Prefer the MCP tool `clodex_handoff_create` to create or resume a durable handoff. -2. Use `clodex_handoff_get` before acting on an existing handoff. -3. Produce the plan, design, acceptance criteria, risks, and test commands before implementation. -4. Delegate implementation-oriented work to Codex through Clodex instead of manually telling the user to run Codex. -5. Use `clodex_handoff_update` to record plan artifacts, phase changes, review notes, changed files, tests, unresolved issues, and diff hashes. -6. Use `clodex_handoff_decide` before claiming completion when Codex has implemented or audited changes. -7. Stop and summarize for the user when Clodex reports `blocked`. +1. Prefer the MCP tool `clodex_handoff_create` to create or resume a durable handoff. Pass `workspace: "git-worktree"` so Codex works in an isolated checkout. +2. Use `clodex_handoff_get` before acting on an existing handoff; it shows the budget, the next expected actor, and any running delegation. +3. Produce the plan, design, acceptance criteria, risks, and test commands before implementation, and record them with `clodex_handoff_update`. +4. Delegate implementation to Codex with `clodex_delegate` (mode `implement`, `instructions` = the accepted plan) instead of telling the user to run Codex. Pass `wait: true`, or poll `clodex_handoff_get`. +5. Review the diff in the handoff's workspace, then record your verdict with `clodex_handoff_update` (actor `claude`, the diff hash from `clodex_handoff_get`, `report.approved`, `report.required_fixes`). +6. Ask Codex for its verdict with `clodex_delegate` mode `audit`. If anyone rejected, use mode `fix` (it applies the open findings) and audit again; approvals only count for the exact diff they name. +7. Use `clodex_handoff_decide` before claiming completion. `approved` means the configured reviewers agree on this exact diff; then `clodex apply ` brings it into the main checkout. +8. Stop and summarize for the user when Clodex reports `blocked`. Default role contract: @@ -95,6 +96,7 @@ def build_agents_block() -> str: 6. Use `clodex_handoff_decide` before claiming completion after implementation or audit work. 7. Ask for clarification through Clodex when product intent or acceptance criteria are unclear. 8. Respect owner, phase, and handoff budget state. When the budget is exhausted, summarize the disagreement for the user. +9. When Clodex delegates work to you it runs `codex exec` in the handoff's workspace and records your result itself, so you do not need to call the handoff tools in that case; just do the work and finish with a concise report. Default role contract: diff --git a/clodex/procs.py b/clodex/procs.py index c52be9c..684c2ca 100644 --- a/clodex/procs.py +++ b/clodex/procs.py @@ -10,9 +10,40 @@ import signal import subprocess import time +from datetime import UTC, datetime from pathlib import Path +HEARTBEAT_INTERVAL = 5.0 +# A live worker beats every few seconds; a pid with an older heartbeat is probably a +# reused pid, so it is never signalled. +STALE_AFTER = 60.0 +START_GRACE = 30.0 + + +def parse_time(value: str | None) -> datetime | None: + if not value: + return None + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + + +def worker_state(pid: int | None, heartbeat_at: str | None, started_at: str | None) -> str: + """`none` (no worker), `dead`, `stale` (alive pid but no recent heartbeat) or `alive`.""" + if not pid: + return "none" + if not pid_alive(int(pid)): + return "dead" + heartbeat = parse_time(heartbeat_at) + reference = heartbeat or parse_time(started_at) + limit = STALE_AFTER if heartbeat else START_GRACE + if reference is not None and (datetime.now(UTC) - reference).total_seconds() > limit: + return "stale" + return "alive" + + def popen_isolation_kwargs() -> dict[str, object]: """Start a child in its own process group/session so its whole tree can be stopped.""" if os.name == "nt": diff --git a/clodex/prompts.py b/clodex/prompts.py index fdf1cdd..e7a9b2b 100644 --- a/clodex/prompts.py +++ b/clodex/prompts.py @@ -108,3 +108,28 @@ def audit_diff_excerpt(diff: str, max_bytes: int) -> str: f"Changed files ({len(files)}): {', '.join(files)}. " "Reject if the omitted part could hide a problem you cannot rule out.]\n" ) + + +def delegate_prompt(mode: str, task: str, instructions: str | None, fixes: list[str] | None = None) -> str: + """The prompt for a Codex job a native handoff delegated (implement or fix).""" + lines = [ + "You are the Codex engineering wave for Clodex, working a native Claude/Codex handoff.", + "", + "Work only inside the current directory: it is an isolated checkout made for this handoff. " + "Keep changes scoped, preserve existing user changes, add or update tests where appropriate, and run the relevant verification.", + "", + "Original task:", + task, + ] + if instructions: + lines += ["", "Instructions from Claude:", instructions] + if mode == "fix": + lines += ["", "Required fixes:", json.dumps(fixes or ["Resolve the open review findings."], indent=2), "", "Apply only the required fixes. Do not broaden scope."] + lines += [ + "", + "When finished, print a concise Markdown report with:", + "- files changed", + "- tests run and pass/fail status", + "- unresolved issues, if any", + ] + return "\n".join(lines) + "\n" diff --git a/clodex/state.py b/clodex/state.py index 46a6396..e7b6f62 100644 --- a/clodex/state.py +++ b/clodex/state.py @@ -10,6 +10,9 @@ TERMINAL_STATUSES = {"approved", "blocked", "failed", "cancelled", "applied", "completed"} HANDOFF_REASON_STATUSES = {"blocked", "failed"} +DELEGATION_MODES = {"implement", "fix", "audit"} +DELEGATION_ACTIVE = {"queued", "running"} +DELEGATION_FINISHED = {"completed", "failed", "cancelled"} HANDOFF_PHASES = {"planning", "implementation", "audit", "fix", "decision", "done"} HANDOFF_ACTIVE_STATUSES = {"handoff"} | HANDOFF_REASON_STATUSES # A finished run stays finished, except that it can then be applied (an approved run @@ -126,6 +129,22 @@ def _init(self) -> None: created_at text not null, released_at text ); + create table if not exists delegations ( + id integer primary key autoincrement, + run_id text not null, + mode text not null, + instructions text, + approval_profile text, + status text not null, + pid integer, + error text, + summary text, + diff_hash text, + created_at text not null, + started_at text, + finished_at text, + heartbeat_at text + ); create table if not exists cancellations ( run_id text primary key, requested integer not null, @@ -137,10 +156,10 @@ def _init(self) -> None: 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("insert into schema_version(version) values (3)") + elif int(current["version"]) < 3: con.execute("delete from schema_version") - con.execute("insert into schema_version(version) values (2)") + con.execute("insert into schema_version(version) values (3)") def _ensure_run_columns(self, con: sqlite3.Connection) -> None: existing = {row["name"] for row in con.execute("pragma table_info(runs)")} @@ -500,6 +519,7 @@ def get_handoff(self, run_id: str) -> dict[str, Any] | None: events.append(event) artifacts = [dict(artifact) for artifact in con.execute("select * from artifacts where run_id=? order by id", (run_id,))] + delegations = [dict(item) for item in con.execute("select * from delegations where run_id=? order by id desc limit 5", (run_id,))] handoff_count = _int_or_default(run.get("handoff_count"), 0) handoff_budget = _int_or_default(run.get("handoff_budget"), 6) last_actor = run.get("last_actor") @@ -513,6 +533,7 @@ def get_handoff(self, run_id: str) -> dict[str, Any] | None: "run": run, "events": events, "artifacts": artifacts, + "delegations": delegations, "budget_remaining": max(handoff_budget - handoff_count, 0), "next_expected_actor": next_expected_actor, } @@ -576,6 +597,67 @@ def touch_run(self, run_id: str) -> None: (now_iso(), run_id, *sorted(TERMINAL_STATUSES)), ) + # ------------------------------------------------------------ delegations + # A delegation is one Codex job a handoff asked for (implement / fix / audit). + + def start_delegation(self, run_id: str, mode: str, instructions: str | None = None, approval_profile: str | None = None) -> dict[str, Any]: + """Record a new delegation unless one is already active for this handoff (atomic).""" + if mode not in DELEGATION_MODES: + raise ValueError(f"unknown delegation mode: {mode} (use one of {sorted(DELEGATION_MODES)})") + timestamp = now_iso() + with self.session() as con: + con.execute("begin immediate") + marks = ",".join("?" for _ in DELEGATION_ACTIVE) + active = con.execute(f"select id from delegations where run_id=? and status in ({marks})", (run_id, *sorted(DELEGATION_ACTIVE))).fetchone() + if active is not None: + raise ValueError(f"a delegation is already running for this handoff (delegation {active['id']})") + cursor = con.execute( + "insert into delegations(run_id, mode, instructions, approval_profile, status, created_at) values (?, ?, ?, ?, 'queued', ?)", + (run_id, mode, instructions, approval_profile, timestamp), + ) + return dict(con.execute("select * from delegations where id=?", (cursor.lastrowid,)).fetchone()) + + _DELEGATION_FIELDS = {"status", "pid", "error", "summary", "diff_hash", "started_at", "finished_at"} + + def update_delegation(self, delegation_id: int, **fields: Any) -> dict[str, Any]: + unknown = set(fields) - self._DELEGATION_FIELDS + if unknown: + raise ValueError(f"unknown delegation fields: {sorted(unknown)}") + if fields.get("status") in DELEGATION_FINISHED: + fields.setdefault("finished_at", now_iso()) + with self.session() as con: + con.execute("begin immediate") + row = con.execute("select * from delegations where id=?", (delegation_id,)).fetchone() + if row is None: + raise ValueError(f"unknown delegation: {delegation_id}") + if str(row["status"]) in DELEGATION_FINISHED and fields.get("status") not in (None, row["status"]): + return dict(row) # a finished delegation stays finished (e.g. a worker that outlived a cancel) + if fields: + assignments = ", ".join(f"{name}=?" for name in fields) + con.execute(f"update delegations set {assignments} where id=?", (*fields.values(), delegation_id)) + return dict(con.execute("select * from delegations where id=?", (delegation_id,)).fetchone()) + + def touch_delegation(self, delegation_id: int) -> None: + marks = ",".join("?" for _ in DELEGATION_ACTIVE) + with self.session() as con: + con.execute(f"update delegations set heartbeat_at=? where id=? and status in ({marks})", (now_iso(), delegation_id, *sorted(DELEGATION_ACTIVE))) + + def get_delegation(self, delegation_id: int) -> dict[str, Any] | None: + with self.session() as con: + row = con.execute("select * from delegations where id=?", (delegation_id,)).fetchone() + return dict(row) if row else None + + def list_delegations(self, run_id: str, limit: int = 10) -> list[dict[str, Any]]: + with self.session() as con: + rows = con.execute("select * from delegations where run_id=? order by id desc limit ?", (run_id, limit)) + return [dict(row) for row in rows] + + def active_delegation(self, run_id: str) -> dict[str, Any] | None: + marks = ",".join("?" for _ in DELEGATION_ACTIVE) + with self.session() as con: + row = con.execute(f"select * from delegations where run_id=? and status in ({marks}) order by id desc limit 1", (run_id, *sorted(DELEGATION_ACTIVE))).fetchone() + return dict(row) if row else None + def release_workspace_lock(self, run_id: str) -> None: with self.session() as con: con.execute("update workspace_locks set released_at=? where run_id=? and released_at is null", (now_iso(), run_id)) diff --git a/clodex/tasks.py b/clodex/tasks.py index 2a6ccf1..920ce66 100644 --- a/clodex/tasks.py +++ b/clodex/tasks.py @@ -5,41 +5,27 @@ import sys import threading import time -from datetime import UTC, datetime +from collections.abc import Callable 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 .models import ensure_usable -from .procs import kill_tree, pid_alive, popen_isolation_kwargs +from .delegate import cancel_active_delegation +from .procs import HEARTBEAT_INTERVAL, kill_tree, popen_isolation_kwargs, worker_state as shared_worker_state from .state import TERMINAL_STATUSES, StateStore from .workflow import WorkflowResult from .workspace import WorkspaceManager ACTIVE_STATUSES = {"queued", "running", "planning", "auditing", "needs-fix", "cancel_requested"} -HEARTBEAT_INTERVAL = 5.0 -# A live worker beats every few seconds; a pid with an older heartbeat is probably a -# reused pid, so it is never signalled. -STALE_AFTER = 60.0 -START_GRACE = 30.0 - - -def _parse_time(value: str | None) -> datetime | None: - if not value: - return None - try: - return datetime.fromisoformat(value.replace("Z", "+00:00")) - except ValueError: - return None class Heartbeat: - """Keeps `runs.heartbeat_at` fresh while a worker process is doing the run.""" + """Keeps a worker's heartbeat fresh while it works (a run's by default, or whatever `beat` touches).""" - def __init__(self, state: StateStore, run_id: str, interval: float = HEARTBEAT_INTERVAL): - self.state = state - self.run_id = run_id + def __init__(self, state: StateStore, key: Any, interval: float = HEARTBEAT_INTERVAL, beat: Callable[[], None] | None = None): + self._beat_once = beat or (lambda: state.touch_run(key)) self.interval = interval self._stop = threading.Event() self._thread = threading.Thread(target=self._beat, name="clodex-heartbeat", daemon=True) @@ -47,7 +33,7 @@ def __init__(self, state: StateStore, run_id: str, interval: float = HEARTBEAT_I def _beat(self) -> None: while True: try: - self.state.touch_run(self.run_id) + self._beat_once() except Exception: # noqa: BLE001 - a missed beat must never kill the worker pass if self._stop.wait(self.interval): @@ -134,17 +120,7 @@ def start( def worker_state(self, run: dict[str, Any]) -> str: """`none` (no worker), `dead`, `stale` (alive pid but no recent heartbeat) or `alive`.""" - pid = run.get("pid") - if not pid: - return "none" - if not pid_alive(int(pid)): - return "dead" - heartbeat = _parse_time(run.get("heartbeat_at")) - reference = heartbeat or _parse_time(run.get("started_at")) or _parse_time(run.get("created_at")) - limit = STALE_AFTER if heartbeat else START_GRACE - if reference is not None and (datetime.now(UTC) - reference).total_seconds() > limit: - return "stale" - return "alive" + return shared_worker_state(run.get("pid"), run.get("heartbeat_at"), run.get("started_at") or run.get("created_at")) def reconcile(self, run: dict[str, Any]) -> dict[str, Any]: """Notice a worker that died without finishing its run, instead of leaving it queued forever.""" @@ -184,6 +160,7 @@ def cancel(self, run_id: str) -> WorkflowResult: if self.worker_state(run) == "alive": kill_tree(int(run["pid"])) killed = True + delegation_stopped = cancel_active_delegation(self.state, run_id) self.state.complete_cancel(run_id) released = self._release_workspace(run) updated = self.state.get_run(run_id) or run @@ -192,7 +169,7 @@ def cancel(self, run_id: str) -> WorkflowResult: run_id, updated.get("task_id"), updated.get("artifacts_dir"), - {"cancel_requested": True, "worker_stopped": killed, "workspace_released": released}, + {"cancel_requested": True, "worker_stopped": killed or delegation_stopped, "workspace_released": released}, ) def _release_workspace(self, run: dict[str, Any]) -> bool: diff --git a/clodex/workflow.py b/clodex/workflow.py index 0b3f20e..b7b2824 100644 --- a/clodex/workflow.py +++ b/clodex/workflow.py @@ -397,6 +397,10 @@ def _run_reviewers( artifacts.write_json("04-codex-audit.json", first_codex) return verdicts + def run_agent_json(self, runner: AgentRunner, command, prompt: str, label: str, timeout: int | None = None) -> dict[str, Any]: + """Run an agent command and return its JSON answer (validated against the command's schema, retried once).""" + return self._run_json_with_retry(runner, command, prompt, label, None, timeout=timeout) + def _run_json_with_retry( self, runner: AgentRunner, diff --git a/tests/support.py b/tests/support.py index bc0fbe0..63a841a 100644 --- a/tests/support.py +++ b/tests/support.py @@ -83,6 +83,7 @@ def __init__( claude_logged_in: bool = True, codex_logged_in: bool = True, codex_catalog: dict | None = None, + codex_fails: bool = False, ): self.reject_once = reject_once self.malformed_once = malformed_once @@ -96,6 +97,7 @@ def __init__( self.claude_logged_in = claude_logged_in self.codex_logged_in = codex_logged_in self.codex_catalog = codex_catalog + self.codex_fails = codex_fails self.sleep_seconds = sleep_seconds self.include_clodex = include_clodex @@ -307,6 +309,9 @@ def emit(text, is_error=False): fail('error: invalid sandbox ' + sandbox) if '-C' in flags: os.chdir(flags['-C'][-1]) + if {self.codex_fails!r} and 'adversarial auditor' not in stdin: + sys.stderr.write('simulated codex crash\\n') + raise SystemExit(5) def finish(text): out = flags.get('-o') or flags.get('--output-last-message') diff --git a/tests/test_delegate.py b/tests/test_delegate.py new file mode 100644 index 0000000..1d62777 --- /dev/null +++ b/tests/test_delegate.py @@ -0,0 +1,409 @@ +"""Tests: delegating a native handoff's work to Codex (implement / fix / audit).""" + +from __future__ import annotations + +import json +import os +import subprocess +import sys +import time +import unittest +from pathlib import Path +from unittest import mock + +from clodex import mcp_server +from clodex.artifacts import current_diff, hash_text +from clodex.config import DEFAULT_CONFIG +from clodex.delegate import DelegationManager, WaitAborted, cancel_active_delegation, open_fixes, reconcile_delegation +from clodex.procs import pid_alive +from clodex.state import StateStore +from clodex.tasks import TaskManager +from tests.support import ROOT, FakeCliPath, TempRepo +from tests.test_handoff_quorum import HandoffCase, update_event + +REVIEWERS = DEFAULT_CONFIG["audit"]["reviewers"] + + +def wait_for(predicate, timeout: float = 30.0, interval: float = 0.1) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(interval) + return predicate() + + +class DelegateCase(HandoffCase): + def setUp(self): + super().setUp() + warning_ctx = mock.patch("warnings.filterwarnings") # keep Popen ResourceWarnings out of the output + warning_ctx.start() + self.addCleanup(warning_ctx.stop) + self.addCleanup(self.stop_workers) + + def stop_workers(self): + state = self.state + with state.session() as con: + run_ids = [row["run_id"] for row in con.execute("select distinct run_id from delegations")] + for run_id in run_ids: + cancel_active_delegation(state, run_id) + + def create(self, run_id: str = "d", workspace: str = "git-worktree", **extra): + is_error, run = self.call("clodex_handoff_create", run_id=run_id, task="Add a feature", workspace=workspace, **extra) + self.assertFalse(is_error, run) + return run + + def delegate(self, run_id: str = "d", wait: bool = True, **arguments): + return self.call("clodex_delegate", run_id=run_id, wait=wait, **arguments) + + def handoff(self, run_id: str = "d") -> dict: + return self.call("clodex_handoff_get", run_id=run_id)[1] + + def workspace(self, run_id: str = "d") -> Path: + return Path(self.state.get_run(run_id)["workspace_path"]) + + +class ImplementTests(DelegateCase): + def test_implement_runs_in_the_isolated_worktree_and_records_the_result(self): + with FakeCliPath(): + self.create() + is_error, result = self.delegate(instructions="Create implemented.txt") + self.assertFalse(is_error, result) + delegation = result["delegation"] + self.assertEqual((delegation["status"], delegation["mode"]), ("completed", "implement")) + self.assertTrue(delegation["pid"]) + self.assertEqual((self.workspace() / "implemented.txt").read_text(encoding="utf-8").strip(), "implemented") + self.assertFalse((self.repo / "implemented.txt").exists(), "the source checkout must stay untouched") + data = self.handoff() + run = data["run"] + self.assertEqual((run["phase"], run["last_actor"], run["handoff_count"]), ("implementation", "codex", 1)) + self.assertEqual(data["budget_remaining"], 5) + self.assertEqual(data["next_expected_actor"], "claude") + self.assertEqual(run["diff_hash"], delegation["diff_hash"]) + self.assertEqual(delegation["diff_hash"], hash_text(current_diff(self.workspace())), "the recorded hash is the workspace's real diff") + report = next(e for e in reversed(data["events"]) if e["event"] == "handoff.update")["data"]["report"] + self.assertEqual((report["mode"], report["changed"], report["summary"]), ("implement", True, "implemented")) + names = {a["name"] for a in data["artifacts"]} + self.assertTrue({f"delegation-{delegation['id']}-codex.md", "changes.diff"} <= names) + self.assertEqual(data["delegations"][0]["id"], delegation["id"]) + self.assertEqual(result["handoff"]["next_expected_actor"], "claude") + + def test_default_returns_at_once_and_progress_is_visible(self): + with FakeCliPath(sleep_seconds=3): + self.create() + started = time.monotonic() + is_error, result = self.delegate(wait=False, instructions="slow job") + self.assertFalse(is_error, result) + self.assertLess(time.monotonic() - started, 20, "must not wait for Codex") + self.assertIn(result["delegation"]["status"], {"queued", "running"}) + self.assertTrue(result["delegation"]["pid"]) + self.assertEqual(self.handoff()["delegations"][0]["id"], result["delegation"]["id"]) + manager = DelegationManager(self.repo) + finished = manager.wait(int(result["delegation"]["id"]), timeout=120) + self.assertEqual(finished["status"], "completed") + self.assertEqual(self.handoff()["run"]["handoff_count"], 1) + + def test_a_worktree_is_created_on_demand_when_the_handoff_has_none(self): + with FakeCliPath(): + self.create(workspace="none") + self.assertIsNone(self.state.get_run("d")["workspace_path"]) + is_error, result = self.delegate(instructions="go") + self.assertFalse(is_error, result) + path = self.workspace() + self.assertTrue(path.is_dir()) + self.assertEqual(path.parent.resolve(), (self.repo / ".clodex" / "workspaces").resolve()) + self.assertTrue((path / "implemented.txt").is_file()) + with self.state.session() as con: + self.assertIsNotNone(con.execute("select 1 from workspace_locks where run_id='d'").fetchone()) + + def test_a_local_workspace_edits_the_repo_itself(self): + with FakeCliPath(): + self.create(workspace="local") + is_error, result = self.delegate(instructions="go") + self.assertFalse(is_error, result) + self.assertTrue((self.repo / "implemented.txt").is_file()) + + def test_codex_failure_fails_the_delegation_without_spending_budget(self): + with FakeCliPath(codex_fails=True): + self.create() + is_error, result = self.delegate(instructions="go") + self.assertTrue(is_error) + self.assertEqual(result["delegation"]["status"], "failed") + self.assertIn("codex exited with code 5", result["delegation"]["error"]) + data = self.handoff() + self.assertEqual(data["run"]["handoff_count"], 0, "a failed job is not a handoff") + self.assertEqual(data["run"]["status"], "handoff", "Claude decides what to do next") + failure = next(e for e in reversed(data["events"]) if e["event"] == "handoff.update")["data"]["report"] + self.assertIn("simulated codex crash", failure["error"]) + self.assertEqual(data["next_expected_actor"], "claude") + self.assertTrue(any(a["name"].endswith("-codex.md") for a in data["artifacts"]), "the transcript is kept") + + +class FixTests(DelegateCase): + def test_open_fixes_collects_unapproved_findings_for_the_latest_diff(self): + data = {"events": [ + update_event("claude", "h1", approved=False, required_fixes=["old problem"]), + update_event("claude", "h2", approved=False, required_fixes=["add tests", "add tests"], findings=[{"severity": "high", "file": "a.py", "line": 3, "message": "crash"}]), + update_event("codex", "h2", approved=True), + update_event("codex", "h2", approved=False, reviewer_id="security", required_fixes=["escape input"]), + {"event": "handoff.create", "data": {}}, + ]} + reviewers = REVIEWERS + self.assertEqual(open_fixes(data, reviewers), ["add tests", "[high] a.py:3: crash"], "security is not configured as required here, so only claude-plan counts") + approved = {"events": [update_event("claude", "h", approved=True), update_event("codex", "h", approved=True)]} + self.assertEqual(open_fixes(approved, reviewers), []) + self.assertEqual(open_fixes({"events": []}, reviewers), []) + + def test_fix_mode_applies_open_findings(self): + with FakeCliPath(): + self.create() + self.delegate(instructions="Create implemented.txt") + run = self.state.get_run("d") + self.call("clodex_handoff_update", run_id="d", actor="claude", diff_hash=run["diff_hash"], report={"approved": False, "required_fixes": ["append a fixed line"]}) + is_error, result = self.delegate(mode="fix") + self.assertFalse(is_error, result) + self.assertEqual(result["delegation"]["mode"], "fix") + self.assertEqual((self.workspace() / "implemented.txt").read_text(encoding="utf-8").replace("\r\n", "\n"), "implemented\nfixed\n", "the fix prompt carried the required fixes") + data = self.handoff() + self.assertEqual(data["run"]["phase"], "fix") + self.assertEqual(data["run"]["handoff_count"], 2) + self.assertNotEqual(data["run"]["diff_hash"], run["diff_hash"], "the diff changed, so earlier approvals no longer count") + + +class AuditTests(DelegateCase): + def test_full_native_loop_implement_codex_audit_claude_verdict_decide(self): + with FakeCliPath(): + self.create() + self.delegate(instructions="Create implemented.txt") + diff_hash = self.state.get_run("d")["diff_hash"] + is_error, result = self.delegate(mode="audit") + self.assertFalse(is_error, result) + self.assertEqual(result["delegation"]["mode"], "audit") + update = next(e for e in reversed(self.handoff()["events"]) if e["event"] == "handoff.update")["data"] + self.assertEqual((update["actor"], update["diff_hash"]), ("codex", diff_hash)) + self.assertTrue(update["report"]["approved"]) + self.assertEqual(update["report"]["reviewer_id"], "codex-architecture") + self.assertEqual(self.call("clodex_handoff_decide", run_id="d")[1]["decision"], "needs_fix", "claude has not given its verdict yet") + self.call("clodex_handoff_update", run_id="d", actor="claude", diff_hash=diff_hash, report={"approved": True, "summary": "plan adherence ok"}) + is_error, decision = self.call("clodex_handoff_decide", run_id="d") + self.assertEqual(decision["decision"], "approved") + self.assertEqual(decision["approved_by"], ["claude", "codex"]) + self.assertEqual(self.state.get_run("d")["status"], "approved") + with self.state.session() as con: + audits = [dict(r) for r in con.execute("select agent, approved from audits where run_id='d' order by id")] + self.assertEqual(audits, [{"agent": "codex-architecture", "approved": 1}, {"agent": "claude", "approved": 1}]) + + def test_a_rejecting_codex_audit_blocks_approval_and_feeds_the_next_fix(self): + with FakeCliPath(reject_reviewers=("codex-architecture",)): + self.create() + self.delegate(instructions="Create implemented.txt") + diff_hash = self.state.get_run("d")["diff_hash"] + self.delegate(mode="audit") + self.call("clodex_handoff_update", run_id="d", actor="claude", diff_hash=diff_hash, report={"approved": True}) + decision = self.call("clodex_handoff_decide", run_id="d")[1] + self.assertEqual(decision["decision"], "needs_fix") + self.assertEqual(decision["required_pending"], ["codex-architecture"]) + self.assertEqual(open_fixes(self.handoff(), self.repo_reviewers()), ["fix from codex-architecture"]) + + def repo_reviewers(self): + from clodex.config import load_config + + return load_config(self.repo).reviewers + + def test_auditing_a_workspace_with_no_changes_fails_cleanly(self): + with FakeCliPath(): + self.create() + is_error, result = self.delegate(mode="audit") + self.assertTrue(is_error) + self.assertIn("nothing to audit", result["delegation"]["error"]) + self.assertEqual(self.handoff()["run"]["handoff_count"], 0) + + +class RefusalTests(DelegateCase): + def test_bad_requests_are_tool_errors(self): + with FakeCliPath(): + is_error, message = self.delegate(run_id="nope", instructions="x") + self.assertTrue(is_error) + self.assertIn("Unknown run: nope", message) + self.create() + is_error, message = self.delegate() + self.assertTrue(is_error) + self.assertIn("instructions are required", message) + is_error, message = self.delegate(mode="vibe", instructions="x") + self.assertTrue(is_error) + self.assertIn("unknown delegation mode", message) + + def test_finished_and_exhausted_handoffs_cannot_delegate(self): + with FakeCliPath(): + self.create(run_id="done", workspace="none") + self.call("clodex_handoff_update", run_id="done", actor="claude", status="blocked", blocked_reason="stop") + is_error, message = self.delegate(run_id="done", instructions="x") + self.assertTrue(is_error) + self.assertIn("the handoff is blocked", message) + self.create(run_id="tiny", handoff_budget=1) + self.assertFalse(self.delegate(run_id="tiny", instructions="first")[0]) + is_error, message = self.delegate(run_id="tiny", instructions="second") + self.assertTrue(is_error) + self.assertTrue("budget exhausted" in message or "is blocked" in message, message) + + def test_a_retired_codex_model_is_refused_up_front(self): + (self.repo / "CLODEX.md").write_text("---\ncodex:\n model: gpt-5.4\n---\nbody\n", encoding="utf-8") + with FakeCliPath(): + self.create(workspace="none") + is_error, message = self.delegate(instructions="x") + self.assertTrue(is_error) + self.assertIn("retired on", message) + self.assertEqual(self.handoff()["delegations"], []) + + def test_only_one_delegation_at_a_time_per_handoff(self): + with FakeCliPath(sleep_seconds=4): + self.create() + first = self.delegate(wait=False, instructions="slow")[1]["delegation"] + is_error, message = self.delegate(wait=False, instructions="second") + self.assertTrue(is_error) + self.assertIn("already running", message) + DelegationManager(self.repo).wait(int(first["id"]), timeout=120) + is_error, result = self.delegate(wait=False, instructions="third") + self.assertFalse(is_error, result) + DelegationManager(self.repo).wait(int(result["delegation"]["id"]), timeout=120) + + +class SupervisionTests(DelegateCase): + def test_a_dead_worker_is_noticed_and_does_not_block_the_handoff(self): + with FakeCliPath(): + self.create(workspace="none") + run = self.state.get_run("d") + gone = subprocess.Popen([sys.executable, "-c", "pass"]) + gone.wait() + delegation = self.state.start_delegation("d", "implement", "x") + self.state.update_delegation(delegation["id"], status="running", pid=gone.pid) + self.call("clodex_handoff_get", run_id="d") # reconciles + fixed = self.state.get_delegation(delegation["id"]) + self.assertEqual(fixed["status"], "failed") + self.assertIn("worker process exited", fixed["error"]) + is_error, result = self.delegate(instructions="try again") + self.assertFalse(is_error, result) + self.assertEqual(run["status"], "handoff") + + def test_a_delegation_that_never_got_a_worker_fails_after_the_grace_period(self): + with TempRepo() as repo: + state = StateStore(repo / "s.sqlite3") + delegation = state.start_delegation("r", "implement", "x") + self.assertEqual(reconcile_delegation(state, delegation)["status"], "queued", "just created: still starting") + with state.session() as con: + con.execute("update delegations set created_at='2020-01-01T00:00:00Z' where id=?", (delegation["id"],)) + fresh = state.get_delegation(delegation["id"]) + self.assertEqual(reconcile_delegation(state, fresh)["status"], "failed") + + def test_cancelling_the_handoff_stops_the_worker_and_releases_the_worktree(self): + with FakeCliPath(sleep_seconds=60): + self.create() + started = self.delegate(wait=False, instructions="slow")[1]["delegation"] + workspace = self.workspace() + self.assertTrue(wait_for(lambda: (workspace / ".fake-pid-codex").exists(), 60), "codex never started") + worker_pid = started["pid"] + codex_pid = int((workspace / ".fake-pid-codex").read_text(encoding="utf-8")) + self.assertTrue(pid_alive(worker_pid) and pid_alive(codex_pid)) + result = TaskManager(self.repo).cancel("d") + self.assertEqual(result.status, "cancelled") + self.assertTrue(result.data["worker_stopped"]) + self.assertTrue(wait_for(lambda: not pid_alive(worker_pid), 15), "worker survived") + self.assertTrue(wait_for(lambda: not pid_alive(codex_pid), 15), "codex survived") + self.assertEqual(self.state.get_delegation(started["id"])["status"], "cancelled") + self.assertEqual(self.state.get_run("d")["status"], "cancelled") + self.assertTrue(result.data["workspace_released"]) + self.assertFalse(workspace.exists()) + + def test_cancel_only_signals_a_worker_with_a_recent_heartbeat(self): + with TempRepo() as repo: + state = StateStore(repo / "s.sqlite3") + delegation = state.start_delegation("r", "implement", "x") + # This pid is the test process itself, with a stale heartbeat: it must never be signalled. + state.update_delegation(delegation["id"], status="running", pid=os.getpid(), started_at="2020-01-01T00:00:00Z") + self.assertFalse(cancel_active_delegation(state, "r")) + self.assertEqual(state.get_delegation(delegation["id"])["status"], "cancelled") + self.assertTrue(pid_alive(os.getpid())) + self.assertFalse(cancel_active_delegation(state, "r"), "nothing left to cancel") + + def test_wait_can_be_aborted(self): + with TempRepo() as repo: + state = StateStore(repo / "s.sqlite3") + state.start_delegation("r", "implement", "x") + manager = DelegationManager(repo) + manager._state = state + calls = [] + with self.assertRaises(WaitAborted): + manager.wait(1, should_stop=lambda: calls.append(1) or len(calls) > 1) + self.assertEqual(manager.wait(1, timeout=0.2)["status"], "queued", "a timeout returns the current state") + + +class DelegationStateTests(unittest.TestCase): + def test_v2_database_upgrades_and_gains_the_table(self): + import sqlite3 + import tempfile + + with tempfile.TemporaryDirectory() as tmp: + db = Path(tmp) / "old.sqlite3" + con = sqlite3.connect(db) + con.executescript("create table schema_version(version integer not null); insert into schema_version values (2);") + con.commit() + con.close() + store = StateStore(db) + self.assertEqual(store.schema_version(), 3) + self.assertIn("delegations", store.table_names()) + + def test_start_is_atomic_updates_are_sticky_and_listing_is_newest_first(self): + with TempRepo() as repo: + state = StateStore(repo / "s.sqlite3") + first = state.start_delegation("r", "audit") + with self.assertRaisesRegex(ValueError, "already running"): + state.start_delegation("r", "fix") + self.assertEqual(state.active_delegation("r")["id"], first["id"]) + state.update_delegation(first["id"], status="completed", summary="ok") + self.assertIsNone(state.active_delegation("r")) + self.assertEqual(state.update_delegation(first["id"], status="running")["status"], "completed", "finished stays finished") + second = state.start_delegation("r", "fix") + self.assertEqual([d["id"] for d in state.list_delegations("r")], [second["id"], first["id"]]) + with self.assertRaises(ValueError): + state.start_delegation("r", "dance") + with self.assertRaises(ValueError): + state.update_delegation(second["id"], colour="red") + with self.assertRaises(ValueError): + state.update_delegation(999, status="failed") + + +class McpSurfaceTests(unittest.TestCase): + def test_tool_definition_and_blocking_rules(self): + tool = mcp_server.TOOL_INDEX["clodex_delegate"] + self.assertEqual(tool["inputSchema"]["required"], ["run_id"]) + self.assertEqual(tool["inputSchema"]["properties"]["mode"]["enum"], ["implement", "fix", "audit"]) + server = mcp_server.McpServer(out=mock.Mock()) + + def blocking(**arguments): + return server.is_blocking("tools/call", {"name": "clodex_delegate", "arguments": arguments}) + + self.assertTrue(blocking(run_id="x", wait=True)) + self.assertFalse(blocking(run_id="x", wait=False)) + self.assertFalse(blocking(run_id="x")) + self.assertFalse(blocking(run_id="x", wait="yes")) + + def test_argument_problems_never_reach_the_manager(self): + with TempRepo() as repo, mock.patch.dict(os.environ, {"CLODEX_REPO_ROOT": str(repo)}): + server = mcp_server.McpServer(out=mock.Mock()) + for arguments, expected in (({}, "Missing required argument: run_id"), ({"run_id": "x", "mode": "vibe"}, "Invalid arguments"), ({"run_id": "x", "wait": "yes"}, "Invalid arguments")): + result = server.tools_call({"name": "clodex_delegate", "arguments": arguments}) + self.assertTrue(result["isError"]) + self.assertIn(expected, result["content"][0]["text"]) + + def test_the_worker_command_rejects_unknown_delegations(self): + with TempRepo() as repo: + result = subprocess.run( + [sys.executable, "-m", "clodex", "task", "delegate-worker", "no-run", "999"], + cwd=repo, env={**os.environ, "PYTHONPATH": str(ROOT)}, capture_output=True, text=True, stdin=subprocess.DEVNULL, + ) + self.assertEqual(result.returncode, 2) + self.assertIn("unknown delegation 999", result.stderr) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_state.py b/tests/test_state.py index 11bd8a5..672a293 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -20,7 +20,7 @@ def test_state_migrations_add_v2_tables(self): self.assertIn("artifacts", tables) self.assertIn("workspace_locks", tables) self.assertIn("cancellations", tables) - self.assertEqual(store.schema_version(), 2) + self.assertEqual(store.schema_version(), 3) def test_state_migrations_add_native_handoff_columns(self): with tempfile.TemporaryDirectory() as tmp: