fix: resilient audit, safe apply, state integrity and process control

README.md

@@ -116,7 +116,8 @@ PowerShell:
 | `clodex build "<task>"` | Run plan, implementation, and dual audit loop in an isolated worktree |
 | `clodex audit --diff` | Audit current uncommitted changes |
 | `clodex run "<task>"` | Alias for `build` |
-| `clodex apply <run-id>` | Apply an approved worktree patch back to the source checkout |
+| `clodex apply <run-id>` | Apply an approved worktree patch back to the source checkout (`--check` to dry-run, `--force` for unapproved runs) |
+| `clodex clean <run-id>` | 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 <run-id>` | 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/<run-id>/`.
 The source checkout is not modified until `clodex apply <run-id>` 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 <run-id>`.
+
+### 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:

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

clodex/artifacts.py

@@ -28,37 +28,43 @@ 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) -> 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")
+        self._record(name, path)
         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()

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
 
 
@@ -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":
@@ -246,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

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},

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

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", errors="replace")
+    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"
+    )