From fcf7d43d81cdc67d7db39b51158cfbcbf6f97901 Mon Sep 17 00:00:00 2001 From: Devil Date: Wed, 30 Sep 2026 22:25:44 -0400 Subject: [PATCH 1/5] fix: resilient audit with a real quorum module - Reviewer failures, timeouts and unparseable output no longer abort the whole run: the reviewer is recorded as not approved with its error. A failed *required* reviewer blocks the run with that error instead of sending Codex a fix request with nothing to fix. - New clodex/quorum.py: unanimous / majority / N-of-required, required reviewers only; replaces the hardcoded claude-plan/codex-architecture fields (agreement now reports required_approved_by / required_pending). - Required fixes come only from blocking required reviewers, are de-duplicated and ordered by severity. - Cap the diff embedded in audit prompts (audit.max_diff_bytes, 200 KB); the hash still covers the full diff. - Fakes can fail, stall or reject specific reviewers; 12 new tests. Co-Authored-By: Claude Sonnet 5.5 --- clodex/config.py | 1 + clodex/prompts.py | 15 +++ clodex/quorum.py | 92 ++++++++++++++++++ clodex/workflow.py | 95 +++++++++--------- tests/support.py | 16 +++ tests/test_audit_correctness.py | 166 ++++++++++++++++++++++++++++++++ 6 files changed, 335 insertions(+), 50 deletions(-) create mode 100644 clodex/quorum.py create mode 100644 tests/test_audit_correctness.py diff --git a/clodex/config.py b/clodex/config.py index 200c395..ea1accb 100644 --- a/clodex/config.py +++ b/clodex/config.py @@ -37,6 +37,7 @@ }, "audit": { "quorum": "unanimous", + "max_diff_bytes": 200000, "personas": ["security", "performance", "portability", "test-gap"], "reviewers": [ {"id": "claude-plan", "backend": "claude", "persona": "plan-adherence", "required": True, "timeout": 600}, diff --git a/clodex/prompts.py b/clodex/prompts.py index 454de3c..0357d2e 100644 --- a/clodex/prompts.py +++ b/clodex/prompts.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import re from typing import Any from .config import ClodexConfig @@ -93,3 +94,17 @@ def fix_prompt(plan: dict[str, Any], findings: list[str]) -> str: When finished, print a concise Markdown report with changed files and tests run. """ + + +def audit_diff_excerpt(diff: str, max_bytes: int) -> str: + """Cap the diff embedded in an audit prompt; the diff hash still covers the whole diff.""" + encoded = diff.encode("utf-8") + if max_bytes <= 0 or len(encoded) <= max_bytes: + return diff + shown = encoded[:max_bytes].decode("utf-8", errors="ignore") + files = list(dict.fromkeys(re.findall(r"^diff --git a/(.+?) b/", diff, flags=re.MULTILINE))) + return ( + f"{shown}\n\n[diff truncated: showing the first {max_bytes} of {len(encoded)} bytes. " + f"Changed files ({len(files)}): {', '.join(files)}. " + "Reject if the omitted part could hide a problem you cannot rule out.]\n" + ) diff --git a/clodex/quorum.py b/clodex/quorum.py new file mode 100644 index 0000000..ccb7c78 --- /dev/null +++ b/clodex/quorum.py @@ -0,0 +1,92 @@ +"""Audit agreement: decide whether the reviewers' verdicts approve a diff.""" + +from __future__ import annotations + +from typing import Any + +SEVERITY_ORDER = {"critical": 0, "high": 1, "medium": 2, "low": 3, "info": 4} + + +def verdict_approved(verdict: dict[str, Any], diff_hash: str) -> bool: + """A verdict approves only if it says so, for exactly this diff, and the reviewer did not fail.""" + return bool(verdict.get("approved")) and verdict.get("diff_hash") == diff_hash and not verdict.get("_error") + + +def parse_quorum(value: Any) -> tuple[str, int]: + """`unanimous`, `majority`, or an integer N (N required approvals).""" + if isinstance(value, bool): + raise ValueError(f"Unsupported audit.quorum: {value!r}") + if isinstance(value, int): + number = value + else: + text = str(value).strip().lower() + if text in {"unanimous", "all"}: + return "unanimous", 0 + if text == "majority": + return "majority", 0 + if not text.isdigit(): + raise ValueError(f"Unsupported audit.quorum: {value!r} (use unanimous, majority or a number)") + number = int(text) + if number < 1: + raise ValueError(f"audit.quorum must be at least 1, got {number}") + return "count", number + + +def quorum_met(approved: int, total: int, quorum: Any) -> bool: + """Whether `approved` of `total` required reviewers satisfies the quorum. No required reviewers never passes.""" + if total == 0: + return False + kind, number = parse_quorum(quorum) + if kind == "unanimous": + return approved == total + if kind == "majority": + return approved * 2 > total + return approved >= min(number, total) + + +def evaluate(verdicts: list[dict[str, Any]], diff_hash: str, attempt: int = 0, quorum: Any = "unanimous") -> dict[str, Any]: + """Combine reviewer verdicts. Only required reviewers count toward the quorum.""" + required = [verdict for verdict in verdicts if verdict.get("_required", True)] + reviewers = { + str(verdict.get("reviewer_id")): { + "approved": verdict_approved(verdict, diff_hash), + "required": bool(verdict.get("_required", True)), + "diff_hash": verdict.get("diff_hash"), + "persona": verdict.get("persona"), + **({"error": verdict["_error"]} if verdict.get("_error") else {}), + } + for verdict in verdicts + } + approved_by = [str(v.get("reviewer_id")) for v in required if verdict_approved(v, diff_hash)] + pending = [str(v.get("reviewer_id")) for v in required if not verdict_approved(v, diff_hash)] + return { + "approved": quorum_met(len(approved_by), len(required), quorum), + "attempt": attempt, + "diff_hash": diff_hash, + "quorum": str(quorum), + "reviewers": reviewers, + "required_approved_by": approved_by, + "required_pending": pending, + } + + +def required_fixes(verdicts: list[dict[str, Any]], diff_hash: str) -> list[str]: + """Fixes asked for by required reviewers that did not approve, de-duplicated, most severe first.""" + fixes: list[str] = [] + findings: list[tuple[int, str]] = [] + for verdict in verdicts: + if not verdict.get("_required", True) or verdict.get("_error") or verdict_approved(verdict, diff_hash): + continue + fixes.extend(str(fix) for fix in verdict.get("required_fixes") or []) + for finding in verdict.get("findings") or []: + if isinstance(finding, dict): + severity = str(finding.get("severity") or "info") + where = finding.get("file") + if where and finding.get("line") is not None: + where = f"{where}:{finding['line']}" + label = f"[{severity}] " + (f"{where}: " if where else "") + findings.append((SEVERITY_ORDER.get(severity, len(SEVERITY_ORDER)), label + str(finding.get("message", "")))) + else: + findings.append((len(SEVERITY_ORDER), str(finding))) + ordered = fixes + [text for _, text in sorted(findings, key=lambda item: item[0])] + return list(dict.fromkeys(ordered)) or ["Resolve audit disagreement and make the diff satisfy the accepted plan."] diff --git a/clodex/workflow.py b/clodex/workflow.py index ec65ecc..1f4bce1 100644 --- a/clodex/workflow.py +++ b/clodex/workflow.py @@ -11,7 +11,8 @@ from .commands import claude_audit_command, claude_plan_command, codex_exec_command, codex_review_command from .config import ClodexConfig, load_config from .jsonutil import AgentEnvelopeError, extract_json_object -from .prompts import audit_prompt, fix_prompt, implementation_prompt, plan_prompt +from .prompts import audit_diff_excerpt, audit_prompt, fix_prompt, implementation_prompt, plan_prompt +from .quorum import evaluate as evaluate_quorum, required_fixes, verdict_approved from .schemas import SchemaValidationError, check as check_schema from .state import StateStore from .trace import TraceWriter @@ -214,6 +215,8 @@ def _audit_loop( include_untracked: bool = False, ) -> WorkflowResult: agreement: dict[str, Any] = {} + block_error: str | None = None + quorum = self.config.audit.get("quorum", "unanimous") for attempt in range(self.config.max_fix_loops + 1): if include_untracked: self._include_untracked(diff_root) @@ -221,7 +224,7 @@ def _audit_loop( diff_hash = hash_text(diff) artifacts.write_text("changes.diff", diff) verdicts = self._run_reviewers(artifacts, plan_json, diff, diff_hash, runner, trace, attempt) - agreement = self._agreement(verdicts, diff_hash, attempt) + agreement = self._agreement(verdicts, diff_hash, attempt, quorum) 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) @@ -229,18 +232,23 @@ def _audit_loop( self.state.update_task(task_id, "done") trace.event("run.complete", {"status": "approved"}) return WorkflowResult("approved", run_id, task_id, str(artifacts.path), agreement) + failed = [v for v in verdicts if v.get("_required", True) and v.get("_error")] + if failed: + # A reviewer that could not run is an infrastructure problem, not something Codex can fix. + block_error = "Required reviewer failed: " + "; ".join(f"{v['reviewer_id']}: {v['_error']}" for v in failed) + break if attempt >= self.config.max_fix_loops: break - findings = self._required_fixes(*verdicts) + findings = self._required_fixes(*verdicts, diff_hash=diff_hash) 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)) 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_run(run_id, "blocked", agreement.get("diff_hash"), error=block_error) self.state.update_task(task_id, "blocked") - trace.event("run.complete", {"status": "blocked"}) - return WorkflowResult("blocked", run_id, task_id, str(artifacts.path), agreement) + trace.event("run.complete", {"status": "blocked", **({"error": block_error} if block_error else {})}) + return WorkflowResult("blocked", run_id, task_id, str(artifacts.path), {**agreement, **({"error": block_error} if block_error else {})}) @staticmethod def _include_untracked(repo_root: Path) -> None: @@ -269,27 +277,42 @@ def _run_reviewers( verdicts: list[dict[str, Any]] = [] first_claude: dict[str, Any] | None = None first_codex: dict[str, Any] | None = None + shown_diff = audit_diff_excerpt(diff, int(self.config.audit.get("max_diff_bytes", 200_000))) 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)) + required = bool(reviewer.get("required", True)) command = claude_audit_command(self.config) if backend == "claude" else codex_review_command(self.config, runner.repo_root) - 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, - ) + try: + verdict = self._run_json_with_retry( + runner, + command, + audit_prompt(backend.title(), plan_json, shown_diff, diff_hash, reviewer_id, persona), + f"{reviewer_id} audit", + trace, + timeout=timeout, + ) + except RuntimeError as exc: + # One flaky or timed-out reviewer must not abort the whole audit. + verdict = { + "approved": False, + "diff_hash": None, + "summary": f"reviewer failed: {exc}", + "findings": [], + "required_fixes": [], + "_error": str(exc), + } + trace.event("audit.reviewer_failed", {"reviewer_id": reviewer_id, "required": required, "error": str(exc)}) verdict.setdefault("reviewer_id", reviewer_id) verdict.setdefault("persona", persona) - verdict["_required"] = bool(reviewer.get("required", True)) + verdict["_required"] = required + approved = verdict_approved(verdict, diff_hash) 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"]}) + self.state.add_audit(artifacts.run_id, reviewer_id, approved, verdict.get("diff_hash"), json.dumps(verdict)) + trace.event("audit.verdict", {"reviewer_id": reviewer_id, "approved": approved, "required": required}) verdicts.append(verdict) if backend == "claude" and first_claude is None: first_claude = verdict @@ -354,43 +377,15 @@ def _cancelled(self, run_id: str, trace: TraceWriter) -> bool: @staticmethod def _approved(audit: dict[str, Any], diff_hash: str) -> bool: - return bool(audit.get("approved")) and audit.get("diff_hash") == diff_hash + return verdict_approved(audit, 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": bool(required) and 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), - } + def _agreement(verdicts: list[dict[str, Any]], diff_hash: str, attempt: int, quorum: Any = "unanimous") -> dict[str, Any]: + return evaluate_quorum(verdicts, diff_hash, attempt, quorum) @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 []: - if isinstance(finding, dict): - fixes.append(str(finding.get("message", finding))) - else: - fixes.append(str(finding)) - return fixes or ["Resolve audit disagreement and make the diff satisfy the accepted plan."] + def _required_fixes(*audits: dict[str, Any], diff_hash: str) -> list[str]: + return required_fixes(list(audits), diff_hash) @staticmethod def _format_agent_report(stdout: str, stderr: str) -> str: diff --git a/tests/support.py b/tests/support.py index 9d2296b..8a3a5c8 100644 --- a/tests/support.py +++ b/tests/support.py @@ -64,12 +64,18 @@ def __init__( envelope_error_once: bool = False, no_structured_output: bool = False, schema_violation_once: bool = False, + fail_reviewers: tuple[str, ...] = (), + slow_reviewers: tuple[str, ...] = (), + reject_reviewers: tuple[str, ...] = (), ): self.reject_once = reject_once self.malformed_once = malformed_once self.envelope_error_once = envelope_error_once self.no_structured_output = no_structured_output self.schema_violation_once = schema_violation_once + self.fail_reviewers = tuple(fail_reviewers) + self.slow_reviewers = tuple(slow_reviewers) + self.reject_reviewers = tuple(reject_reviewers) self.sleep_seconds = sleep_seconds self.include_clodex = include_clodex @@ -156,6 +162,16 @@ def audit_verdict(approved, summary, fixes): reviewer = reviewer_match.group(1).strip() if persona_match: persona = persona_match.group(1).strip() + if reviewer in {self.fail_reviewers!r}: + sys.stderr.write('simulated reviewer crash: ' + reviewer + '\\n') + raise SystemExit(3) + if reviewer in {self.slow_reviewers!r}: + import time + time.sleep(4) + if reviewer in {self.reject_reviewers!r}: + approved = False + summary = 'rejected by ' + reviewer + fixes = ['fix from ' + reviewer] return json.dumps({{'approved': approved, 'diff_hash': h, 'reviewer_id': reviewer, 'persona': persona, 'summary': summary, 'findings': [], 'required_fixes': fixes}}) if name == 'claude': diff --git a/tests/test_audit_correctness.py b/tests/test_audit_correctness.py new file mode 100644 index 0000000..f556300 --- /dev/null +++ b/tests/test_audit_correctness.py @@ -0,0 +1,166 @@ +"""Tests: audit quorum, reviewer resilience and fix collection.""" + +from __future__ import annotations + +import json +import unittest +from pathlib import Path + +from clodex.prompts import audit_diff_excerpt +from clodex.quorum import evaluate, parse_quorum, quorum_met, required_fixes, verdict_approved +from clodex.workflow import ClodexWorkflow +from tests.support import FakeCliPath, TempRepo + +H = "a" * 64 + + +def verdict(reviewer_id: str, approved: bool = True, required: bool = True, **extra) -> dict: + return {"reviewer_id": reviewer_id, "approved": approved, "diff_hash": H, "persona": "p", "_required": required, **extra} + + +class QuorumTests(unittest.TestCase): + def test_parse_quorum(self): + self.assertEqual(parse_quorum("unanimous"), ("unanimous", 0)) + self.assertEqual(parse_quorum(" ALL "), ("unanimous", 0)) + self.assertEqual(parse_quorum("majority"), ("majority", 0)) + self.assertEqual(parse_quorum(2), ("count", 2)) + self.assertEqual(parse_quorum("3"), ("count", 3)) + for bad in ("most", True, 0, "-1", ""): + with self.assertRaises(ValueError, msg=repr(bad)): + parse_quorum(bad) + + def test_quorum_met(self): + self.assertTrue(quorum_met(2, 2, "unanimous")) + self.assertFalse(quorum_met(1, 2, "unanimous")) + self.assertFalse(quorum_met(0, 0, "unanimous"), "no required reviewers must never pass") + self.assertFalse(quorum_met(0, 0, "majority")) + self.assertTrue(quorum_met(2, 3, "majority")) + self.assertFalse(quorum_met(1, 2, "majority")) + self.assertTrue(quorum_met(3, 4, "majority")) + self.assertTrue(quorum_met(2, 3, 2)) + self.assertFalse(quorum_met(1, 3, 2)) + self.assertTrue(quorum_met(2, 2, 5), "N larger than the reviewer count is clamped") + + def test_evaluate_counts_required_reviewers_only_and_exact_diff_hash(self): + verdicts = [ + verdict("a"), + verdict("b", approved=False), + verdict("opt", approved=False, required=False), + verdict("stale", required=False, diff_hash="other"), + ] + result = evaluate(verdicts, H, attempt=1, quorum="unanimous") + self.assertFalse(result["approved"]) + self.assertEqual(result["required_approved_by"], ["a"]) + self.assertEqual(result["required_pending"], ["b"]) + self.assertEqual(result["attempt"], 1) + self.assertFalse(result["reviewers"]["stale"]["approved"]) + self.assertNotIn("claude_approved", result) + self.assertTrue(evaluate([verdict("a"), verdict("opt", approved=False, required=False)], H)["approved"]) + self.assertTrue(evaluate(verdicts[:2], H, quorum="majority")["approved"] is False) + + def test_errored_reviewer_never_approves(self): + broken = verdict("a", _error="exit code 124") + self.assertFalse(verdict_approved(broken, H)) + result = evaluate([broken], H) + self.assertFalse(result["approved"]) + self.assertEqual(result["reviewers"]["a"]["error"], "exit code 124") + + def test_required_fixes_only_from_blocking_reviewers_deduplicated_and_ordered(self): + verdicts = [ + verdict("a", approved=False, required_fixes=["add tests", "add tests"], findings=[ + {"severity": "low", "file": None, "line": None, "message": "nit"}, + {"severity": "critical", "file": "x.py", "line": 3, "message": "crash"}, + ]), + verdict("b", approved=False, required_fixes=["add tests", "update docs"]), + verdict("ok", approved=True, required_fixes=["ignored: approved"]), + verdict("opt", approved=False, required=False, required_fixes=["ignored: optional"]), + verdict("broken", approved=False, _error="boom", required_fixes=["ignored: errored"]), + ] + self.assertEqual( + required_fixes(verdicts, H), + ["add tests", "update docs", "[critical] x.py:3: crash", "[low] nit"], + ) + self.assertEqual(len(required_fixes([verdict("ok")], H)), 1, "falls back to a generic instruction") + + +class DiffExcerptTests(unittest.TestCase): + def test_small_diff_is_unchanged(self): + self.assertEqual(audit_diff_excerpt("diff --git a/x b/x\n+1\n", 1000), "diff --git a/x b/x\n+1\n") + self.assertEqual(audit_diff_excerpt("anything", 0), "anything") + + def test_large_diff_is_truncated_with_file_list(self): + diff = "diff --git a/one.py b/one.py\n" + "+" + "é" * 500 + "\ndiff --git a/two.py b/two.py\n+x\n" + shown = audit_diff_excerpt(diff, 100) + self.assertLess(len(shown.encode("utf-8")), len(diff.encode("utf-8"))) + self.assertIn("diff truncated", shown) + self.assertIn("one.py, two.py", shown) + + +def contract(repo: Path, reviewers: list[dict], quorum: str = "unanimous") -> None: + (repo / "CLODEX.md").write_text( + f"---\nmax_fix_loops: 0\naudit:\n quorum: {quorum}\n reviewers: {json.dumps(reviewers)}\n---\nbody\n", + encoding="utf-8", + ) + + +def reviewer(reviewer_id: str, backend: str, required: bool, timeout: int = 600) -> dict: + return {"id": reviewer_id, "backend": backend, "persona": reviewer_id, "required": required, "timeout": timeout} + + +REQUIRED_PAIR = [reviewer("claude-plan", "claude", True), reviewer("codex-architecture", "codex", True)] + + +class AuditWorkflowTests(unittest.TestCase): + def test_failed_optional_reviewer_does_not_abort_the_audit(self): + with TempRepo() as repo, FakeCliPath(fail_reviewers=("security",)): + contract(repo, [*REQUIRED_PAIR, reviewer("security", "codex", False)]) + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "approved") + security = result.data["reviewers"]["security"] + self.assertFalse(security["approved"]) + self.assertIn("exit code 3", security["error"]) + saved = json.loads((Path(result.artifacts_dir) / "reviewers" / "security.json").read_text(encoding="utf-8")) + self.assertIn("_error", saved) + + def test_timed_out_optional_reviewer_does_not_abort_the_audit(self): + with TempRepo() as repo, FakeCliPath(slow_reviewers=("security",)): + contract(repo, [*REQUIRED_PAIR, reviewer("security", "codex", False, timeout=1)]) + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "approved") + self.assertIn("exit code 124", result.data["reviewers"]["security"]["error"]) + + def test_failed_required_reviewer_blocks_without_a_pointless_fix_attempt(self): + with TempRepo() as repo, FakeCliPath(fail_reviewers=("codex-architecture",)): + contract(repo, REQUIRED_PAIR) + workflow = ClodexWorkflow(repo) + result = workflow.build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "blocked") + self.assertIn("codex-architecture", result.data["error"]) + self.assertFalse(list(Path(result.artifacts_dir).glob("fix-attempt-*"))) + run = workflow.state.get_run(result.run_id) + self.assertEqual(run["status"], "blocked") + self.assertIn("codex-architecture", run["error"]) + + def test_majority_quorum_approves_despite_one_required_rejection(self): + reviewers = [*REQUIRED_PAIR, reviewer("security", "codex", True)] + with TempRepo() as repo, FakeCliPath(reject_reviewers=("security",)): + contract(repo, reviewers, quorum="majority") + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "approved") + self.assertEqual(result.data["required_pending"], ["security"]) + self.assertEqual(sorted(result.data["required_approved_by"]), ["claude-plan", "codex-architecture"]) + with TempRepo() as repo, FakeCliPath(reject_reviewers=("security",)): + contract(repo, reviewers, quorum="unanimous") + self.assertEqual(ClodexWorkflow(repo).build("implement fixture", workspace_backend="local").status, "blocked") + + def test_agreement_has_no_hardcoded_reviewer_fields(self): + with TempRepo() as repo, FakeCliPath(): + contract(repo, [reviewer("alpha", "claude", True), reviewer("beta", "codex", True)]) + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "approved") + self.assertNotIn("claude_approved", result.data) + self.assertEqual(sorted(result.data["required_approved_by"]), ["alpha", "beta"]) + + +if __name__ == "__main__": + unittest.main() From f8698a438b5b15cdad6f25f17392ac462e047c06 Mon Sep 17 00:00:00 2001 From: Devil Date: Wed, 30 Sep 2026 22:30:19 -0400 Subject: [PATCH 2/5] fix: failed runs, worktree cleanup and safe apply - Unexpected errors in plan/build/audit now mark the run and task `failed` (with an error.txt traceback) and remove the run's git worktree, instead of leaving the run `running` forever. Cancelled runs release theirs too. Release only ever touches paths under workspace_root. - New `clodex clean ` removes the kept worktree of a finished run; the artifacts and patch stay. - `clodex apply` now refuses runs that are not approved (or already applied) unless --force, refuses a patch whose hash no longer matches the approved diff, and reports local-workspace runs as already in the tree. - Diffs are read and written byte-exact (no universal-newline decoding), so CRLF and non-UTF-8 content hashes and applies correctly; previously the saved patch could be corrupted on Windows. Co-Authored-By: Claude Sonnet 5.5 --- clodex/artifacts.py | 24 +++-- clodex/cli.py | 12 ++- clodex/prompts.py | 2 +- clodex/state.py | 4 + clodex/workflow.py | 123 +++++++++++++++++++---- clodex/workspace.py | 15 +++ tests/test_run_lifecycle.py | 189 ++++++++++++++++++++++++++++++++++++ 7 files changed, 334 insertions(+), 35 deletions(-) create mode 100644 tests/test_run_lifecycle.py diff --git a/clodex/artifacts.py b/clodex/artifacts.py index 699bf44..ddb93b4 100644 --- a/clodex/artifacts.py +++ b/clodex/artifacts.py @@ -40,25 +40,23 @@ def write_json(self, name: str, data: dict[str, Any]) -> Path: 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: + def write_text(self, name: str, text: str, exact: bool = False) -> Path: + """Write text; `exact=True` keeps bytes as-is (no newline translation), for diffs and patches.""" path = self.path / name path.parent.mkdir(parents=True, exist_ok=True) - path.write_text(text, encoding="utf-8") + if exact: + path.write_bytes(text.encode("utf-8", errors="surrogateescape")) + else: + path.write_text(text, encoding="utf-8") return path def current_diff(repo_root: Path) -> str: - result = subprocess.run( - ["git", "diff", "--binary", "HEAD"], - cwd=repo_root, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - check=False, - ) - return result.stdout + # Read bytes, not text: universal-newline decoding would rewrite CRLF content and + # the diff would no longer apply to (or hash like) the real files. + result = subprocess.run(["git", "diff", "--binary", "HEAD"], cwd=repo_root, capture_output=True, check=False) + return result.stdout.decode("utf-8", errors="surrogateescape") def hash_text(text: str) -> str: - return hashlib.sha256(text.encode("utf-8")).hexdigest() + return hashlib.sha256(text.encode("utf-8", errors="surrogateescape")).hexdigest() diff --git a/clodex/cli.py b/clodex/cli.py index ad7b656..b00df21 100644 --- a/clodex/cli.py +++ b/clodex/cli.py @@ -89,6 +89,10 @@ def main(argv: list[str] | None = None) -> int: 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") + apply_cmd.add_argument("--force", action="store_true", help="Apply even if the run is not approved or its patch changed") + + clean_cmd = sub.add_parser("clean", help="Remove the kept git worktree of a finished run") + clean_cmd.add_argument("run_id") trace = sub.add_parser("trace", help="Inspect run traces") trace_sub = trace.add_subparsers(dest="trace_command", required=True) @@ -187,9 +191,13 @@ def main(argv: list[str] | None = None) -> int: 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) + result = workflow.apply_run(args.run_id, check=args.check, force=args.force) + print_result(result.__dict__, args.json) + return 0 if result.status not in {"apply-failed", "apply-check-failed", "apply-refused"} else 1 + if args.command == "clean": + result = workflow.clean_run(args.run_id) print_result(result.__dict__, args.json) - return 0 if result.status not in {"apply-failed", "apply-check-failed"} else 1 + return 0 if result.status != "clean-refused" else 1 if args.command == "trace": return handle_trace(args, workflow, args.json) if args.command == "eval": diff --git a/clodex/prompts.py b/clodex/prompts.py index 0357d2e..fdf1cdd 100644 --- a/clodex/prompts.py +++ b/clodex/prompts.py @@ -98,7 +98,7 @@ def fix_prompt(plan: dict[str, Any], findings: list[str]) -> str: def audit_diff_excerpt(diff: str, max_bytes: int) -> str: """Cap the diff embedded in an audit prompt; the diff hash still covers the whole diff.""" - encoded = diff.encode("utf-8") + encoded = diff.encode("utf-8", errors="replace") if max_bytes <= 0 or len(encoded) <= max_bytes: return diff shown = encoded[:max_bytes].decode("utf-8", errors="ignore") diff --git a/clodex/state.py b/clodex/state.py index 5d66c35..50d8a15 100644 --- a/clodex/state.py +++ b/clodex/state.py @@ -499,6 +499,10 @@ def add_workspace_lock(self, run_id: str, source_path: str, workspace_path: str, (run_id, source_path, workspace_path, backend, now_iso()), ) + 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)) + def request_cancel(self, run_id: str) -> None: with self.session() as con: con.execute( diff --git a/clodex/workflow.py b/clodex/workflow.py index 1f4bce1..a90e78f 100644 --- a/clodex/workflow.py +++ b/clodex/workflow.py @@ -2,6 +2,8 @@ import json import subprocess +import traceback +from collections.abc import Callable from dataclasses import dataclass from pathlib import Path from typing import Any @@ -14,7 +16,7 @@ from .prompts import audit_diff_excerpt, audit_prompt, fix_prompt, implementation_prompt, plan_prompt from .quorum import evaluate as evaluate_quorum, required_fixes, verdict_approved from .schemas import SchemaValidationError, check as check_schema -from .state import StateStore +from .state import TERMINAL_STATUSES, StateStore from .trace import TraceWriter from .workspace import DirtyWorkspaceError, WorkspaceManager, WorkspaceRef @@ -60,14 +62,18 @@ def plan(self, task: str, dry_run: bool = False) -> WorkflowResult: self.state.upsert_task(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(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 action() -> WorkflowResult: + prompt = plan_prompt(self.config, task) + 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) + + return self._guarded(run_id, task_id, artifacts, trace, action) def build( self, @@ -123,28 +129,51 @@ def audit(self, dry_run: bool = False) -> WorkflowResult: 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, AgentRunner(self.repo_root), self.repo_root, trace, None) + return self._guarded( + run_id, + task_id, + artifacts, + trace, + lambda: self._audit_loop(task_id, run_id, artifacts, plan_json, AgentRunner(self.repo_root), self.repo_root, trace, None), + ) - def apply_run(self, run_id: str | None, check: bool = False) -> WorkflowResult: + def apply_run(self, run_id: str | None, check: bool = False, force: 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}") + task_id = run.get("task_id") artifacts_dir = Path(str(run.get("artifacts_dir") or self.config.runs_root / run_id)) + status = str(run["status"]) + if status != "approved" and not force: + reason = "already applied" if status == "applied" else f"run is {status}; only approved runs can be applied (use --force to override)" + return WorkflowResult("apply-refused", run_id, task_id, str(artifacts_dir), {"error": reason}) + + workspace_path = run.get("workspace_path") + if workspace_path and Path(str(workspace_path)).resolve() == self.repo_root.resolve(): + # Local workspace: the agents already edited this working tree; there is no patch to apply. + note = "local workspace run: the changes are already in the working tree" + if not check: + self.state.update_run(run_id, "applied") + return WorkflowResult("apply-check" if check else "applied", run_id, task_id, str(artifacts_dir), {"note": note}) + 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}") + approved_hash = run.get("diff_hash") + if status == "approved" and approved_hash and hash_text(patch.read_bytes().decode("utf-8", errors="surrogateescape")) != approved_hash and not force: + return WorkflowResult("apply-refused", run_id, task_id, str(artifacts_dir), {"error": f"{patch.name} does not match the approved diff hash"}) 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}) + return WorkflowResult("apply-check-failed" if check else "apply-failed", run_id, 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)}) + return WorkflowResult("apply-check", run_id, 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)}) + return WorkflowResult("applied", run_id, task_id, str(artifacts_dir), {"patch": str(patch)}) def _execute_build( self, @@ -158,6 +187,25 @@ def _execute_build( ) -> 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}) + return self._guarded( + run_id, + task_id, + artifacts, + trace, + lambda: self._build_steps(task, task_id, run_id, artifacts, workspace_backend, approval_profile, apply_changes, trace), + ) + + def _build_steps( + self, + task: str, + task_id: str, + run_id: str, + artifacts: ArtifactStore, + workspace_backend: str | None, + approval_profile: str | None, + apply_changes: bool, + trace: TraceWriter, + ) -> WorkflowResult: try: workspace = WorkspaceManager(self.repo_root, self.config).prepare(run_id, workspace_backend) except DirtyWorkspaceError as exc: @@ -172,9 +220,7 @@ def _execute_build( 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", {}) + if self._cancelled(run_id, trace): 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) @@ -196,7 +242,7 @@ def _execute_build( 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)) + patch = artifacts.write_text("apply.patch", current_diff(workspace.path), exact=True) result.data["apply_patch"] = str(patch) if apply_changes: return self.apply_run(run_id) @@ -222,7 +268,7 @@ def _audit_loop( self._include_untracked(diff_root) diff = current_diff(diff_root) diff_hash = hash_text(diff) - artifacts.write_text("changes.diff", diff) + artifacts.write_text("changes.diff", diff, exact=True) verdicts = self._run_reviewers(artifacts, plan_json, diff, diff_hash, runner, trace, attempt) agreement = self._agreement(verdicts, diff_hash, attempt, quorum) artifacts.write_json("05-agreement.json", agreement) @@ -372,9 +418,48 @@ 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", {}) + self._release_workspace(run_id) return True return False + def _guarded(self, run_id: str, task_id: str, artifacts: ArtifactStore, trace: TraceWriter, action: Callable[[], WorkflowResult]) -> WorkflowResult: + """Run `action`; an unexpected error marks the run failed instead of leaving it `running`.""" + try: + return action() + except Exception as exc: # noqa: BLE001 - any failure must be recorded on the run + error = f"{type(exc).__name__}: {exc}" + released = False + try: + artifacts.write_text("error.txt", traceback.format_exc()) + self.state.update_run(run_id, "failed", error=error) + self.state.update_task(task_id, "failed") + trace.event("run.failed", {"error": error}) + released = self._release_workspace(run_id) + except Exception: # noqa: BLE001 - bookkeeping must not mask the original error + pass + return WorkflowResult("failed", run_id, task_id, str(artifacts.path), {"error": error, "workspace_released": released}) + + def _release_workspace(self, run_id: str) -> bool: + run = self.state.get_run(run_id) + path = (run or {}).get("workspace_path") + released = bool(path) and WorkspaceManager(self.repo_root, self.config).release(str(path)) + if released: + self.state.release_workspace_lock(run_id) + return released + + def clean_run(self, run_id: str | None) -> WorkflowResult: + """Remove the kept git worktree of a finished run; its artifacts and patch stay.""" + 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}") + status = str(run["status"]) + if status not in TERMINAL_STATUSES: + return WorkflowResult("clean-refused", run_id, run.get("task_id"), run.get("artifacts_dir"), {"error": f"run is still {status}; cancel it first"}) + released = self._release_workspace(run_id) + return WorkflowResult("cleaned" if released else "nothing-to-clean", run_id, run.get("task_id"), run.get("artifacts_dir"), {"workspace_path": run.get("workspace_path")}) + @staticmethod def _approved(audit: dict[str, Any], diff_hash: str) -> bool: return verdict_approved(audit, diff_hash) diff --git a/clodex/workspace.py b/clodex/workspace.py index fdc273a..65d12d2 100644 --- a/clodex/workspace.py +++ b/clodex/workspace.py @@ -47,6 +47,21 @@ def prepare(self, run_id: str, backend: str | None = None) -> WorkspaceRef: 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 release(self, path: Path | str) -> bool: + """Remove a Clodex-created git worktree. Never touches the source repo or paths outside workspace_root.""" + target = Path(path).resolve() + if target == self.repo_root: + return False + try: + target.relative_to(self.config.workspace_root.resolve()) + except ValueError: + return False + if not target.exists(): + return False + subprocess.run(["git", "worktree", "remove", "--force", str(target)], cwd=self.repo_root, capture_output=True, text=True, check=False) + subprocess.run(["git", "worktree", "prune"], cwd=self.repo_root, capture_output=True, text=True, check=False) + return not target.exists() + 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") diff --git a/tests/test_run_lifecycle.py b/tests/test_run_lifecycle.py new file mode 100644 index 0000000..56f5a6a --- /dev/null +++ b/tests/test_run_lifecycle.py @@ -0,0 +1,189 @@ +"""Tests: run failure handling, worktree cleanup and safe apply.""" + +from __future__ import annotations + +import hashlib +import json +import subprocess +import unittest +from pathlib import Path +from unittest import mock + +from clodex.artifacts import ArtifactStore, current_diff, hash_text +from clodex.config import load_config +from clodex.workflow import ClodexWorkflow +from tests.support import FakeCliPath, TempRepo + + +def git(repo: Path, *args: str) -> str: + return subprocess.run(["git", *args], cwd=repo, capture_output=True, text=True, check=True).stdout + + +def worktrees(repo: Path) -> list[str]: + return [line for line in git(repo, "worktree", "list", "--porcelain").splitlines() if line.startswith("worktree ")] + + +def lock_released(workflow: ClodexWorkflow, run_id: str): + with workflow.state.session() as con: + row = con.execute("select released_at from workspace_locks where run_id=?", (run_id,)).fetchone() + return row["released_at"] if row else "no-lock" + + +class FailureHandlingTests(unittest.TestCase): + def test_unexpected_error_marks_build_failed_and_removes_its_worktree(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + with mock.patch.object(ClodexWorkflow, "_run_json_with_retry", side_effect=RuntimeError("boom")): + result = workflow.build("implement fixture", workspace_backend="git-worktree") + self.assertEqual(result.status, "failed") + self.assertIn("boom", result.data["error"]) + self.assertTrue(result.data["workspace_released"]) + run = workflow.state.get_run(result.run_id) + self.assertEqual(run["status"], "failed") + self.assertIn("boom", run["error"]) + self.assertFalse(Path(run["workspace_path"]).exists()) + self.assertEqual(len(worktrees(repo)), 1, "only the main checkout should remain") + self.assertIsNotNone(lock_released(workflow, result.run_id)) + self.assertIn("RuntimeError", (Path(result.artifacts_dir) / "error.txt").read_text(encoding="utf-8")) + task = next(t for t in workflow.state.list_tasks() if t["id"] == result.task_id) + self.assertEqual(task["status"], "failed") + + def test_unexpected_error_marks_plan_failed(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + with mock.patch.object(ClodexWorkflow, "_run_json_with_retry", side_effect=RuntimeError("no plan")): + result = workflow.plan("plan fixture") + self.assertEqual(result.status, "failed") + self.assertEqual(workflow.state.get_run(result.run_id)["status"], "failed") + + def test_failed_local_build_never_removes_the_source_checkout(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + with mock.patch.object(ClodexWorkflow, "_run_json_with_retry", side_effect=RuntimeError("boom")): + result = workflow.build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "failed") + self.assertFalse(result.data["workspace_released"]) + self.assertTrue((repo / "seed.txt").exists()) + + +class CleanTests(unittest.TestCase): + def test_clean_removes_finished_worktree_but_keeps_artifacts(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + built = workflow.build("implement fixture", workspace_backend="git-worktree") + self.assertEqual(built.status, "approved") + workspace = Path(built.data["workspace"]["path"]) + self.assertTrue(workspace.exists()) + cleaned = workflow.clean_run(built.run_id) + self.assertEqual(cleaned.status, "cleaned") + self.assertFalse(workspace.exists()) + self.assertEqual(len(worktrees(repo)), 1) + self.assertTrue((Path(built.artifacts_dir) / "apply.patch").exists()) + self.assertEqual(workflow.clean_run(built.run_id).status, "nothing-to-clean") + self.assertEqual(workflow.apply_run(built.run_id, check=True).status, "apply-check", "the patch outlives the worktree") + + def test_clean_refuses_runs_that_are_still_active(self): + with TempRepo() as repo: + workflow = ClodexWorkflow(repo) + workflow.state.upsert_task("t", "t", "running") + workflow.state.create_run("r1", "t", "p", "running") + self.assertEqual(workflow.clean_run("r1").status, "clean-refused") + with self.assertRaises(ValueError): + workflow.clean_run("nope") + + def test_release_only_removes_paths_under_workspace_root(self): + with TempRepo() as repo: + from clodex.workspace import WorkspaceManager + + manager = WorkspaceManager(repo, load_config(repo)) + self.assertFalse(manager.release(repo)) + self.assertFalse(manager.release(repo / "seed.txt")) + self.assertTrue((repo / "seed.txt").exists()) + + +class ApplySafetyTests(unittest.TestCase): + def test_approved_worktree_run_applies_once(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + built = workflow.build("implement fixture", workspace_backend="git-worktree") + self.assertEqual(workflow.apply_run(built.run_id, check=True).status, "apply-check") + self.assertFalse((repo / "implemented.txt").exists(), "--check must not modify the tree") + applied = workflow.apply_run(built.run_id) + self.assertEqual(applied.status, "applied") + self.assertTrue((repo / "implemented.txt").exists()) + again = workflow.apply_run(built.run_id) + self.assertEqual(again.status, "apply-refused") + self.assertIn("already applied", again.data["error"]) + + def test_blocked_run_is_refused_unless_forced(self): + with TempRepo() as repo, FakeCliPath(reject_reviewers=("codex-architecture",)): + (repo / "CLODEX.md").write_text("---\nmax_fix_loops: 0\n---\nbody\n", encoding="utf-8") + git(repo, "add", "CLODEX.md") + git(repo, "commit", "-m", "no fix loops") + workflow = ClodexWorkflow(repo) + built = workflow.build("implement fixture", workspace_backend="git-worktree") + self.assertEqual(built.status, "blocked") + refused = workflow.apply_run(built.run_id) + self.assertEqual(refused.status, "apply-refused") + self.assertIn("blocked", refused.data["error"]) + self.assertFalse((repo / "implemented.txt").exists()) + self.assertEqual(workflow.apply_run(built.run_id, force=True).status, "applied") + self.assertTrue((repo / "implemented.txt").exists()) + + def test_patch_that_no_longer_matches_the_approved_hash_is_refused(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + built = workflow.build("implement fixture", workspace_backend="git-worktree") + patch = Path(built.artifacts_dir) / "apply.patch" + patch.write_bytes(patch.read_bytes().replace(b"implemented", b"tampered!!!")) + refused = workflow.apply_run(built.run_id) + self.assertEqual(refused.status, "apply-refused") + self.assertIn("approved diff hash", refused.data["error"]) + self.assertFalse((repo / "implemented.txt").exists()) + + def test_local_workspace_run_reports_changes_already_in_the_tree(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + built = workflow.build("implement fixture", workspace_backend="local") + self.assertEqual(built.status, "approved") + result = workflow.apply_run(built.run_id) + self.assertEqual(result.status, "applied") + self.assertIn("already in the working tree", result.data["note"]) + self.assertEqual(workflow.state.get_run(built.run_id)["status"], "applied") + + +class ByteExactDiffTests(unittest.TestCase): + def test_crlf_content_round_trips_through_diff_artifact_and_hash(self): + with TempRepo() as repo: + git(repo, "config", "core.autocrlf", "false") + (repo / "crlf.txt").write_bytes(b"a\r\nb\r\n") + git(repo, "add", "crlf.txt") + git(repo, "commit", "-m", "crlf") + (repo / "crlf.txt").write_bytes(b"a\r\nb\r\nc\r\n") + raw = subprocess.run(["git", "diff", "--binary", "HEAD"], cwd=repo, capture_output=True, check=True).stdout + diff = current_diff(repo) + self.assertIn("+c\r\n", diff) + self.assertEqual(hash_text(diff), hashlib.sha256(raw).hexdigest()) + saved = ArtifactStore(load_config(repo), "run-x").write_text("changes.diff", diff, exact=True) + self.assertEqual(saved.read_bytes(), raw) + # Reset everything: the fixture's seed files were written under the host's autocrlf setting. + git(repo, "checkout", "--", ".") + applied =subprocess.run(["git", "apply", "--check", str(saved)], cwd=repo, capture_output=True, text=True) + self.assertEqual(applied.returncode, 0, applied.stderr) + + def test_non_utf8_bytes_survive_the_diff_round_trip(self): + with TempRepo() as repo: + git(repo, "config", "core.autocrlf", "false") + (repo / "latin.txt").write_bytes(b"caf\xe9\n") + git(repo, "add", "latin.txt") + git(repo, "commit", "-m", "latin1") + (repo / "latin.txt").write_bytes(b"caf\xe9 au lait\n") + raw = subprocess.run(["git", "diff", "--binary", "HEAD"], cwd=repo, capture_output=True, check=True).stdout + diff = current_diff(repo) + self.assertEqual(hash_text(diff), hashlib.sha256(raw).hexdigest()) + saved = ArtifactStore(load_config(repo), "run-y").write_text("changes.diff", diff, exact=True) + self.assertEqual(saved.read_bytes(), raw) + + +if __name__ == "__main__": + unittest.main() From 7fc8cd87601f66debb94c04185c602340e5a0b49 Mon Sep 17 00:00:00 2001 From: Devil Date: Wed, 30 Sep 2026 22:33:23 -0400 Subject: [PATCH 3/5] fix: state-store integrity - Finished runs stay finished: update_run rejects any move out of a terminal status (except to `applied`, so approved runs and `apply --force` still work). A late worker can no longer overwrite a cancellation. - Cancel no longer overwrites approved/applied/completed runs, and only records completion when something was actually cancelled. - update_handoff validates phase (planning/implementation/audit/fix/ decision/done) and status instead of accepting any string. - ArtifactStore records every artifact in the ledger (one row per name); the artifacts table was previously never populated outside tests. - release_workspace_lock sets released_at once. Co-Authored-By: Claude Sonnet 5.5 --- clodex/artifacts.py | 10 ++- clodex/state.py | 34 ++++++++- clodex/tasks.py | 2 +- clodex/workflow.py | 8 +- tests/test_state_integrity.py | 134 ++++++++++++++++++++++++++++++++++ 5 files changed, 179 insertions(+), 9 deletions(-) create mode 100644 tests/test_state_integrity.py diff --git a/clodex/artifacts.py b/clodex/artifacts.py index ddb93b4..af88303 100644 --- a/clodex/artifacts.py +++ b/clodex/artifacts.py @@ -28,16 +28,23 @@ def make_run_id(task_id: str) -> str: class ArtifactStore: - def __init__(self, config: ClodexConfig, run_id: str): + def __init__(self, config: ClodexConfig, run_id: str, state: Any = None): self.config = config self.run_id = run_id + self.state = state self.path = config.runs_root / run_id self.path.mkdir(parents=True, exist_ok=True) + def _record(self, name: str, path: Path) -> None: + """Index the file in the state ledger so `handoff_get` and exports can find it.""" + if self.state is not None: + self.state.add_artifact(self.run_id, name, str(path), path.suffix.lstrip(".") or "file") + 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") + self._record(name, path) return path def write_text(self, name: str, text: str, exact: bool = False) -> Path: @@ -48,6 +55,7 @@ def write_text(self, name: str, text: str, exact: bool = False) -> Path: path.write_bytes(text.encode("utf-8", errors="surrogateescape")) else: path.write_text(text, encoding="utf-8") + self._record(name, path) return path diff --git a/clodex/state.py b/clodex/state.py index 50d8a15..262f6b8 100644 --- a/clodex/state.py +++ b/clodex/state.py @@ -10,6 +10,12 @@ TERMINAL_STATUSES = {"approved", "blocked", "failed", "cancelled", "applied", "completed"} HANDOFF_REASON_STATUSES = {"blocked", "failed"} +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 +# normally; any other finished run via `apply --force`). This also stops a late worker +# from overwriting a cancellation. +TERMINAL_EXITS: dict[str, set[str]] = {status: {"applied"} for status in ("approved", "blocked", "failed", "cancelled")} HANDOFF_REJECTED_STATUSES = TERMINAL_STATUSES - HANDOFF_REASON_STATUSES @@ -220,6 +226,12 @@ def update_run( timestamp = now_iso() completed = timestamp if status in TERMINAL_STATUSES else None with self.session() as con: + con.execute("begin immediate") + row = con.execute("select status from runs where id=?", (run_id,)).fetchone() + if row is not None: + current = str(row["status"]) + if current in TERMINAL_STATUSES and status != current and status not in TERMINAL_EXITS.get(current, set()): + raise ValueError(f"illegal run transition: {current} -> {status} ({run_id})") con.execute( """ update runs set @@ -290,6 +302,10 @@ def update_handoff( requested_reason = _reason_text(blocked_reason) or _report_reason(report) if status in HANDOFF_REJECTED_STATUSES: raise ValueError(f"handoff status cannot be set via update_handoff: {status}") + if status is not None and status not in HANDOFF_ACTIVE_STATUSES: + raise ValueError(f"unknown handoff status: {status} (use one of {sorted(HANDOFF_ACTIVE_STATUSES)})") + if phase is not None and phase not in HANDOFF_PHASES: + raise ValueError(f"unknown handoff phase: {phase} (use one of {sorted(HANDOFF_PHASES)})") with self.session() as con: con.execute("begin immediate") @@ -480,6 +496,8 @@ def _insert_event(self, con: sqlite3.Connection, run_id: str, event: str, data: def add_artifact(self, run_id: str, name: str, path: str, kind: str) -> None: with self.session() as con: + # Files like 05-agreement.json are rewritten every attempt; keep one row per name. + con.execute("delete from artifacts where run_id=? and name=?", (run_id, name)) con.execute( "insert into artifacts(run_id, name, path, kind, created_at) values (?, ?, ?, ?, ?)", (run_id, name, path, kind, now_iso()), @@ -513,7 +531,11 @@ def request_cancel(self, run_id: str) -> None: """, (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)) + marks = ",".join("?" for _ in TERMINAL_STATUSES) + con.execute( + f"update runs set status=?, updated_at=? where id=? and status not in ({marks})", + ("cancel_requested", now_iso(), run_id, *sorted(TERMINAL_STATUSES)), + ) def cancellation_requested(self, run_id: str) -> bool: with self.session() as con: @@ -521,10 +543,16 @@ def cancellation_requested(self, run_id: str) -> bool: return bool(row and row["requested"]) def complete_cancel(self, run_id: str) -> None: + """Mark an unfinished run cancelled. A run that already finished keeps its status.""" timestamp = now_iso() + marks = ",".join("?" for _ in TERMINAL_STATUSES) with self.session() as con: - con.execute("update cancellations set completed_at=? where run_id=?", (timestamp, run_id)) - con.execute("update runs set status=?, completed_at=?, updated_at=? where id=?", ("cancelled", timestamp, timestamp, run_id)) + cursor = con.execute( + f"update runs set status=?, completed_at=?, updated_at=? where id=? and status not in ({marks})", + ("cancelled", timestamp, timestamp, run_id, *sorted(TERMINAL_STATUSES)), + ) + if cursor.rowcount: + con.execute("update cancellations set completed_at=? where run_id=?", (timestamp, run_id)) def list_tasks(self) -> list[dict[str, Any]]: with self.session() as con: diff --git a/clodex/tasks.py b/clodex/tasks.py index 9358cbc..b88b632 100644 --- a/clodex/tasks.py +++ b/clodex/tasks.py @@ -46,7 +46,7 @@ def start( {"workspace": selected_workspace, "approval_profile": selected_profile}, ) - artifacts = ArtifactStore(self.config, run_id) + artifacts = ArtifactStore(self.config, run_id, self.state) 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") diff --git a/clodex/workflow.py b/clodex/workflow.py index a90e78f..ae94d2e 100644 --- a/clodex/workflow.py +++ b/clodex/workflow.py @@ -57,7 +57,7 @@ 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) + artifacts = ArtifactStore(self.config, run_id, self.state) 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", artifacts_dir=str(artifacts.path)) @@ -99,7 +99,7 @@ def build( task_id = make_task_id(task) run_id = make_run_id(task_id) - artifacts = ArtifactStore(self.config, run_id) + artifacts = ArtifactStore(self.config, run_id, self.state) self.state.upsert_task(task_id, task, "running") 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) @@ -114,7 +114,7 @@ def run_existing( run = self.state.get_run(run_id) if run is None: raise ValueError(f"Unknown run: {run_id}") - artifacts = ArtifactStore(self.config, run_id) + artifacts = ArtifactStore(self.config, run_id, self.state) 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: @@ -123,7 +123,7 @@ def audit(self, dry_run: bool = False) -> WorkflowResult: task = "Audit current uncommitted changes" task_id = make_task_id(task) run_id = make_run_id(task_id) - artifacts = ArtifactStore(self.config, run_id) + artifacts = ArtifactStore(self.config, run_id, self.state) 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", workspace_path=str(self.repo_root), artifacts_dir=str(artifacts.path)) diff --git a/tests/test_state_integrity.py b/tests/test_state_integrity.py new file mode 100644 index 0000000..ef70e56 --- /dev/null +++ b/tests/test_state_integrity.py @@ -0,0 +1,134 @@ +"""Tests: state-store integrity (terminal states, cancellation, validation, artifact ledger).""" + +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from clodex.state import HANDOFF_PHASES, TERMINAL_STATUSES, StateStore +from clodex.workflow import ClodexWorkflow +from tests.support import FakeCliPath, TempRepo + + +class StateCase(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.store = StateStore(Path(self.tmp.name) / "state.sqlite3") + self.store.upsert_task("t", "task", "running") + + def make_run(self, run_id: str, status: str = "running") -> str: + self.store.create_run(run_id, "t", "prompt", status) + return run_id + + def status(self, run_id: str) -> str: + return str(self.store.get_run(run_id)["status"]) + + +class TerminalStateTests(StateCase): + def test_finished_runs_cannot_move_to_another_status(self): + for terminal in sorted(TERMINAL_STATUSES - {"applied", "completed"}): + run = self.make_run(f"r-{terminal}", terminal) + with self.assertRaises(ValueError, msg=terminal): + self.store.update_run(run, "running") + with self.assertRaises(ValueError, msg=terminal): + self.store.update_run(run, "approved" if terminal != "approved" else "blocked") + self.assertEqual(self.status(run), terminal) + + def test_applied_and_completed_are_final(self): + for terminal in ("applied", "completed"): + run = self.make_run(f"r-{terminal}", terminal) + with self.assertRaises(ValueError): + self.store.update_run(run, "approved") + + def test_same_status_update_is_allowed_and_finished_runs_can_be_applied(self): + run = self.make_run("r1", "blocked") + self.store.update_run(run, "blocked", error="more detail") + self.assertEqual(self.store.get_run(run)["error"], "more detail") + self.store.update_run(run, "applied") + self.assertEqual(self.status(run), "applied") + + def test_unfinished_runs_move_freely(self): + run = self.make_run("r1", "queued") + for status in ("running", "needs-fix", "cancel_requested", "approved"): + self.store.update_run(run, status) + self.assertEqual(self.status(run), status) + + def test_late_worker_cannot_overwrite_a_cancellation(self): + run = self.make_run("r1") + self.store.request_cancel(run) + self.store.complete_cancel(run) + self.assertEqual(self.status(run), "cancelled") + with self.assertRaises(ValueError): + self.store.update_run(run, "approved", diff_hash="abc") + self.assertEqual(self.status(run), "cancelled") + self.assertIsNone(self.store.get_run(run)["diff_hash"]) + + +class CancelTests(StateCase): + def test_cancel_of_a_running_run(self): + run = self.make_run("r1") + self.store.request_cancel(run) + self.assertEqual(self.status(run), "cancel_requested") + self.assertTrue(self.store.cancellation_requested(run)) + self.store.complete_cancel(run) + self.assertEqual(self.status(run), "cancelled") + + def test_cancel_never_overwrites_any_finished_status(self): + for terminal in sorted(TERMINAL_STATUSES): + run = self.make_run(f"r-{terminal}", terminal) + self.store.request_cancel(run) + self.store.complete_cancel(run) + self.assertEqual(self.status(run), terminal, terminal) + with self.store.session() as con: + row = con.execute("select completed_at from cancellations where run_id=?", (run,)).fetchone() + self.assertIsNone(row["completed_at"], f"{terminal}: nothing was cancelled") + + +class HandoffValidationTests(StateCase): + def test_unknown_phase_and_status_are_rejected_known_ones_accepted(self): + self.store.create_handoff("h1", "t", "prompt", "claude") + for phase in sorted(HANDOFF_PHASES): + self.assertEqual(self.store.update_handoff("h1", phase=phase)["phase"], phase) + with self.assertRaisesRegex(ValueError, "unknown handoff phase"): + self.store.update_handoff("h1", phase="vibing") + with self.assertRaisesRegex(ValueError, "unknown handoff status"): + self.store.update_handoff("h1", status="running") + self.assertEqual(self.store.update_handoff("h1", status="handoff")["status"], "handoff") + self.assertEqual(self.store.get_run("h1")["phase"], sorted(HANDOFF_PHASES)[-1], "rejected updates must not change the phase") + + +class WorkspaceLockTests(StateCase): + def test_release_sets_released_at_once(self): + self.make_run("r1") + self.store.add_workspace_lock("r1", "/src", "/ws", "git-worktree") + self.store.release_workspace_lock("r1") + with self.store.session() as con: + first = con.execute("select released_at from workspace_locks where run_id='r1'").fetchone()["released_at"] + self.assertIsNotNone(first) + self.store.release_workspace_lock("r1") + with self.store.session() as con: + self.assertEqual(con.execute("select released_at from workspace_locks where run_id='r1'").fetchone()["released_at"], first) + + +class ArtifactLedgerTests(unittest.TestCase): + def test_build_records_artifact_rows_without_duplicates(self): + with TempRepo() as repo, FakeCliPath(): + workflow = ClodexWorkflow(repo) + result = workflow.build("implement fixture", workspace_backend="git-worktree") + self.assertEqual(result.status, "approved") + with workflow.state.session() as con: + rows = [dict(r) for r in con.execute("select name, path, kind from artifacts where run_id=?", (result.run_id,))] + names = [row["name"] for row in rows] + for expected in ("01-claude-plan.json", "02-codex-implementation.md", "05-agreement.json", "changes.diff", "apply.patch"): + self.assertIn(expected, names) + self.assertEqual(len(names), len(set(names)), "rewritten artifacts must not duplicate rows") + kinds = {row["name"]: row["kind"] for row in rows} + self.assertEqual(kinds["01-claude-plan.json"], "json") + self.assertEqual(kinds["apply.patch"], "patch") + self.assertTrue(all(Path(row["path"]).exists() for row in rows)) + + +if __name__ == "__main__": + unittest.main() From 26b1aa663049f52a0e49d96a0a67abe6fd1bfc0c Mon Sep 17 00:00:00 2001 From: Devil Date: Wed, 30 Sep 2026 22:38:54 -0400 Subject: [PATCH 4/5] fix: stop worker and agent process trees; supervise task workers - New clodex/procs.py: start children in their own process group/session, stop a whole tree (killpg / taskkill /T), and check liveness without os.kill (on Windows that calls TerminateProcess, even for signal 0). - AgentRunner runs agents in their own group and stops the whole tree on timeout; before, a timeout left the agent (behind the .cmd shim) running and holding the pipes and working directory. - TaskManager.cancel stops the worker and the agents it started, releases the run's worktree, and no longer touches runs that already finished. It only signals a worker with a recent heartbeat, so a reused pid is never killed. - Workers write a heartbeat; get/list notice a worker that died without finishing and mark the run failed instead of leaving it queued forever (or complete a pending cancel). - Fixes the intermittent Windows failure of the cancel lifecycle test. Co-Authored-By: Claude Sonnet 5.5 --- clodex/agents.py | 54 ++++----- clodex/cli.py | 5 +- clodex/procs.py | 87 +++++++++++++++ clodex/state.py | 10 ++ clodex/tasks.py | 139 ++++++++++++++++++++--- tests/support.py | 2 + tests/test_process_control.py | 204 ++++++++++++++++++++++++++++++++++ 7 files changed, 458 insertions(+), 43 deletions(-) create mode 100644 clodex/procs.py create mode 100644 tests/test_process_control.py diff --git a/clodex/agents.py b/clodex/agents.py index 39919c0..3e61704 100644 --- a/clodex/agents.py +++ b/clodex/agents.py @@ -8,6 +8,7 @@ from pathlib import Path from .commands import AgentCommand +from .procs import kill_tree, popen_isolation_kwargs @dataclass @@ -47,31 +48,42 @@ def run(self, command: AgentCommand, prompt: str, timeout: int | None = None) -> at = len(argv) - 1 if argv and argv[-1] == "-" else len(argv) argv[at:at] = ["-o", last_message_path] try: + # Own process group, so a timeout (or cancel) can stop the agent *and* whatever it spawned. + process = subprocess.Popen( + argv, + cwd=self.repo_root, + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + text=True, + encoding="utf-8", + errors="replace", + **popen_isolation_kwargs(), + ) try: - result = subprocess.run( - argv, - cwd=self.repo_root, - input=prompt, - capture_output=True, - text=True, - encoding="utf-8", - errors="replace", - check=False, - timeout=timeout, - ) - except subprocess.TimeoutExpired as exc: + stdout, stderr = process.communicate(input=prompt, timeout=timeout) + except subprocess.TimeoutExpired: + kill_tree(process.pid) + try: + stdout, stderr = process.communicate(timeout=10) + except subprocess.TimeoutExpired: + process.kill() + stdout, stderr = "", "" return AgentResult( command=command, - stdout=_as_text(exc.stdout), - stderr=_as_text(exc.stderr) + f"\ntimed out after {timeout}s", + stdout=stdout or "", + stderr=(stderr or "") + f"\ntimed out after {timeout}s", returncode=124, timed_out=True, ) + except BaseException: + kill_tree(process.pid) + raise return AgentResult( command=command, - stdout=result.stdout, - stderr=result.stderr, - returncode=result.returncode, + stdout=stdout, + stderr=stderr, + returncode=process.returncode, last_message=_read_text(last_message_path), ) finally: @@ -90,11 +102,3 @@ def _read_text(path: str | None) -> str | None: except OSError: return None return text.strip() or None - - -def _as_text(value: str | bytes | None) -> str: - if value is None: - return "" - if isinstance(value, bytes): - return value.decode("utf-8", errors="replace") - return value diff --git a/clodex/cli.py b/clodex/cli.py index b00df21..b97a0df 100644 --- a/clodex/cli.py +++ b/clodex/cli.py @@ -17,7 +17,7 @@ native_status, plan_native_install, ) -from .tasks import TaskManager +from .tasks import Heartbeat, TaskManager from .workflow import ClodexWorkflow @@ -254,7 +254,8 @@ def handle_task(args: argparse.Namespace, workflow: ClodexWorkflow, as_json: boo 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) + with Heartbeat(workflow.state, args.run_id): + 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 diff --git a/clodex/procs.py b/clodex/procs.py new file mode 100644 index 0000000..fde82fd --- /dev/null +++ b/clodex/procs.py @@ -0,0 +1,87 @@ +"""Process helpers: isolate child processes and reliably stop a whole process tree. + +On Windows `os.kill(pid, sig)` calls TerminateProcess (even for signal 0), so liveness +checks and tree kills go through ctypes / taskkill there instead. +""" + +from __future__ import annotations + +import os +import signal +import subprocess +import time + + +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": + return {"creationflags": subprocess.CREATE_NEW_PROCESS_GROUP} + return {"start_new_session": True} + + +def pid_alive(pid: int | None) -> bool: + if not pid or pid <= 0: + return False + if os.name == "nt": + return _windows_pid_alive(int(pid)) + try: + os.kill(int(pid), 0) + except ProcessLookupError: + return False + except PermissionError: + return True + return True + + +def _windows_pid_alive(pid: int) -> bool: + import ctypes + from ctypes import wintypes + + process_query_limited_information = 0x1000 + still_active = 259 + error_access_denied = 5 + kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) + kernel32.OpenProcess.restype = wintypes.HANDLE + handle = kernel32.OpenProcess(process_query_limited_information, False, pid) + if not handle: + # Access denied means the process exists but belongs to someone else. + return ctypes.get_last_error() == error_access_denied + try: + code = wintypes.DWORD() + if not kernel32.GetExitCodeProcess(handle, ctypes.byref(code)): + return True + return code.value == still_active + finally: + kernel32.CloseHandle(handle) + + +def kill_tree(pid: int, grace: float = 3.0) -> None: + """Stop `pid` and everything it started. Best effort; never raises.""" + if not pid or pid <= 0 or pid == os.getpid(): + return + if os.name == "nt": + subprocess.run(["taskkill", "/PID", str(pid), "/T", "/F"], capture_output=True, check=False) + return + try: + group = os.getpgid(pid) + except OSError: + return + if group == os.getpgrp(): + # Not isolated (never put it in its own group): only stop that one process. + try: + os.kill(pid, signal.SIGTERM) + except OSError: + pass + return + try: + os.killpg(group, signal.SIGTERM) + except OSError: + return + deadline = time.monotonic() + grace + while time.monotonic() < deadline and pid_alive(pid): + time.sleep(0.05) + if pid_alive(pid): + try: + os.killpg(group, signal.SIGKILL) + except OSError: + pass diff --git a/clodex/state.py b/clodex/state.py index 262f6b8..daca4e0 100644 --- a/clodex/state.py +++ b/clodex/state.py @@ -157,6 +157,7 @@ def _ensure_run_columns(self, con: sqlite3.Connection) -> None: "handoff_budget": "integer not null default 6", "last_actor": "text", "blocked_reason": "text", + "heartbeat_at": "text", } for name, kind in columns.items(): if name not in existing: @@ -517,6 +518,15 @@ def add_workspace_lock(self, run_id: str, source_path: str, workspace_path: str, (run_id, source_path, workspace_path, backend, now_iso()), ) + def touch_run(self, run_id: str) -> None: + """Worker heartbeat: proves the process that owns this run is still alive.""" + marks = ",".join("?" for _ in TERMINAL_STATUSES) + with self.session() as con: + con.execute( + f"update runs set heartbeat_at=? where id=? and status not in ({marks})", + (now_iso(), run_id, *sorted(TERMINAL_STATUSES)), + ) + 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 b88b632..bd97762 100644 --- a/clodex/tasks.py +++ b/clodex/tasks.py @@ -1,16 +1,64 @@ from __future__ import annotations import os -import signal import subprocess import sys +import threading +import time +from datetime import UTC, datetime 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 .procs import kill_tree, pid_alive, popen_isolation_kwargs +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.""" + + def __init__(self, state: StateStore, run_id: str, interval: float = HEARTBEAT_INTERVAL): + self.state = state + self.run_id = run_id + self.interval = interval + self._stop = threading.Event() + self._thread = threading.Thread(target=self._beat, name="clodex-heartbeat", daemon=True) + + def _beat(self) -> None: + while True: + try: + self.state.touch_run(self.run_id) + except Exception: # noqa: BLE001 - a missed beat must never kill the worker + pass + if self._stop.wait(self.interval): + return + + def __enter__(self) -> "Heartbeat": + self._thread.start() + return self + + def __exit__(self, *exc_info: object) -> None: + self._stop.set() + self._thread.join(timeout=2) class TaskManager: @@ -66,36 +114,95 @@ def start( 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) + # Own process group/session: cancel can then stop the worker and the agents it started. + process = subprocess.Popen( + argv, + cwd=self.repo_root, + stdin=subprocess.DEVNULL, + stdout=stdout, + stderr=stderr, + env=env, + **popen_isolation_kwargs(), + ) 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 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" + + 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.""" + if str(run["status"]) not in ACTIVE_STATUSES: + return run + state = self.worker_state(run) + if state == "dead": + run_id = str(run["id"]) + if str(run["status"]) == "cancel_requested": + self.state.complete_cancel(run_id) + else: + self.state.update_run(run_id, "failed", error="worker process exited before finishing the run") + if run.get("task_id"): + self.state.update_task(str(run["task_id"]), "failed") + self._release_workspace(run) + return self.state.get_run(run_id) or run + return {**run, "worker_state": state} + 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} + return {"run": self.reconcile(run)} def list(self) -> dict[str, Any]: - return {"runs": self.state.list_runs(), "tasks": self.state.list_tasks()} + return {"runs": [self.reconcile(run) for run in 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}") + status = str(run["status"]) + if status in TERMINAL_STATUSES: + return WorkflowResult(status, run_id, run.get("task_id"), run.get("artifacts_dir"), {"cancel_requested": False, "note": f"run already {status}"}) self.state.request_cancel(run_id) - pid = run.get("pid") - if pid: - self._terminate(int(pid)) + killed = False + if self.worker_state(run) == "alive": + kill_tree(int(run["pid"])) + killed = True self.state.complete_cancel(run_id) + released = self._release_workspace(run) 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 + return WorkflowResult( + str(updated["status"]), + run_id, + updated.get("task_id"), + updated.get("artifacts_dir"), + {"cancel_requested": True, "worker_stopped": killed, "workspace_released": released}, + ) + + def _release_workspace(self, run: dict[str, Any]) -> bool: + path = run.get("workspace_path") + if not path: + return False + manager = WorkspaceManager(self.repo_root, self.config) + released = False + for _ in range(10): # the stopped worker may take a moment to let go of the directory + released = manager.release(str(path)) + if released or not Path(str(path)).exists(): + break + time.sleep(0.2) + if released: + self.state.release_workspace_lock(str(run["id"])) + return released diff --git a/tests/support.py b/tests/support.py index 8a3a5c8..4c5f4d8 100644 --- a/tests/support.py +++ b/tests/support.py @@ -118,6 +118,8 @@ def _write_fake(self): if {self.sleep_seconds!r}: import time + # Lets tests find this process (the real agent, behind the .cmd shim) and check it is stopped. + Path('.fake-pid-' + name).write_text(str(os.getpid())) time.sleep({self.sleep_seconds!r}) def diff_hash(): diff --git a/tests/test_process_control.py b/tests/test_process_control.py new file mode 100644 index 0000000..283b04b --- /dev/null +++ b/tests/test_process_control.py @@ -0,0 +1,204 @@ +"""Tests: process-tree control, worker supervision and cancellation.""" + +from __future__ import annotations + +import os +import subprocess +import sys +import time +import unittest +import warnings +from datetime import UTC, datetime, timedelta +from pathlib import Path + +from clodex.agents import AgentRunner +from clodex.commands import claude_plan_command +from clodex.config import load_config +from clodex.procs import kill_tree, pid_alive, popen_isolation_kwargs +from clodex.tasks import Heartbeat, TaskManager +from clodex.workflow import ClodexWorkflow +from tests.support import FakeCliPath, TempRepo + + +def wait_for(predicate, timeout: float, interval: float = 0.1) -> bool: + deadline = time.monotonic() + timeout + while time.monotonic() < deadline: + if predicate(): + return True + time.sleep(interval) + return predicate() + + +def read_pid(path: Path) -> int | None: + try: + return int(path.read_text(encoding="utf-8").strip()) + except (OSError, ValueError): + return None + + +def finished_pid() -> int: + child = subprocess.Popen([sys.executable, "-c", "pass"]) + child.wait() + return child.pid + + +class ProcessHelperTests(unittest.TestCase): + def test_pid_alive(self): + self.assertTrue(pid_alive(os.getpid())) + self.assertFalse(pid_alive(None)) + self.assertFalse(pid_alive(0)) + self.assertFalse(pid_alive(finished_pid())) + + def test_kill_tree_stops_the_process_and_its_children(self): + child_pid_file = Path(self.id() + ".pid") + script = ( + "import subprocess, sys, time\n" + "child = subprocess.Popen([sys.executable, '-c', 'import time; time.sleep(60)'])\n" + f"open({str(child_pid_file)!r}, 'w').write(str(child.pid))\n" + "time.sleep(60)\n" + ) + self.addCleanup(lambda: child_pid_file.unlink(missing_ok=True)) + parent = subprocess.Popen([sys.executable, "-c", script], **popen_isolation_kwargs()) + self.addCleanup(lambda: parent.poll() is None and parent.kill()) + self.assertTrue(wait_for(lambda: read_pid(child_pid_file) is not None, 15)) + child_pid = read_pid(child_pid_file) + self.assertTrue(pid_alive(parent.pid) and pid_alive(child_pid)) + kill_tree(parent.pid) + self.assertTrue(wait_for(lambda: not pid_alive(parent.pid), 10), "parent still alive") + self.assertTrue(wait_for(lambda: not pid_alive(child_pid), 10), "child outlived its parent") + parent.wait(timeout=10) + + def test_kill_tree_ignores_bad_pids(self): + kill_tree(0) + kill_tree(-5) + kill_tree(os.getpid()) # never kills the caller + self.assertTrue(pid_alive(os.getpid())) + + +class AgentTimeoutTests(unittest.TestCase): + def test_timeout_stops_the_agent_process_not_just_its_shim(self): + with TempRepo() as repo, FakeCliPath(sleep_seconds=30): + config = load_config(repo) + started = time.monotonic() + result = AgentRunner(repo).run(claude_plan_command(config), "prompt", timeout=2) + self.assertTrue(result.timed_out) + self.assertEqual(result.returncode, 124) + self.assertLess(time.monotonic() - started, 20, "should not wait for the orphan to finish sleeping") + pid = read_pid(repo / ".fake-pid-claude") + self.assertIsNotNone(pid, "fake agent never started") + self.assertTrue(wait_for(lambda: not pid_alive(pid), 10), "agent process survived the timeout") + + +class WorkerSupervisionTests(unittest.TestCase): + def setUp(self): + self.repo_cm = TempRepo() + self.repo = self.repo_cm.__enter__() + self.addCleanup(self.repo_cm.__exit__, None, None, None) + self.manager = TaskManager(self.repo) + self.state = self.manager.state + self.state.upsert_task("t", "task", "running") + + def make_run(self, run_id: str, status: str, pid: int | None, heartbeat_age: float | None = None) -> None: + self.state.create_run(run_id, "t", "prompt", status, pid=pid) + if heartbeat_age is not None: + stamp = (datetime.now(UTC) - timedelta(seconds=heartbeat_age)).replace(microsecond=0).isoformat().replace("+00:00", "Z") + with self.state.session() as con: + con.execute("update runs set heartbeat_at=? where id=?", (stamp, run_id)) + + def test_dead_worker_is_detected_and_the_run_failed(self): + self.make_run("r1", "running", finished_pid()) + data = self.manager.get("r1") + self.assertEqual(data["run"]["status"], "failed") + self.assertIn("worker process exited", data["run"]["error"]) + task = next(t for t in self.state.list_tasks() if t["id"] == "t") + self.assertEqual(task["status"], "failed") + + def test_dead_worker_of_a_cancelling_run_completes_the_cancel(self): + self.make_run("r1", "cancel_requested", finished_pid()) + self.assertEqual(self.manager.get("r1")["run"]["status"], "cancelled") + + def test_list_reconciles_every_active_run_and_leaves_finished_ones(self): + self.make_run("dead", "queued", finished_pid()) + self.make_run("done", "approved", finished_pid()) + runs = {run["id"]: run["status"] for run in self.manager.list()["runs"]} + self.assertEqual(runs, {"dead": "failed", "done": "approved"}) + + def test_worker_state_classification(self): + self.make_run("fresh", "running", os.getpid(), heartbeat_age=1) + self.make_run("quiet", "running", os.getpid(), heartbeat_age=600) + self.make_run("gone", "running", finished_pid()) + self.make_run("nopid", "queued", None) + states = {rid: self.manager.worker_state(self.state.get_run(rid)) for rid in ("fresh", "quiet", "gone", "nopid")} + self.assertEqual(states, {"fresh": "alive", "quiet": "stale", "gone": "dead", "nopid": "none"}) + + def test_cancel_never_signals_a_live_pid_without_a_recent_heartbeat(self): + # This pid is the test process itself: a stale heartbeat means it may be a reused pid. + self.make_run("r1", "running", os.getpid(), heartbeat_age=600) + result = self.manager.cancel("r1") + self.assertEqual(result.status, "cancelled") + self.assertFalse(result.data["worker_stopped"]) + self.assertTrue(pid_alive(os.getpid())) + + def test_cancel_of_a_finished_run_changes_nothing(self): + self.make_run("r1", "approved", None) + result = self.manager.cancel("r1") + self.assertEqual(result.status, "approved") + self.assertFalse(result.data["cancel_requested"]) + self.assertEqual(self.state.get_run("r1")["status"], "approved") + with self.assertRaises(ValueError): + self.manager.cancel("missing") + + def test_heartbeat_updates_while_running_and_not_after_finishing(self): + self.make_run("r1", "running", None) + self.assertIsNone(self.state.get_run("r1")["heartbeat_at"]) + with Heartbeat(self.state, "r1", interval=0.05): + self.assertTrue(wait_for(lambda: self.state.get_run("r1")["heartbeat_at"] is not None, 5)) + self.state.update_run("r1", "approved") + self.state.touch_run("r1") + before = self.state.get_run("r1")["heartbeat_at"] + time.sleep(1.1) + self.state.touch_run("r1") + self.assertEqual(self.state.get_run("r1")["heartbeat_at"], before, "finished runs are no longer touched") + + +class CancelEndToEndTests(unittest.TestCase): + def setUp(self): + # TaskManager.start hands back a detached worker whose Popen handle is never reaped here. + context = warnings.catch_warnings() + context.__enter__() + self.addCleanup(context.__exit__, None, None, None) + warnings.simplefilter("ignore", ResourceWarning) + + def test_cancel_stops_the_worker_and_the_agent_it_is_running(self): + with TempRepo() as repo, FakeCliPath(sleep_seconds=60): + manager = TaskManager(repo) + started = manager.start("slow fixture", workspace_backend="local") + run_id = started.run_id + worker_pid = started.data["pid"] + agent_pid_file = repo / ".fake-pid-claude" + self.assertTrue(wait_for(lambda: read_pid(agent_pid_file) is not None, 45), "worker never reached the agent") + agent_pid = read_pid(agent_pid_file) + self.assertTrue(pid_alive(worker_pid) and pid_alive(agent_pid)) + result = manager.cancel(run_id) + self.assertEqual(result.status, "cancelled") + self.assertTrue(result.data["worker_stopped"]) + self.assertTrue(wait_for(lambda: not pid_alive(worker_pid), 10), "worker survived cancel") + self.assertTrue(wait_for(lambda: not pid_alive(agent_pid), 10), "agent survived cancel") + self.assertEqual(manager.state.get_run(run_id)["status"], "cancelled") + + def test_cancelled_worktree_run_releases_its_worktree(self): + with TempRepo() as repo, FakeCliPath(sleep_seconds=60): + manager = TaskManager(repo) + started = manager.start("slow fixture", workspace_backend="git-worktree") + workspace = Path(manager.config.workspace_root) / started.run_id + self.assertTrue(wait_for(lambda: workspace.exists() and read_pid(workspace / ".fake-pid-claude") is not None, 45)) + result = manager.cancel(started.run_id) + self.assertEqual(result.status, "cancelled") + self.assertTrue(result.data["workspace_released"]) + self.assertFalse(workspace.exists()) + worktrees = subprocess.run(["git", "worktree", "list", "--porcelain"], cwd=repo, capture_output=True, text=True).stdout + self.assertEqual(worktrees.count("worktree "), 1) + + +if __name__ == "__main__": + unittest.main() From ef5bd673d66303a17ac8ecc23ebefbb5c47387c7 Mon Sep 17 00:00:00 2001 From: Devil Date: Wed, 30 Sep 2026 22:39:05 -0400 Subject: [PATCH 5/5] docs: document quorum, apply safety, clean and async task supervision Co-Authored-By: Claude Sonnet 5.5 --- README.md | 23 ++++++++++++++++++++++- 1 file changed, 22 insertions(+), 1 deletion(-) diff --git a/README.md b/README.md index 4d46efd..1ec8f7b 100644 --- a/README.md +++ b/README.md @@ -116,7 +116,8 @@ PowerShell: | `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 apply ` | Apply an approved worktree patch back to the source checkout (`--check` to dry-run, `--force` for unapproved runs) | +| `clodex clean ` | Remove the kept git worktree of a finished run (artifacts and patch stay) | | `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 | @@ -169,6 +170,26 @@ 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. +`clodex apply` only applies **approved** runs, and only if `apply.patch` still matches the +approved diff hash; a local-workspace run is reported as already in the working tree. A run +that fails unexpectedly is marked `failed` (traceback in `error.txt`) and its worktree is +removed; approved and blocked runs keep theirs until `clodex clean `. + +### Audit quorum + +`audit.quorum` decides when the reviewers agree: `unanimous` (default), `majority`, or a number +N. Only reviewers with `required: true` count; optional reviewers are recorded but never block. +A reviewer that fails or times out counts as not approved. If a *required* reviewer cannot run, +the run is `blocked` with that error rather than sent back to Codex for a fix. Very large diffs +are truncated in the audit prompt (`audit.max_diff_bytes`, default 200000); the diff hash still +covers the whole diff. + +### Async tasks + +`clodex task start` runs the build in a detached worker. `task cancel` stops the worker and any +agent it started, and removes the run's worktree. Workers write a heartbeat; `task get/list` +mark a run `failed` if its worker died without finishing. + ## MCP Tools The MCP server exposes: