feat: quorum-driven handoff decisions, handoff workspaces and ledger rows

README.md

@@ -307,6 +307,25 @@ The server speaks MCP `2025-11-25` (and negotiates down to `2025-06-18`, `2025-0
 - `clodex_audit` accepts `base` (everything since that ref diverged) and `commit` (one commit) as
   well as the default uncommitted diff.
 
+### Native handoffs
+
+- **Decisions follow your audit settings.** `clodex_handoff_decide` evaluates the reports recorded
+  with `clodex_handoff_update` against the same `audit.reviewers` and `audit.quorum` as a classic
+  run (`unanimous`, `majority` or a number; only `required` reviewers count). A report is a verdict
+  when it has `approved` and a diff hash (on the update or in the report); it only counts for that
+  hash, a new hash resets everything, and a hashless `approved: false` withdraws the reviewer's
+  approval. `report.reviewer_id` names which configured reviewer the verdict is for; without it
+  the actor (`claude` / `codex`) stands for the first *required* reviewer of that backend, so the
+  default two-reviewer flow needs no extra fields. An unknown `reviewer_id` is rejected. The
+  decision lists `required_pending` and `approved_reviewers` so the orchestrator knows whom to ask.
+- **Workspaces.** `clodex_handoff_create` takes `workspace: "git-worktree" | "local" | "none"`
+  (default `none`). A worktree is an isolated checkout under `.clodex/workspaces/<run-id>` (the
+  repo must have no uncommitted changes to tracked files); if the handoff cannot be recorded the
+  worktree is rolled back. `clodex clean <run-id>` removes it once the handoff has finished.
+- **Ledger.** A handoff gets a row in the task ledger (shown by `clodex status`, kept in step
+  with the run: `done` / `blocked` / `failed`), every verdict report becomes an `audits` row, and
+  `report.artifacts` (paths, or `{name, path, kind}`) become artifact rows.
+
 ## Safety
 
 Clodex defaults to git worktree isolation, Codex `workspace-write` sandboxing,

clodex/mcp_server.py

@@ -3,6 +3,7 @@
 import base64
 import json
 import sqlite3
+import subprocess
 import sys
 import threading
 import uuid
@@ -12,9 +13,11 @@
 
 from . import __version__
 from .config import resolve_repo_root
+from .quorum import evaluate_handoff, resolve_reviewer
 from .schemas import validate as validate_schema
 from .tasks import TaskManager
 from .workflow import ClodexWorkflow
+from .workspace import DirtyWorkspaceError, WorkspaceManager
 
 
 TOOLS = [
@@ -105,14 +108,19 @@
                 "owner": {"type": "string"},
                 "phase": {"type": "string"},
                 "handoff_budget": {"type": "integer"},
+                "workspace": {
+                    "type": "string",
+                    "enum": ["none", "git-worktree", "local"],
+                    "description": "Give the handoff its own isolated git worktree (or the repo itself) for Codex to work in. Default none.",
+                },
             },
             "required": ["task"],
         },
     },
     {
         "name": "clodex_handoff_update",
         "title": "Update Clodex handoff",
-        "description": "Record native handoff phase, actor, report, status, and budget usage.",
+        "description": "Record native handoff phase, actor, report, status, and budget usage. A report with `approved` and a diff hash is a verdict; `reviewer_id` names which configured reviewer it is for (default: the first required reviewer of the actor's backend).",
         "inputSchema": {
             "type": "object",
             "properties": {
@@ -532,20 +540,41 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
             return call_text(str(exc), is_error=True)
         return {"content": [{"type": "text", "text": json.dumps(result.__dict__, indent=2)}], "isError": False}
     elif name == "clodex_handoff_create":
+        run_id = str(arguments.get("run_id") or f"native-{uuid.uuid4().hex[:12]}")
+        workspace = None
+        manager = WorkspaceManager(workflow.repo_root, workflow.config)
         try:
             handoff_budget = handoff_budget_argument(arguments)
+            task = str(required_argument(arguments, "task"))
+            backend = arguments.get("workspace")
+            if backend not in (None, "none", "git-worktree", "local"):
+                raise ValueError("workspace must be one of: none, git-worktree, local")
+            if backend in {"git-worktree", "local"}:
+                if workflow.state.get_run(run_id) is not None:
+                    raise ValueError(f"run already exists: {run_id}")  # before a worktree is created for it
+                workspace = manager.prepare(run_id, backend)
             run = workflow.state.create_handoff(
-                str(arguments.get("run_id") or f"native-{uuid.uuid4().hex[:12]}"),
-                str(required_argument(arguments, "task")),
+                run_id,
+                task,
                 owner=str(arguments.get("owner") or "claude"),
                 phase=str(arguments.get("phase") or "planning"),
                 handoff_budget=handoff_budget,
+                workspace_path=str(workspace.path) if workspace else None,
             )
-        except (KeyError, TypeError, ValueError, sqlite3.IntegrityError) as exc:
+            if workspace is not None:
+                workflow.state.add_workspace_lock(run_id, str(workspace.source_path), str(workspace.path), workspace.backend)
+        except (KeyError, TypeError, ValueError, sqlite3.IntegrityError, DirtyWorkspaceError, subprocess.CalledProcessError) as exc:
+            if workspace is not None and workspace.is_worktree:
+                manager.release(workspace.path)  # do not leave a worktree behind for a handoff that was never created
             return call_text(expected_handoff_error(exc), is_error=True)
-        return call_json(run)
+        return call_json({**run, **({"workspace": workspace.as_dict()} if workspace else {})})
     elif name == "clodex_handoff_update":
         try:
+            report_arg = arguments.get("report") if isinstance(arguments.get("report"), dict) else None
+            named = (report_arg or {}).get("reviewer_id")
+            if named is not None and resolve_reviewer(named, workflow.config.reviewers) is None:
+                configured = ", ".join(str(r.get("id")) for r in workflow.config.reviewers)
+                raise ValueError(f"unknown reviewer_id: {named} (configured reviewers: {configured})")
             run = workflow.state.update_handoff(
                 str(required_argument(arguments, "run_id")),
                 phase=arguments.get("phase"),
@@ -580,6 +609,7 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
 
         run = data["run"]
         status = run.get("status")
+        agreement = evaluate_handoff(data, workflow.config.reviewers, workflow.config.audit.get("quorum", "unanimous"))
         if status == "approved":
             decision = {"decision": "approved", "run_id": run["id"], "diff_hash": run.get("diff_hash")}
             is_error = False
@@ -591,7 +621,7 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
                 "blocked_reason": run.get("blocked_reason") or run.get("error"),
             }
             is_error = True
-        elif agreement := handoff_agreement(data):
+        elif agreement["approved"]:
             try:
                 approved_run = workflow.state.approve_handoff(run["id"], agreement["diff_hash"], approved_by=agreement["approved_by"])
             except (KeyError, TypeError, ValueError, sqlite3.IntegrityError) as exc:
@@ -601,6 +631,7 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
                 "run_id": approved_run["id"],
                 "diff_hash": approved_run.get("diff_hash"),
                 "approved_by": agreement["approved_by"],
+                "approved_reviewers": agreement["approved_reviewers"],
             }
             is_error = False
         else:
@@ -610,6 +641,9 @@ def tool_call(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
                 "phase": run.get("phase"),
                 "budget_remaining": data["budget_remaining"],
                 "next_expected_actor": data["next_expected_actor"],
+                "required_pending": agreement["required_pending"],
+                "approved_reviewers": agreement["approved_reviewers"],
+                "quorum": agreement["quorum"],
             }
             is_error = False
         workflow.state.add_event(run["id"], "handoff.decide", decision)
@@ -642,57 +676,12 @@ def handoff_budget_argument(arguments: dict[str, Any]) -> int:
     return value
 
 
-def handoff_agreement(data: dict[str, Any]) -> dict[str, Any] | None:
-    latest_hash: str | None = None
-    approvals: set[str] = set()
-    for event in data.get("events") or []:
-        if event.get("event") != "handoff.update":
-            continue
-        event_data = event.get("data")
-        if not isinstance(event_data, dict):
-            continue
-        report = event_data.get("report")
-        if not isinstance(report, dict):
-            report = {}
-        actor = normalized_actor(event_data.get("actor") or report.get("actor"))
-        diff_hash = normalized_diff_hash(event_data.get("diff_hash") or report.get("diff_hash"))
-        if diff_hash is None:
-            if latest_hash is not None and actor is not None and report.get("approved") is False:
-                approvals.discard(actor)
-            continue
-        if diff_hash != latest_hash:
-            latest_hash = diff_hash
-            approvals = set()
-        if actor is None:
-            continue
-        if report.get("approved") is True:
-            approvals.add(actor)
-        elif report.get("approved") is False:
-            approvals.discard(actor)
-
-    if latest_hash is not None and {"claude", "codex"}.issubset(approvals):
-        return {"diff_hash": latest_hash, "approved_by": sorted(approvals)}
-    return None
-
-
-def normalized_actor(value: Any) -> str | None:
-    if value is None:
-        return None
-    actor = str(value).strip().lower()
-    return actor if actor in {"claude", "codex"} else None
-
-
-def normalized_diff_hash(value: Any) -> str | None:
-    if value is None:
-        return None
-    diff_hash = str(value).strip()
-    return diff_hash or None
-
-
 def expected_handoff_error(exc: Exception) -> str:
     if isinstance(exc, KeyError):
         key = exc.args[0] if exc.args else "argument"
         return f"Missing required argument: {key}"
+    if isinstance(exc, subprocess.CalledProcessError):
+        return (exc.stderr or "").strip() or str(exc)
     return str(exc)
 
 

clodex/quorum.py

@@ -90,3 +90,95 @@ def required_fixes(verdicts: list[dict[str, Any]], diff_hash: str) -> list[str]:
                 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."]
+
+
+# ---------------------------------------------------------------- native handoffs
+
+
+def normalized_actor(value: Any) -> str | None:
+    if value is None:
+        return None
+    actor = str(value).strip().lower()
+    return actor if actor in {"claude", "codex"} else None
+
+
+def normalized_diff_hash(value: Any) -> str | None:
+    if value is None:
+        return None
+    diff_hash = str(value).strip()
+    return diff_hash or None
+
+
+def resolve_reviewer(identifier: Any, reviewers: list[dict[str, Any]]) -> dict[str, Any] | None:
+    """Map a report's reviewer id, or just its actor (`claude` / `codex`), onto a configured reviewer.
+
+    An actor with no explicit reviewer id stands for the first *required* reviewer of its backend
+    (the first of that backend if none is required), so the default claude/codex flow keeps working.
+    """
+    if identifier is None:
+        return None
+    name = str(identifier).strip()
+    for reviewer in reviewers:
+        if str(reviewer.get("id")) == name:
+            return reviewer
+    backend = normalized_actor(name)
+    if backend is None:
+        return None
+    same_backend = [r for r in reviewers if str(r.get("backend")) == backend]
+    required = [r for r in same_backend if r.get("required", True)]
+    return (required or same_backend or [None])[0]
+
+
+def evaluate_handoff(data: dict[str, Any], reviewers: list[dict[str, Any]], quorum: Any = "unanimous") -> dict[str, Any]:
+    """Whether the reports recorded on a handoff satisfy the configured reviewers and quorum.
+
+    Same rules as the classic audit: a report approves only for the diff hash it names, a new
+    hash resets every approval, and a hashless rejection withdraws that reviewer's approval.
+    """
+    latest_hash: str | None = None
+    approvals: dict[str, bool] = {}
+    for event in data.get("events") or []:
+        if event.get("event") != "handoff.update":
+            continue
+        event_data = event.get("data")
+        if not isinstance(event_data, dict):
+            continue
+        report = event_data.get("report")
+        if not isinstance(report, dict):
+            report = {}
+        reviewer = resolve_reviewer(report.get("reviewer_id") or event_data.get("reviewer_id") or event_data.get("actor") or report.get("actor"), reviewers)
+        diff_hash = normalized_diff_hash(event_data.get("diff_hash") or report.get("diff_hash"))
+        reviewer_id = str(reviewer["id"]) if reviewer else None
+        if diff_hash is None:
+            if latest_hash is not None and reviewer_id is not None and report.get("approved") is False:
+                approvals[reviewer_id] = False
+            continue
+        if diff_hash != latest_hash:
+            latest_hash = diff_hash
+            approvals = {}
+        if reviewer_id is None:
+            continue
+        if report.get("approved") is True:
+            approvals[reviewer_id] = True
+        elif report.get("approved") is False:
+            approvals[reviewer_id] = False
+
+    verdicts = [
+        {
+            "reviewer_id": str(reviewer["id"]),
+            "approved": bool(approvals.get(str(reviewer["id"]))),
+            "diff_hash": latest_hash if approvals.get(str(reviewer["id"])) else None,
+            "persona": reviewer.get("persona"),
+            "_required": bool(reviewer.get("required", True)),
+        }
+        for reviewer in reviewers
+    ]
+    result = evaluate(verdicts, latest_hash or "", 0, quorum)
+    result["approved"] = bool(result["approved"] and latest_hash)
+    result["diff_hash"] = latest_hash
+    by_id = {str(reviewer["id"]): reviewer for reviewer in reviewers}
+    approving = [rid for rid in result["required_approved_by"] if rid in by_id]
+    result["approved_reviewers"] = approving
+    # `approved_by` lists the agents (claude / codex) behind the approving reviewers, as it always has.
+    result["approved_by"] = sorted({str(by_id[rid].get("backend")) for rid in approving})
+    return result

clodex/state.py

@@ -262,21 +262,32 @@ def create_handoff(
         phase: str = "planning",
         handoff_budget: int = 6,
         task_id: str | None = None,
+        workspace_path: str | None = None,
     ) -> dict[str, Any]:
         if handoff_budget < 0:
             raise ValueError("handoff_budget must be non-negative")
+        if phase not in HANDOFF_PHASES:
+            raise ValueError(f"unknown handoff phase: {phase} (use one of {sorted(HANDOFF_PHASES)})")
 
         timestamp = now_iso()
+        task_id = task_id or run_id  # every handoff shows up in the task ledger (`clodex status`)
         with self.session() as con:
             con.execute(
                 """
                 insert into runs(
                     id, task_id, status, prompt, owner, phase, handoff_count,
-                    handoff_budget, created_at, updated_at, started_at
+                    handoff_budget, workspace_path, created_at, updated_at, started_at
                 )
-                values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                values (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (run_id, task_id, "handoff", prompt, owner, phase, 0, handoff_budget, workspace_path, timestamp, timestamp, timestamp),
+            )
+            con.execute(
+                """
+                insert into tasks(id, title, status, created_at, updated_at) values (?, ?, ?, ?, ?)
+                on conflict(id) do update set status=excluded.status, updated_at=excluded.updated_at
                 """,
-                (run_id, task_id, "handoff", prompt, owner, phase, 0, handoff_budget, timestamp, timestamp, timestamp),
+                (task_id, prompt.strip().splitlines()[0][:200] if prompt.strip() else run_id, "handoff", timestamp, timestamp),
             )
             self._insert_event(
                 con,
@@ -369,6 +380,8 @@ def update_handoff(
 
             updated = con.execute("select * from runs where id=?", (run_id,)).fetchone()
             updated_run = dict(updated)
+            self._record_report(con, run_id, actor, report, diff_hash, timestamp)
+            self._sync_task(con, updated_run, timestamp)
             updated_count = _int_or_default(updated_run.get("handoff_count"), 0)
             updated_budget = _int_or_default(updated_run.get("handoff_budget"), 6)
             event_data: dict[str, Any] = {
@@ -430,8 +443,44 @@ def approve_handoff(self, run_id: str, diff_hash: str, approved_by: list[str] |
                 timestamp,
             )
             updated = con.execute("select * from runs where id=?", (run_id,)).fetchone()
+            self._sync_task(con, dict(updated), timestamp)
             return dict(updated)
 
+    # Run status -> task status for the ledger; handoff runs in flight leave the task alone.
+    _TASK_STATUS = {"approved": "done", "applied": "done", "completed": "done", "blocked": "blocked", "failed": "failed", "cancelled": "cancelled"}
+
+    def _sync_task(self, con: sqlite3.Connection, run: dict[str, Any], timestamp: str) -> None:
+        task_status = self._TASK_STATUS.get(str(run.get("status")))
+        if task_status and run.get("task_id"):
+            con.execute("update tasks set status=?, updated_at=? where id=?", (task_status, timestamp, run["task_id"]))
+
+    def _record_report(self, con: sqlite3.Connection, run_id: str, actor: str | None, report: dict[str, Any] | None, diff_hash: str | None, timestamp: str) -> None:
+        """Put what a handoff report says into the same ledger the classic workflow uses.
+
+        A verdict (`approved` true/false) becomes an `audits` row, and `artifacts` entries (paths or
+        {name, path, kind}) become `artifacts` rows, so `status`, exports and the MCP tools see them.
+        """
+        if not isinstance(report, dict):
+            return
+        if isinstance(report.get("approved"), bool):
+            agent = str(report.get("reviewer_id") or actor or "unknown")
+            con.execute(
+                "insert into audits(run_id, agent, approved, diff_hash, verdict_json, created_at) values (?, ?, ?, ?, ?, ?)",
+                (run_id, agent, int(report["approved"]), diff_hash or report.get("diff_hash"), json.dumps(report, default=str), timestamp),
+            )
+        artifacts = report.get("artifacts")
+        for item in artifacts if isinstance(artifacts, list) else []:
+            path = item.get("path") if isinstance(item, dict) else item
+            if not isinstance(path, str) or not path.strip():
+                continue
+            name = str(item.get("name")) if isinstance(item, dict) and item.get("name") else Path(path).name
+            kind = str(item.get("kind")) if isinstance(item, dict) and item.get("kind") else (Path(path).suffix.lstrip(".") or "file")
+            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, timestamp),
+            )
+
     def get_handoff(self, run_id: str) -> dict[str, Any] | None:
         with self.session() as con:
             row = con.execute("select * from runs where id=?", (run_id,)).fetchone()