diff --git a/README.md b/README.md index 960f22c..2e3c341 100644 --- a/README.md +++ b/README.md @@ -83,6 +83,9 @@ Native coordination uses these MCP tools: - `clodex_handoff_get` - `clodex_handoff_decide` - `clodex_delegate` +- `clodex_clarify` +- `clodex_answer` +- `clodex_messages` If MCP is unavailable, use the CLI fallbacks: @@ -348,6 +351,21 @@ non-interactive `codex exec` has no MCP access to report for itself: 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. +- **Clarifications.** Either agent can ask the other a question instead of guessing: + `clodex_clarify` (`run_id`, `actor`, `question`) opens it, `clodex_answer` (`run_id`, + `message_id`, `answer`, `actor`; the asker cannot answer its own question) closes it, and + `clodex_messages` lists them (`status`: open / answered / delivered). While a question is open + `clodex_handoff_get` shows it under `open_clarifications` and the turn passes to whoever must + answer, and `clodex_handoff_decide` returns `needs_fix` even if the reviewers agree. When + Codex is run by `clodex_delegate` it is told to end its final message with + `{"clarifications": ["..."]}` instead of guessing; the worker posts those as questions, and the + answers are put into Codex's next delegation prompt (then marked `delivered`). A question + handed back counts as a handoff against the budget. +- **Guards.** `clodex_handoff_decide` refuses while a delegation is still running, so you never + decide on a diff that is changing. `actor` and `owner` must be `claude` or `codex` (case is + normalized); anything else is rejected rather than recorded. Turns are *not* strictly + alternated: an orchestrator legitimately records several updates in a row, so only the worker's + own handoffs spend budget automatically (interactive agents still pass `increment_handoff`). - **Workspaces.** `clodex_handoff_create` takes `workspace: "git-worktree" | "local" | "none"` (default `none`). A worktree is an isolated checkout under `.clodex/workspaces/` (the repo must have no uncommitted changes to tracked files); if the handoff cannot be recorded the diff --git a/clodex/delegate.py b/clodex/delegate.py index 0ba7549..bbde035 100644 --- a/clodex/delegate.py +++ b/clodex/delegate.py @@ -7,7 +7,9 @@ from __future__ import annotations +import json import os +import re import subprocess import sys import time @@ -29,12 +31,41 @@ POLL_SECONDS = 0.5 SUMMARY_LIMIT = 4000 +MAX_CLARIFICATIONS = 10 +_FENCED_JSON = re.compile(r"```(?:json)?\s*(\{.*?\})\s*```", re.DOTALL) class WaitAborted(Exception): """The caller stopped waiting (cancelled request or closed connection).""" +def extract_clarifications(text: str) -> list[str]: + """Questions Codex asked instead of guessing: `{"clarifications": ["...", ...]}` in its final message. + + Entries may be strings or `{"question": "..."}`. Anything else (including other JSON objects + in a report that quotes code) is ignored; at most MAX_CLARIFICATIONS are kept. + """ + candidates = list(_FENCED_JSON.findall(text)) + for match in re.finditer(r"\{", text): # the last JSON object in the message, fenced or not + candidates.append(text[match.start():]) + for candidate in reversed(candidates): + try: + payload, _ = json.JSONDecoder().raw_decode(candidate.lstrip()) + except ValueError: + continue + items = payload.get("clarifications") if isinstance(payload, dict) else None + if not isinstance(items, list): + continue + questions = [] + for item in items: + question = item.get("question") if isinstance(item, dict) else item + if isinstance(question, str) and question.strip(): + questions.append(question.strip()) + if questions: + return questions[:MAX_CLARIFICATIONS] + return [] + + def delegation_worker_state(delegation: dict[str, Any]) -> str: return worker_state(delegation.get("pid"), delegation.get("heartbeat_at"), delegation.get("started_at") or delegation.get("created_at")) @@ -236,14 +267,22 @@ def _execute(config: ClodexConfig, state: StateStore, run_id: str, delegation: d return _audit(config, state, run_id, delegation, run, runner, artifacts, workspace, instructions, task) fixes = open_fixes(data, config.reviewers) if mode == "fix" else None - prompt = delegate_prompt(mode, task, instructions, fixes) + answered = state.answered_clarifications(run_id) + prompt = delegate_prompt(mode, task, instructions, fixes, [(item["question"], item["answer"]) for item in answered]) result = runner.run(codex_exec_command(config, workspace, approval_profile=delegation.get("approval_profile")), prompt) report_path = artifacts.write_text(f"delegation-{delegation['id']}-codex.md", _report_text(result.stdout, result.stderr)) if not result.ok: raise RuntimeError(f"codex exited with code {result.returncode}: {(result.stderr or result.stdout).strip()[:500]}") diff, diff_hash = _prepare_diff(workspace, config.repo_root) diff_path = artifacts.write_text("changes.diff", diff, exact=True) + if answered: + state.mark_delivered([int(item["id"]) for item in answered]) # Codex has now seen them summary = result.stdout.strip() or "(codex printed no report)" + questions = extract_clarifications(result.stdout) + for question in questions: # before the handoff update: that may block the handoff, which then takes no messages + state.add_clarification(run_id, "codex", question) + if questions and not diff.strip(): + summary = "Codex needs clarification: " + "; ".join(questions) state.update_handoff( run_id, phase="implementation" if mode == "implement" else "fix", @@ -253,6 +292,7 @@ def _execute(config: ClodexConfig, state: StateStore, run_id: str, delegation: d report={ "summary": summary[:SUMMARY_LIMIT], "changed": bool(diff.strip()), + **({"clarifications": questions} if questions else {}), "delegation": delegation["id"], "mode": mode, "artifacts": [str(report_path), str(diff_path)], diff --git a/clodex/mcp_server.py b/clodex/mcp_server.py index 473ba1c..679ed8d 100644 --- a/clodex/mcp_server.py +++ b/clodex/mcp_server.py @@ -17,6 +17,7 @@ from .delegate import DelegationManager, WaitAborted from .quorum import evaluate_handoff, resolve_reviewer from .schemas import validate as validate_schema +from .state import TERMINAL_STATUSES from .tasks import TaskManager from .workflow import ClodexWorkflow from .workspace import DirtyWorkspaceError, WorkspaceManager @@ -166,6 +167,40 @@ "required": ["run_id"], }, }, + { + "name": "clodex_clarify", + "title": "Ask a clarifying question", + "description": "Ask the other agent a question about a handoff instead of guessing. It stays open (and blocks approval) until answered with clodex_answer; answers are given to Codex in its next delegation.", + "inputSchema": { + "type": "object", + "properties": {"run_id": {"type": "string"}, "actor": {"type": "string", "enum": ["claude", "codex"]}, "question": {"type": "string"}}, + "required": ["run_id", "actor", "question"], + }, + }, + { + "name": "clodex_answer", + "title": "Answer a clarifying question", + "description": "Answer an open clarification (by message_id, as listed by clodex_handoff_get or clodex_messages). The asker's agent cannot answer its own question.", + "inputSchema": { + "type": "object", + "properties": {"run_id": {"type": "string"}, "message_id": {"type": "integer"}, "answer": {"type": "string"}, "actor": {"type": "string", "enum": ["claude", "codex"]}}, + "required": ["run_id", "message_id", "answer", "actor"], + }, + }, + { + "name": "clodex_messages", + "title": "List handoff messages", + "description": "List a handoff's clarification messages, optionally filtered by status (open, answered, delivered, final) or kind (clarification, answer).", + "inputSchema": { + "type": "object", + "properties": { + "run_id": {"type": "string"}, + "status": {"type": "string", "enum": ["open", "answered", "delivered", "final"]}, + "kind": {"type": "string", "enum": ["clarification", "answer"]}, + }, + "required": ["run_id"], + }, + }, { "name": "clodex_handoff_decide", "title": "Decide Clodex handoff", @@ -618,6 +653,26 @@ def tool_call(name: str, arguments: dict[str, Any], should_stop: Callable[[], bo except (KeyError, TypeError, ValueError, sqlite3.IntegrityError) as exc: return call_text(expected_handoff_error(exc), is_error=True) return call_json(run, is_error=run.get("status") == "blocked") + elif name == "clodex_clarify": + try: + message = workflow.state.add_clarification(str(arguments["run_id"]), str(arguments["actor"]), str(arguments["question"])) + except ValueError as exc: + return call_text(str(exc), is_error=True) + return call_json(message) + elif name == "clodex_answer": + try: + message = workflow.state.answer_clarification(str(arguments["run_id"]), int(arguments["message_id"]), str(arguments["answer"]), str(arguments["actor"])) + except ValueError as exc: + return call_text(str(exc), is_error=True) + return call_json(message) + elif name == "clodex_messages": + run_id = str(arguments["run_id"]) + if workflow.state.get_run(run_id) is None: + return call_text(f"Unknown run: {run_id}", is_error=True) + try: + return call_json({"messages": workflow.state.list_messages(run_id, arguments.get("status"), arguments.get("kind"))}) + except ValueError as exc: + return call_text(str(exc), is_error=True) elif name == "clodex_delegate": manager = DelegationManager(root) try: @@ -652,8 +707,11 @@ def tool_call(name: str, arguments: dict[str, Any], should_stop: Callable[[], bo if data is None: return call_text(f"Unknown run: {run_id}", is_error=True) + DelegationManager(root).reconcile_active(run_id) run = data["run"] status = run.get("status") + if status not in TERMINAL_STATUSES and workflow.state.active_delegation(run_id) is not None: + return call_text("a delegation is still running for this handoff; wait for it to finish before deciding", is_error=True) 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")} @@ -666,6 +724,19 @@ def tool_call(name: str, arguments: dict[str, Any], should_stop: Callable[[], bo "blocked_reason": run.get("blocked_reason") or run.get("error"), } is_error = True + elif data["open_clarifications"]: + # Reviewers may agree, but a question nobody has answered means the work is not settled. + decision = { + "decision": "needs_fix", + "run_id": run["id"], + "reason": "open clarifications must be answered first", + "open_clarifications": [{"message_id": m["id"], "asked_by": m["actor"], "question": m["body"]} for m in data["open_clarifications"]], + "next_expected_actor": data["next_expected_actor"], + "required_pending": agreement["required_pending"], + "approved_reviewers": agreement["approved_reviewers"], + "quorum": agreement["quorum"], + } + is_error = False elif agreement["approved"]: try: approved_run = workflow.state.approve_handoff(run["id"], agreement["diff_hash"], approved_by=agreement["approved_by"]) diff --git a/clodex/native.py b/clodex/native.py index f8c0c6a..863a52b 100644 --- a/clodex/native.py +++ b/clodex/native.py @@ -53,7 +53,7 @@ def build_claude_block() -> str: When a user asks for non-trivial design, implementation, debugging, refactoring, or review work: 1. Prefer the MCP tool `clodex_handoff_create` to create or resume a durable handoff. Pass `workspace: "git-worktree"` so Codex works in an isolated checkout. -2. Use `clodex_handoff_get` before acting on an existing handoff; it shows the budget, the next expected actor, and any running delegation. +2. Use `clodex_handoff_get` before acting on an existing handoff; it shows the budget, the next expected actor, any running delegation, and `open_clarifications`. Answer questions Codex asked with `clodex_answer`, then delegate again: answers are included in Codex's next prompt. 3. Produce the plan, design, acceptance criteria, risks, and test commands before implementation, and record them with `clodex_handoff_update`. 4. Delegate implementation to Codex with `clodex_delegate` (mode `implement`, `instructions` = the accepted plan) instead of telling the user to run Codex. Pass `wait: true`, or poll `clodex_handoff_get`. 5. Review the diff in the handoff's workspace, then record your verdict with `clodex_handoff_update` (actor `claude`, the diff hash from `clodex_handoff_get`, `report.approved`, `report.required_fixes`). @@ -94,7 +94,7 @@ def build_agents_block() -> str: 4. Use `clodex_handoff_update` to record changed files, tests run, unresolved issues, audit verdicts, and diff hash. 5. Use `clodex_handoff_create` only when the user asks Codex to start a native collaboration directly. 6. Use `clodex_handoff_decide` before claiming completion after implementation or audit work. -7. Ask for clarification through Clodex when product intent or acceptance criteria are unclear. +7. Ask for clarification with `clodex_clarify` instead of guessing when product intent or acceptance criteria are unclear (when Clodex runs you for a delegation, end your final message with a JSON object `{{"clarifications": ["..."]}}` instead). 8. Respect owner, phase, and handoff budget state. When the budget is exhausted, summarize the disagreement for the user. 9. When Clodex delegates work to you it runs `codex exec` in the handoff's workspace and records your result itself, so you do not need to call the handoff tools in that case; just do the work and finish with a concise report. diff --git a/clodex/prompts.py b/clodex/prompts.py index e7a9b2b..62583df 100644 --- a/clodex/prompts.py +++ b/clodex/prompts.py @@ -110,7 +110,7 @@ def audit_diff_excerpt(diff: str, max_bytes: int) -> str: ) -def delegate_prompt(mode: str, task: str, instructions: str | None, fixes: list[str] | None = None) -> str: +def delegate_prompt(mode: str, task: str, instructions: str | None, fixes: list[str] | None = None, answers: list[tuple[str, str]] | None = None) -> str: """The prompt for a Codex job a native handoff delegated (implement or fix).""" lines = [ "You are the Codex engineering wave for Clodex, working a native Claude/Codex handoff.", @@ -123,11 +123,19 @@ def delegate_prompt(mode: str, task: str, instructions: str | None, fixes: list[ ] if instructions: lines += ["", "Instructions from Claude:", instructions] + if answers: + lines += ["", "Clarifications from Claude (your earlier questions, now answered):"] + for question, answer in answers: + lines += [f"- Q: {question}", f" A: {answer}"] if mode == "fix": lines += ["", "Required fixes:", json.dumps(fixes or ["Resolve the open review findings."], indent=2), "", "Apply only the required fixes. Do not broaden scope."] lines += [ "", - "When finished, print a concise Markdown report with:", + "If you cannot proceed without a decision from Claude (product intent or acceptance criteria are unclear), " + "do not guess: stop and end your final message with a single JSON object " + '{"clarifications": ["one specific question", "..."]} and nothing after it.', + "", + "Otherwise, when finished, print a concise Markdown report with:", "- files changed", "- tests run and pass/fail status", "- unresolved issues, if any", diff --git a/clodex/state.py b/clodex/state.py index e7b6f62..4a6f0b0 100644 --- a/clodex/state.py +++ b/clodex/state.py @@ -10,6 +10,8 @@ TERMINAL_STATUSES = {"approved", "blocked", "failed", "cancelled", "applied", "completed"} HANDOFF_REASON_STATUSES = {"blocked", "failed"} +HANDOFF_ACTORS = ("claude", "codex") +MESSAGE_STATUSES = {"open", "answered", "delivered"} DELEGATION_MODES = {"implement", "fix", "audit"} DELEGATION_ACTIVE = {"queued", "running"} DELEGATION_FINISHED = {"completed", "failed", "cancelled"} @@ -30,6 +32,14 @@ def _int_or_default(value: Any, default: int) -> int: return default if value is None else int(value) +def normalize_actor(value: Any, what: str = "actor") -> str: + """`claude` or `codex` (case-insensitive); anything else is rejected rather than silently recorded.""" + actor = str(value).strip().lower() if value is not None else "" + if actor not in HANDOFF_ACTORS: + raise ValueError(f"unknown {what}: {value!r} (use claude or codex)") + return actor + + def _reason_text(value: Any) -> str | None: if value is None: return None @@ -154,12 +164,13 @@ def _init(self) -> None: """ ) self._ensure_run_columns(con) + self._ensure_message_columns(con) current = con.execute("select version from schema_version order by version desc limit 1").fetchone() if current is None: - con.execute("insert into schema_version(version) values (3)") - elif int(current["version"]) < 3: + con.execute("insert into schema_version(version) values (4)") + elif int(current["version"]) < 4: con.execute("delete from schema_version") - con.execute("insert into schema_version(version) values (3)") + con.execute("insert into schema_version(version) values (4)") def _ensure_run_columns(self, con: sqlite3.Connection) -> None: existing = {row["name"] for row in con.execute("pragma table_info(runs)")} @@ -182,6 +193,14 @@ def _ensure_run_columns(self, con: sqlite3.Connection) -> None: if name not in existing: con.execute(f"alter table runs add column {name} {kind}") + def _ensure_message_columns(self, con: sqlite3.Connection) -> None: + """v4: messages grow from free-form notes into the clarification channel of a handoff.""" + existing = {row["name"] for row in con.execute("pragma table_info(messages)")} + columns = {"run_id": "text", "actor": "text", "kind": "text", "reply_to": "integer", "status": "text"} + for name, kind in columns.items(): + if name not in existing: + con.execute(f"alter table messages add column {name} {kind}") + def table_names(self) -> set[str]: with self.session() as con: return {str(row["name"]) for row in con.execute("select name from sqlite_master where type='table'")} @@ -287,6 +306,7 @@ def create_handoff( 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)})") + owner = normalize_actor(owner, "owner") timestamp = now_iso() task_id = task_id or run_id # every handoff shows up in the task ledger (`clodex status`) @@ -337,6 +357,8 @@ def update_handoff( 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)})") + actor = normalize_actor(actor) if actor is not None else None + owner = normalize_actor(owner, "owner") if owner is not None else None with self.session() as con: con.execute("begin immediate") @@ -520,6 +542,7 @@ def get_handoff(self, run_id: str) -> dict[str, Any] | None: artifacts = [dict(artifact) for artifact in con.execute("select * from artifacts where run_id=? order by id", (run_id,))] delegations = [dict(item) for item in con.execute("select * from delegations where run_id=? order by id desc limit 5", (run_id,))] + open_clarifications = [dict(item) for item in con.execute("select * from messages where run_id=? and kind='clarification' and status='open' order by id", (run_id,))] handoff_count = _int_or_default(run.get("handoff_count"), 0) handoff_budget = _int_or_default(run.get("handoff_budget"), 6) last_actor = run.get("last_actor") @@ -529,11 +552,16 @@ def get_handoff(self, run_id: str) -> dict[str, Any] | None: next_expected_actor = "claude" else: next_expected_actor = run.get("owner") or "claude" + if open_clarifications: + # An open question hands the turn to whoever has to answer it. + asker = open_clarifications[-1].get("actor") + next_expected_actor = "claude" if asker == "codex" else "codex" if asker == "claude" else next_expected_actor return { "run": run, "events": events, "artifacts": artifacts, "delegations": delegations, + "open_clarifications": open_clarifications, "budget_remaining": max(handoff_budget - handoff_count, 0), "next_expected_actor": next_expected_actor, } @@ -555,6 +583,89 @@ def add_message(self, task_id: str | None, topic: str, body: str) -> None: (task_id, topic, body, now_iso()), ) + # ------------------------------------------------------------ clarifications + # Codex (or Claude) asks a question about a handoff; the other side answers; answers reach the + # next delegation. status: open -> answered -> delivered (included in a delegation's prompt). + + def _live_handoff(self, con: sqlite3.Connection, run_id: str) -> dict[str, Any]: + row = con.execute("select * from runs where id=?", (run_id,)).fetchone() + if row is None: + raise ValueError(f"unknown run: {run_id}") + if str(row["status"]) in TERMINAL_STATUSES: + raise ValueError(f"the handoff is {row['status']}; it cannot take new messages") + return dict(row) + + def add_clarification(self, run_id: str, actor: str, question: str) -> dict[str, Any]: + actor = normalize_actor(actor) + text = _reason_text(question) + if text is None: + raise ValueError("a clarification needs a question") + timestamp = now_iso() + with self.session() as con: + con.execute("begin immediate") + run = self._live_handoff(con, run_id) + cursor = con.execute( + "insert into messages(task_id, topic, body, created_at, run_id, actor, kind, status) values (?, ?, ?, ?, ?, ?, 'clarification', 'open')", + (run.get("task_id"), "clarification", text, timestamp, run_id, actor), + ) + self._insert_event(con, run_id, "handoff.clarify", {"message_id": cursor.lastrowid, "actor": actor, "question": text}, timestamp) + return dict(con.execute("select * from messages where id=?", (cursor.lastrowid,)).fetchone()) + + def answer_clarification(self, run_id: str, message_id: int, answer: str, actor: str) -> dict[str, Any]: + actor = normalize_actor(actor) + text = _reason_text(answer) + if text is None: + raise ValueError("an answer cannot be empty") + timestamp = now_iso() + with self.session() as con: + con.execute("begin immediate") + run = self._live_handoff(con, run_id) + question = con.execute("select * from messages where id=? and run_id=? and kind='clarification'", (message_id, run_id)).fetchone() + if question is None: + raise ValueError(f"unknown clarification {message_id} for this handoff") + if question["status"] != "open": + raise ValueError(f"clarification {message_id} was already answered") + if question["actor"] == actor: + raise ValueError(f"clarification {message_id} was asked by {actor}; the other agent answers it") + cursor = con.execute( + "insert into messages(task_id, topic, body, created_at, run_id, actor, kind, reply_to, status) values (?, ?, ?, ?, ?, ?, 'answer', ?, 'final')", + (run.get("task_id"), "answer", text, timestamp, run_id, actor, message_id), + ) + con.execute("update messages set status='answered' where id=?", (message_id,)) + self._insert_event(con, run_id, "handoff.answer", {"message_id": message_id, "answer_id": cursor.lastrowid, "actor": actor, "answer": text}, timestamp) + return dict(con.execute("select * from messages where id=?", (cursor.lastrowid,)).fetchone()) + + def list_messages(self, run_id: str, status: str | None = None, kind: str | None = None) -> list[dict[str, Any]]: + if status is not None and status not in MESSAGE_STATUSES | {"final"}: + raise ValueError(f"unknown message status: {status} (use one of {sorted(MESSAGE_STATUSES)})") + clauses, params = ["run_id=?"], [run_id] + if status is not None: + clauses.append("status=?") + params.append(status) + if kind is not None: + clauses.append("kind=?") + params.append(kind) + with self.session() as con: + return [dict(row) for row in con.execute(f"select * from messages where {' and '.join(clauses)} order by id", params)] + + def answered_clarifications(self, run_id: str) -> list[dict[str, Any]]: + """Question/answer pairs the next delegation has not been told about yet.""" + with self.session() as con: + rows = con.execute( + """ + select q.id as id, q.body as question, a.body as answer, q.actor as asked_by, a.actor as answered_by + from messages q join messages a on a.reply_to = q.id and a.kind = 'answer' + where q.run_id=? and q.kind='clarification' and q.status='answered' order by q.id + """, + (run_id,), + ) + return [dict(row) for row in rows] + + def mark_delivered(self, message_ids: list[int]) -> None: + with self.session() as con: + for message_id in message_ids: + con.execute("update messages set status='delivered' where id=? and kind='clarification' and status='answered'", (message_id,)) + def add_event(self, run_id: str, event: str, data: dict[str, Any]) -> None: with self.session() as con: self._insert_event(con, run_id, event, data, now_iso()) diff --git a/tests/support.py b/tests/support.py index 63a841a..4c1a13f 100644 --- a/tests/support.py +++ b/tests/support.py @@ -84,6 +84,7 @@ def __init__( codex_logged_in: bool = True, codex_catalog: dict | None = None, codex_fails: bool = False, + codex_clarifies: bool = False, ): self.reject_once = reject_once self.malformed_once = malformed_once @@ -98,6 +99,7 @@ def __init__( self.codex_logged_in = codex_logged_in self.codex_catalog = codex_catalog self.codex_fails = codex_fails + self.codex_clarifies = codex_clarifies self.sleep_seconds = sleep_seconds self.include_clodex = include_clodex @@ -320,6 +322,9 @@ def finish(text): print(text) raise SystemExit(0) + if {self.codex_clarifies!r} and 'adversarial auditor' not in stdin and 'Clarifications from Claude' not in stdin: + # Ask instead of guessing, until the answers show up in the prompt. + finish('I cannot proceed without a decision.\\n\\n' + json.dumps({{'clarifications': ['Which database should be used?', {{'question': 'Must it stay backwards compatible?'}}]}})) if 'adversarial auditor' in stdin: finish(audit_verdict(True, 'ok', [])) if 'Required fixes' in stdin: diff --git a/tests/test_clarification.py b/tests/test_clarification.py new file mode 100644 index 0000000..42c0291 --- /dev/null +++ b/tests/test_clarification.py @@ -0,0 +1,275 @@ +"""Tests: the handoff clarification channel, actor validation and decide guards.""" + +from __future__ import annotations + +import json +import os +import sqlite3 +import tempfile +import time +import unittest +from pathlib import Path + +from clodex.delegate import extract_clarifications +from clodex.prompts import delegate_prompt +from clodex.state import StateStore +from tests.support import FakeCliPath +from tests.test_delegate import DelegateCase +from tests.test_handoff_quorum import HandoffCase + + +class ExtractClarificationsTests(unittest.TestCase): + def test_finds_questions_in_bare_fenced_and_mixed_output(self): + self.assertEqual(extract_clarifications('{"clarifications": ["Which DB?"]}'), ["Which DB?"]) + fenced = 'Blocked.\n\n```json\n{"clarifications": ["A?", {"question": "B?"}]}\n```\n' + self.assertEqual(extract_clarifications(fenced), ["A?", "B?"]) + mixed = 'Report with code: `{"a": 1}`\n\nI need input.\n{"clarifications": ["Only the last JSON counts?"]}' + self.assertEqual(extract_clarifications(mixed), ["Only the last JSON counts?"]) + + def test_ignores_everything_that_is_not_a_clarification_request(self): + for text in ("", "all done", '{"clarifications": []}', '{"clarifications": "nope"}', '{"other": ["x"]}', '{"clarifications": [1, null, " ", {"q": "x"}]}', "{broken json", "[1, 2]"): + self.assertEqual(extract_clarifications(text), [], text) + + def test_blank_entries_are_dropped_and_the_count_is_capped(self): + many = {"clarifications": [" keep ", "", *[f"q{i}" for i in range(30)]]} + found = extract_clarifications(json.dumps(many)) + self.assertEqual(found[0], "keep") + self.assertEqual(len(found), 10) + + +class PromptTests(unittest.TestCase): + def test_prompt_tells_codex_how_to_ask_and_delivers_answers(self): + plain = delegate_prompt("implement", "Add search", "Use the existing index") + self.assertIn('{"clarifications": [', plain) + self.assertIn("do not guess", plain) + self.assertNotIn("Clarifications from Claude", plain) + answered = delegate_prompt("fix", "Add search", None, ["handle empty input"], [("Which DB?", "SQLite"), ("Compat?", "Yes")]) + self.assertIn("Clarifications from Claude", answered) + self.assertIn("- Q: Which DB?\n A: SQLite", answered) + self.assertLess(answered.index("Clarifications from Claude"), answered.index("Required fixes")) + + +class StateTests(unittest.TestCase): + def setUp(self): + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + self.store = StateStore(Path(self.tmp.name) / "s.sqlite3") + self.store.create_handoff("h", "Add search", owner="claude") + + def test_lifecycle_open_answered_delivered(self): + question = self.store.add_clarification("h", "codex", " Which DB? ") + self.assertEqual((question["kind"], question["status"], question["actor"], question["body"]), ("clarification", "open", "codex", "Which DB?")) + self.assertEqual(self.store.answered_clarifications("h"), [], "nothing to deliver before an answer") + answer = self.store.answer_clarification("h", question["id"], "SQLite", "claude") + self.assertEqual((answer["kind"], answer["reply_to"], answer["actor"]), ("answer", question["id"], "claude")) + self.assertEqual(self.store.list_messages("h", kind="clarification")[0]["status"], "answered") + pending = self.store.answered_clarifications("h") + self.assertEqual([(p["question"], p["answer"], p["asked_by"], p["answered_by"]) for p in pending], [("Which DB?", "SQLite", "codex", "claude")]) + self.store.mark_delivered([question["id"]]) + self.assertEqual(self.store.answered_clarifications("h"), []) + self.assertEqual(self.store.list_messages("h", kind="clarification")[0]["status"], "delivered") + events = [e["event"] for e in self.store.get_handoff("h")["events"]] + self.assertEqual(events[-2:], ["handoff.clarify", "handoff.answer"]) + + def test_rules(self): + question = self.store.add_clarification("h", "codex", "Q?") + with self.assertRaisesRegex(ValueError, "the other agent answers"): + self.store.answer_clarification("h", question["id"], "mine", "codex") + self.store.answer_clarification("h", question["id"], "A", "claude") + with self.assertRaisesRegex(ValueError, "already answered"): + self.store.answer_clarification("h", question["id"], "again", "claude") + with self.assertRaisesRegex(ValueError, "unknown clarification"): + self.store.answer_clarification("h", 9999, "x", "claude") + self.store.create_handoff("other", "t") + other = self.store.add_clarification("other", "claude", "Elsewhere?") + with self.assertRaisesRegex(ValueError, "unknown clarification"): + self.store.answer_clarification("h", other["id"], "cross-run", "codex") + for bad_question in ("", " "): + with self.assertRaises(ValueError): + self.store.add_clarification("h", "claude", bad_question) + with self.assertRaisesRegex(ValueError, "unknown run"): + self.store.add_clarification("nope", "claude", "Q?") + with self.assertRaisesRegex(ValueError, "empty"): + self.store.answer_clarification("h", self.store.add_clarification("h", "claude", "Q2?")["id"], " ", "codex") + + def test_finished_handoffs_take_no_messages(self): + self.store.update_handoff("h", actor="claude", status="blocked", blocked_reason="stop") + with self.assertRaisesRegex(ValueError, "cannot take new messages"): + self.store.add_clarification("h", "claude", "Q?") + + def test_open_questions_hand_the_turn_to_the_answerer(self): + self.assertEqual(self.store.get_handoff("h")["next_expected_actor"], "claude") + question = self.store.add_clarification("h", "claude", "Codex, which flag?") + data = self.store.get_handoff("h") + self.assertEqual((data["next_expected_actor"], [m["id"] for m in data["open_clarifications"]]), ("codex", [question["id"]])) + self.store.answer_clarification("h", question["id"], "--fast", "codex") + data = self.store.get_handoff("h") + self.assertEqual((data["next_expected_actor"], data["open_clarifications"]), ("claude", [])) + + def test_list_filters_and_validation(self): + q1 = self.store.add_clarification("h", "codex", "one") + self.store.add_clarification("h", "codex", "two") + self.store.answer_clarification("h", q1["id"], "ans", "claude") + self.assertEqual([m["body"] for m in self.store.list_messages("h", status="open")], ["two"]) + self.assertEqual([m["kind"] for m in self.store.list_messages("h", kind="answer")], ["answer"]) + self.assertEqual(len(self.store.list_messages("h")), 3) + with self.assertRaises(ValueError): + self.store.list_messages("h", status="pending") + + def test_actors_and_owners_are_validated_and_normalized(self): + self.store.update_handoff("h", actor=" Claude ") + self.assertEqual(self.store.get_run("h")["last_actor"], "claude") + for bad in ("bob", "", "gemini"): + with self.assertRaisesRegex(ValueError, "unknown actor"): + self.store.update_handoff("h", actor=bad) + with self.assertRaisesRegex(ValueError, "unknown owner"): + self.store.update_handoff("h", owner="nobody") + with self.assertRaisesRegex(ValueError, "unknown owner"): + self.store.create_handoff("x", "t", owner="nobody") + self.assertEqual(self.store.create_handoff("y", "t", owner="Codex")["owner"], "codex") + + def test_v3_database_upgrades_keeping_old_messages_readable(self): + db = Path(self.tmp.name) / "v3.sqlite3" + con = sqlite3.connect(db) + con.executescript( + """ + create table schema_version(version integer not null); insert into schema_version values (3); + create table messages(id integer primary key autoincrement, task_id text, topic text not null, body text not null, created_at text not null); + insert into messages(task_id, topic, body, created_at) values ('t', 'planning', 'old note', '2026-01-01T00:00:00Z'); + """ + ) + con.commit() + con.close() + store = StateStore(db) + self.assertEqual(store.schema_version(), 4) + with store.session() as conn: + columns = {row["name"] for row in conn.execute("pragma table_info(messages)")} + old = dict(conn.execute("select * from messages").fetchone()) + self.assertTrue({"run_id", "actor", "kind", "reply_to", "status"} <= columns) + self.assertEqual((old["body"], old["kind"], old["run_id"]), ("old note", None, None)) + store.create_handoff("h", "t") + self.assertEqual(store.add_clarification("h", "claude", "Q?")["status"], "open") + + +class ToolTests(HandoffCase): + def test_clarify_answer_and_messages_tools(self): + self.call("clodex_handoff_create", run_id="h", task="Add search") + is_error, message = self.call("clodex_clarify", run_id="h", actor="codex", question="Which DB?") + self.assertFalse(is_error, message) + self.assertEqual((message["status"], message["actor"]), ("open", "codex")) + is_error, listed = self.call("clodex_messages", run_id="h", status="open") + self.assertEqual([m["id"] for m in listed["messages"]], [message["id"]]) + is_error, answer = self.call("clodex_answer", run_id="h", message_id=message["id"], answer="SQLite", actor="claude") + self.assertFalse(is_error, answer) + self.assertEqual(self.call("clodex_messages", run_id="h", status="open")[1]["messages"], []) + self.assertEqual(len(self.call("clodex_messages", run_id="h")[1]["messages"]), 2) + data = self.call("clodex_handoff_get", run_id="h")[1] + self.assertEqual(data["open_clarifications"], []) + + def test_bad_input_is_a_tool_error(self): + self.call("clodex_handoff_create", run_id="h", task="t") + for name, arguments, expected in ( + ("clodex_clarify", {"run_id": "h", "actor": "claude", "question": " "}, "needs a question"), + ("clodex_clarify", {"run_id": "nope", "actor": "claude", "question": "Q?"}, "unknown run"), + ("clodex_answer", {"run_id": "h", "message_id": 99, "answer": "x", "actor": "claude"}, "unknown clarification"), + ("clodex_messages", {"run_id": "nope"}, "Unknown run: nope"), + ): + is_error, message = self.call(name, **arguments) + self.assertTrue(is_error, name) + self.assertIn(expected, message) + from clodex import mcp_server + + server = mcp_server.McpServer(out=object()) + for name, arguments, expected in ( + ("clodex_clarify", {"run_id": "h", "actor": "bob", "question": "Q?"}, "Invalid arguments"), + ("clodex_clarify", {"run_id": "h", "actor": "claude"}, "Missing required argument: question"), + ("clodex_answer", {"run_id": "h", "message_id": "one", "answer": "x", "actor": "claude"}, "Invalid arguments"), + ("clodex_messages", {"run_id": "h", "status": "pending"}, "Invalid arguments"), + ): + result = server.tools_call({"name": name, "arguments": arguments}) + self.assertTrue(result["isError"], (name, arguments)) + self.assertIn(expected, result["content"][0]["text"]) + + def test_unknown_actor_on_update_and_create_is_a_tool_error(self): + is_error, message = self.call("clodex_handoff_create", run_id="o", task="t", owner="bob") + self.assertTrue(is_error) + self.assertIn("unknown owner", message) + self.call("clodex_handoff_create", run_id="h", task="t") + is_error, message = self.call("clodex_handoff_update", run_id="h", actor="bob") + self.assertTrue(is_error) + self.assertIn("unknown actor", message) + + def test_decide_will_not_approve_over_an_open_question_and_does_once_answered(self): + self.call("clodex_handoff_create", run_id="h", task="t") + self.call("clodex_handoff_update", run_id="h", actor="claude", diff_hash="d", report={"approved": True}) + self.call("clodex_handoff_update", run_id="h", actor="codex", diff_hash="d", report={"approved": True}) + question = self.call("clodex_clarify", run_id="h", actor="codex", question="Is v2 in scope?")[1] + is_error, decision = self.call("clodex_handoff_decide", run_id="h") + self.assertFalse(is_error) + self.assertEqual(decision["decision"], "needs_fix", "the reviewers agree, but a question is open") + self.assertEqual(decision["open_clarifications"], [{"message_id": question["id"], "asked_by": "codex", "question": "Is v2 in scope?"}]) + self.assertEqual(decision["next_expected_actor"], "claude") + self.assertEqual(decision["approved_reviewers"], ["claude-plan", "codex-architecture"]) + self.call("clodex_answer", run_id="h", message_id=question["id"], answer="No", actor="claude") + self.assertEqual(self.call("clodex_handoff_decide", run_id="h")[1]["decision"], "approved") + + def test_decide_refuses_while_a_delegation_is_running(self): + self.call("clodex_handoff_create", run_id="h", task="t") + self.call("clodex_handoff_update", run_id="h", actor="claude", diff_hash="d", report={"approved": True}) + self.call("clodex_handoff_update", run_id="h", actor="codex", diff_hash="d", report={"approved": True}) + delegation = self.state.start_delegation("h", "implement", "x") + self.state.update_delegation(delegation["id"], status="running", pid=os.getpid(), started_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())) + self.state.touch_delegation(delegation["id"]) + is_error, message = self.call("clodex_handoff_decide", run_id="h") + self.assertTrue(is_error) + self.assertIn("delegation is still running", message) + self.state.update_delegation(delegation["id"], status="completed") + self.assertEqual(self.call("clodex_handoff_decide", run_id="h")[1]["decision"], "approved") + + +class WorkerFlowTests(DelegateCase): + def test_codex_asks_claude_answers_and_the_answers_reach_the_next_delegation(self): + with FakeCliPath(codex_clarifies=True): + self.create() + is_error, result = self.delegate(instructions="Add search") + self.assertFalse(is_error, result) + delegation = result["delegation"] + self.assertEqual(delegation["status"], "completed") + self.assertTrue(delegation["summary"].startswith("Codex needs clarification: Which database should be used?")) + self.assertFalse((self.workspace() / "implemented.txt").exists(), "Codex asked instead of guessing") + + data = self.handoff() + asked = data["open_clarifications"] + self.assertEqual([(m["actor"], m["body"]) for m in asked], [("codex", "Which database should be used?"), ("codex", "Must it stay backwards compatible?")]) + self.assertEqual(data["next_expected_actor"], "claude") + self.assertEqual(data["run"]["handoff_count"], 1, "handing the question back is a handoff") + report = next(e for e in reversed(data["events"]) if e["event"] == "handoff.update")["data"]["report"] + self.assertEqual(report["clarifications"], [m["body"] for m in asked]) + self.assertFalse(report["changed"]) + + self.assertEqual(self.call("clodex_handoff_decide", run_id="d")[1]["decision"], "needs_fix") + for message in asked: + self.assertFalse(self.call("clodex_answer", run_id="d", message_id=message["id"], answer="SQLite, and yes", actor="claude")[0]) + self.assertEqual(self.handoff()["open_clarifications"], []) + + is_error, second = self.delegate(instructions="Add search") + self.assertFalse(is_error, second) + self.assertEqual(second["delegation"]["status"], "completed") + self.assertTrue((self.workspace() / "implemented.txt").is_file(), "with the answers in the prompt Codex proceeded") + statuses = {m["id"]: m["status"] for m in self.call("clodex_messages", run_id="d", kind="clarification")[1]["messages"]} + self.assertEqual(set(statuses.values()), {"delivered"}) + self.assertEqual(self.handoff()["run"]["handoff_count"], 2) + + def test_answers_are_delivered_once(self): + with FakeCliPath(codex_clarifies=True): + self.create() + self.delegate(instructions="Add search") + for message in self.handoff()["open_clarifications"]: + self.call("clodex_answer", run_id="d", message_id=message["id"], answer="yes", actor="claude") + self.assertEqual(len(self.state.answered_clarifications("d")), 2) + self.delegate(instructions="Add search") + self.assertEqual(self.state.answered_clarifications("d"), [], "already delivered") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_delegate.py b/tests/test_delegate.py index 1d62777..d1c552f 100644 --- a/tests/test_delegate.py +++ b/tests/test_delegate.py @@ -349,7 +349,7 @@ def test_v2_database_upgrades_and_gains_the_table(self): con.commit() con.close() store = StateStore(db) - self.assertEqual(store.schema_version(), 3) + self.assertEqual(store.schema_version(), 4) self.assertIn("delegations", store.table_names()) def test_start_is_atomic_updates_are_sticky_and_listing_is_newest_first(self): diff --git a/tests/test_state.py b/tests/test_state.py index 672a293..14b6ffb 100644 --- a/tests/test_state.py +++ b/tests/test_state.py @@ -20,7 +20,7 @@ def test_state_migrations_add_v2_tables(self): self.assertIn("artifacts", tables) self.assertIn("workspace_locks", tables) self.assertIn("cancellations", tables) - self.assertEqual(store.schema_version(), 3) + self.assertEqual(store.schema_version(), 4) def test_state_migrations_add_native_handoff_columns(self): with tempfile.TemporaryDirectory() as tmp: