From c4fdaecee0e8ed365c34a683f14d83a66c05351e Mon Sep 17 00:00:00 2001 From: Devil Date: Wed, 30 Sep 2026 22:21:08 -0400 Subject: [PATCH] feat: structured agent output and per-role Claude settings - Ship plan / audit_verdict JSON schemas (OpenAI strict-mode shape, no $schema key: Claude's validator rejects the 2020-12 URI). Claude plan and audit pass them inline via --json-schema; Codex audits pass --output-schema and the final message is read back from `-o `. Verified against the real CLIs, including nested nullable fields. - Parsing prefers the envelope's structured_output, falls back to the result string, then validates against the schema (stdlib validator) and retries once with the specific problems. - Claude planner and auditor are now separate roles: plan = opus/max, audit = opus/high. claude.plan.* / claude.audit.* override; legacy flat claude.model/effort keys still drive both. Optional fallback_model, max_budget_usd and json_schema (auto|true|false; auto skips Windows .cmd shims). codex.audit.* overrides model/effort for Codex audits. - Fakes validate --json-schema/--output-schema like the real CLIs and emit structured_output; 16 new tests. Co-Authored-By: Claude Sonnet 5.5 --- CLODEX.md | 8 +- README.md | 10 +- clodex/agents.py | 78 +++++--- clodex/commands.py | 83 +++++--- clodex/config.py | 33 +++- clodex/prompts.py | 2 +- clodex/schemas.py | 88 +++++++++ clodex/schemas/audit_verdict.schema.json | 28 +++ clodex/schemas/plan.schema.json | 15 ++ clodex/workflow.py | 13 +- package.json | 1 + pyproject.toml | 3 + skills/clodex-workflow/SKILL.md | 3 +- tests/support.py | 31 ++- tests/test_config_commands.py | 17 +- tests/test_structured_output.py | 231 +++++++++++++++++++++++ 16 files changed, 584 insertions(+), 60 deletions(-) create mode 100644 clodex/schemas.py create mode 100644 clodex/schemas/audit_verdict.schema.json create mode 100644 clodex/schemas/plan.schema.json create mode 100644 tests/test_structured_output.py diff --git a/CLODEX.md b/CLODEX.md index 9c13dda..04cb9c6 100644 --- a/CLODEX.md +++ b/CLODEX.md @@ -9,9 +9,13 @@ workspace: backend: git-worktree apply_mode: manual claude: - model: opus - effort: max permission_mode: plan + plan: + model: opus + effort: max + audit: + model: opus + effort: high codex: model: gpt-6.1-sol reasoning_effort: xhigh diff --git a/README.md b/README.md index ae728ad..4d46efd 100644 --- a/README.md +++ b/README.md @@ -4,7 +4,7 @@ Clodex is a native collaboration layer for **Claude Code CLI** and **Codex CLI** It installs repo instructions and MCP tools that let either agent hand work to the other without the user manually driving every step. -- **Claude Code** plans first with Opus at max effort. +- **Claude Code** plans first with Opus at max effort and audits at high effort. - **Codex** implements from that accepted plan with GPT-6.1 Sol at xhigh reasoning. - **Both agents audit the same final diff hash** and must agree before the task is considered complete. @@ -132,9 +132,13 @@ prompt body. Defaults: ```yaml claude: - model: opus - effort: max permission_mode: plan + plan: + model: opus + effort: max + audit: + model: opus + effort: high codex: model: gpt-6.1-sol reasoning_effort: xhigh diff --git a/clodex/agents.py b/clodex/agents.py index a72b142..39919c0 100644 --- a/clodex/agents.py +++ b/clodex/agents.py @@ -1,7 +1,9 @@ from __future__ import annotations +import os import subprocess import shutil +import tempfile from dataclasses import dataclass from pathlib import Path @@ -15,11 +17,18 @@ class AgentResult: stderr: str returncode: int timed_out: bool = False + # Final agent message read from `-o ` when the command asked for it. + last_message: str | None = None @property def ok(self) -> bool: return self.returncode == 0 + @property + def output(self) -> str: + """The text to parse: the captured final message if any, else stdout.""" + return self.last_message if self.last_message else self.stdout + class AgentRunner: def __init__(self, repo_root: Path): @@ -30,32 +39,57 @@ def run(self, command: AgentCommand, prompt: str, timeout: int | None = None) -> resolved = shutil.which(argv[0]) if resolved: argv[0] = resolved + last_message_path: str | None = None + if command.capture_last_message: + handle, last_message_path = tempfile.mkstemp(prefix="clodex-last-message-", suffix=".txt") + os.close(handle) + # Keep the trailing "-" (read prompt from stdin) last. + at = len(argv) - 1 if argv and argv[-1] == "-" else len(argv) + argv[at:at] = ["-o", last_message_path] 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: + 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: + return AgentResult( + command=command, + stdout=_as_text(exc.stdout), + stderr=_as_text(exc.stderr) + f"\ntimed out after {timeout}s", + returncode=124, + timed_out=True, + ) return AgentResult( command=command, - stdout=_as_text(exc.stdout), - stderr=_as_text(exc.stderr) + f"\ntimed out after {timeout}s", - returncode=124, - timed_out=True, + stdout=result.stdout, + stderr=result.stderr, + returncode=result.returncode, + last_message=_read_text(last_message_path), ) - return AgentResult( - command=command, - stdout=result.stdout, - stderr=result.stderr, - returncode=result.returncode, - ) + finally: + if last_message_path: + try: + os.unlink(last_message_path) + except OSError: + pass + + +def _read_text(path: str | None) -> str | None: + if not path: + return None + try: + text = Path(path).read_text(encoding="utf-8", errors="replace") + except OSError: + return None + return text.strip() or None def _as_text(value: str | bytes | None) -> str: diff --git a/clodex/commands.py b/clodex/commands.py index f37bbc3..c95e0b0 100644 --- a/clodex/commands.py +++ b/clodex/commands.py @@ -1,18 +1,28 @@ from __future__ import annotations +import shutil from dataclasses import dataclass from pathlib import Path from .config import ClodexConfig +from .schemas import compact_schema, schema_path @dataclass(frozen=True) class AgentCommand: name: str argv: list[str] + # When set, the agent's JSON output is validated against this packaged schema. + schema_name: str | None = None + # When set, the runner adds `-o ` so the final message is read from a file + # instead of being scraped out of stdout (codex exec). + capture_last_message: bool = False def display(self) -> str: - return " ".join(quote_arg(arg) for arg in self.argv) + shown: list[str] = [] + for index, arg in enumerate(self.argv): + shown.append(f"<{self.schema_name} schema>" if index and self.argv[index - 1] == "--json-schema" else arg) + return " ".join(quote_arg(arg) for arg in shown) def quote_arg(arg: str) -> str: @@ -23,27 +33,53 @@ def quote_arg(arg: str) -> str: return arg +def inline_schema_supported(setting: object = "auto", resolved: str | None = None) -> bool: + """Whether to pass `claude --json-schema `. + + The flag takes inline JSON only. A Windows .cmd/.bat shim re-parses the command + line with cmd.exe and mangles the quotes, so `auto` turns the flag off for shims. + """ + if isinstance(setting, bool): + return setting + text = str(setting).strip().lower() + if text in {"true", "yes", "on"}: + return True + if text in {"false", "no", "off"}: + return False + resolved = resolved or shutil.which("claude") + return not (resolved and resolved.lower().endswith((".cmd", ".bat"))) + + +def _claude_command(name: str, config: ClodexConfig, role: str, schema_name: str) -> AgentCommand: + settings = config.claude_role(role) + argv = [ + "claude", + "-p", + "--model", + str(settings["model"]), + "--effort", + str(settings["effort"]), + "--permission-mode", + str(settings["permission_mode"]), + "--output-format", + "json", + ] + fallback = settings.get("fallback_model") + if fallback: + argv.extend(["--fallback-model", ",".join(fallback) if isinstance(fallback, (list, tuple)) else str(fallback)]) + if settings.get("max_budget_usd") not in (None, ""): + argv.extend(["--max-budget-usd", str(settings["max_budget_usd"])]) + if inline_schema_supported(settings.get("json_schema", "auto")): + argv.extend(["--json-schema", compact_schema(schema_name)]) + return AgentCommand(name=name, argv=argv, schema_name=schema_name) + + def claude_plan_command(config: ClodexConfig) -> AgentCommand: - claude = config.claude - return AgentCommand( - name="claude-plan", - argv=[ - "claude", - "-p", - "--model", - str(claude["model"]), - "--effort", - str(claude["effort"]), - "--permission-mode", - str(claude["permission_mode"]), - "--output-format", - "json", - ], - ) + return _claude_command("claude-plan", config, "plan", "plan") def claude_audit_command(config: ClodexConfig) -> AgentCommand: - return claude_plan_command(config) + return _claude_command("claude-audit", config, "audit", "audit_verdict") def codex_exec_command(config: ClodexConfig, repo_root: Path, approval_profile: str | None = None) -> AgentCommand: @@ -67,9 +103,10 @@ def codex_exec_command(config: ClodexConfig, repo_root: Path, approval_profile: def codex_review_command(config: ClodexConfig, repo_root: Path) -> AgentCommand: - # `codex review --uncommitted` rejects a prompt argument, so audits run as a - # read-only `codex exec`; the audit prompt already embeds the diff. - codex = config.codex + # `codex exec review` cannot take a prompt together with --uncommitted and ignores + # --output-schema, so audits run as a read-only `codex exec`; the audit prompt + # already embeds the diff. + codex = config.codex_role("audit") return AgentCommand( name="codex-audit", argv=[ @@ -86,6 +123,10 @@ def codex_review_command(config: ClodexConfig, repo_root: Path) -> AgentCommand: "--ephemeral", "-C", str(repo_root), + "--output-schema", + str(schema_path("audit_verdict")), "-", ], + schema_name="audit_verdict", + capture_last_message=True, ) diff --git a/clodex/config.py b/clodex/config.py index 995aa49..200c395 100644 --- a/clodex/config.py +++ b/clodex/config.py @@ -24,9 +24,10 @@ "apply_mode": "manual", }, "claude": { - "model": "opus", - "effort": "max", "permission_mode": "plan", + # Planning gets the most reasoning; audits run often, so they default lower. + "plan": {"model": "opus", "effort": "max"}, + "audit": {"model": "opus", "effort": "high"}, }, "codex": { "model": "gpt-6.1-sol", @@ -55,6 +56,9 @@ } +ROLES = ("plan", "audit") + + class ConfigError(ValueError): """CLODEX.md could not be loaded.""" @@ -86,6 +90,9 @@ class ClodexConfig: repo_root: Path raw: dict[str, Any] = field(default_factory=dict) prompt_body: str = "" + # Only what CLODEX.md set explicitly (no defaults), so legacy flat keys can + # outrank the per-role defaults. + user: dict[str, Any] = field(default_factory=dict) @property def max_fix_loops(self) -> int: @@ -111,10 +118,30 @@ def workspace(self) -> dict[str, Any]: def claude(self) -> dict[str, Any]: return dict(DEFAULT_CONFIG["claude"] | self.raw.get("claude", {})) + def claude_role(self, role: str) -> dict[str, Any]: + """Effective Claude settings for `plan` or `audit`. + + Precedence, lowest to highest: built-in role defaults, flat `claude.*` keys in + CLODEX.md (the pre-0.2 layout, which drove both roles), then `claude..*`. + """ + defaults = DEFAULT_CONFIG["claude"] + user = self.user.get("claude") + user = user if isinstance(user, dict) else {} + flat = {key: value for key, value in user.items() if key not in ROLES} + explicit = user.get(role) + explicit = explicit if isinstance(explicit, dict) else {} + return {"permission_mode": defaults["permission_mode"], **defaults[role], **flat, **explicit} + @property def codex(self) -> dict[str, Any]: return dict(DEFAULT_CONFIG["codex"] | self.raw.get("codex", {})) + def codex_role(self, role: str) -> dict[str, Any]: + """Codex settings for a role; `codex.audit.*` overrides model/effort for audits.""" + base = {key: value for key, value in self.codex.items() if key not in ROLES} + override = self.codex.get(role) + return {**base, **(override if isinstance(override, dict) else {})} + @property def audit(self) -> dict[str, Any]: return dict(DEFAULT_CONFIG["audit"] | self.raw.get("audit", {})) @@ -152,7 +179,7 @@ def load_config(repo_root: Path | None = None) -> ClodexConfig: parsed = parse_front_matter(front_matter, str(contract)) merged = deep_merge(DEFAULT_CONFIG, parsed) warn_if_model_retiring(str(merged["codex"].get("model", ""))) - return ClodexConfig(repo_root=root, raw=merged, prompt_body=body.strip()) + return ClodexConfig(repo_root=root, raw=merged, prompt_body=body.strip(), user=parsed) def split_front_matter(text: str) -> tuple[str, str]: diff --git a/clodex/prompts.py b/clodex/prompts.py index 3c5e3b2..454de3c 100644 --- a/clodex/prompts.py +++ b/clodex/prompts.py @@ -73,7 +73,7 @@ def audit_prompt(agent_name: str, plan: dict[str, Any], diff: str, diff_hash: st "persona": "{selected_persona}", "summary": "short verdict", "findings": [ - {{"severity": "high|medium|low", "file": "path", "line": 1, "message": "finding"}} + {{"severity": "critical|high|medium|low|info", "file": "path or null", "line": 1, "message": "finding"}} ], "required_fixes": ["specific fix"] }} diff --git a/clodex/schemas.py b/clodex/schemas.py new file mode 100644 index 0000000..f7a5b51 --- /dev/null +++ b/clodex/schemas.py @@ -0,0 +1,88 @@ +"""JSON schemas for structured agent output, plus a small stdlib validator.""" + +from __future__ import annotations + +import json +from functools import lru_cache +from pathlib import Path +from typing import Any + +SCHEMA_DIR = Path(__file__).resolve().parent / "schemas" +SCHEMA_NAMES = ("plan", "audit_verdict") + + +class SchemaValidationError(ValueError): + """Agent output parsed as JSON but did not match the required schema.""" + + +def schema_path(name: str) -> Path: + """Path to a packaged schema file (Codex's --output-schema needs a real file).""" + if name not in SCHEMA_NAMES: + raise ValueError(f"Unknown schema: {name}") + return SCHEMA_DIR / f"{name}.schema.json" + + +@lru_cache(maxsize=None) +def load_schema(name: str) -> dict[str, Any]: + return json.loads(schema_path(name).read_text(encoding="utf-8")) + + +def compact_schema(name: str) -> str: + """Single-line JSON for `claude --json-schema` (which only accepts inline JSON).""" + return json.dumps(load_schema(name), separators=(",", ":")) + + +def validate(value: Any, schema: dict[str, Any], path: str = "$") -> list[str]: + """Check type, enum, required and nested items/properties. + + Deliberately lenient about extra properties: the CLIs enforce the strict + schema themselves, and an unexpected key is not worth a retry. + """ + expected = schema.get("type") + if expected is not None: + types = expected if isinstance(expected, list) else [expected] + if not any(_is_type(value, kind) for kind in types): + return [f"{path}: expected {' or '.join(types)}, got {_type_name(value)}"] + errors: list[str] = [] + if "enum" in schema and value not in schema["enum"]: + errors.append(f"{path}: {value!r} is not one of {schema['enum']}") + if isinstance(value, dict): + properties = schema.get("properties", {}) + for key in schema.get("required", []): + if key not in value: + errors.append(f"{path}.{key}: missing required property") + for key, subschema in properties.items(): + if key in value: + errors.extend(validate(value[key], subschema, f"{path}.{key}")) + elif isinstance(value, list) and "items" in schema: + for index, item in enumerate(value): + errors.extend(validate(item, schema["items"], f"{path}[{index}]")) + return errors + + +def check(value: Any, name: str) -> None: + errors = validate(value, load_schema(name)) + if errors: + raise SchemaValidationError("; ".join(errors[:8])) + + +def _is_type(value: Any, kind: str) -> bool: + if kind == "string": + return isinstance(value, str) + if kind == "boolean": + return isinstance(value, bool) + if kind == "integer": + return isinstance(value, int) and not isinstance(value, bool) + if kind == "number": + return isinstance(value, (int, float)) and not isinstance(value, bool) + if kind == "array": + return isinstance(value, list) + if kind == "object": + return isinstance(value, dict) + if kind == "null": + return value is None + return True + + +def _type_name(value: Any) -> str: + return "null" if value is None else type(value).__name__ diff --git a/clodex/schemas/audit_verdict.schema.json b/clodex/schemas/audit_verdict.schema.json new file mode 100644 index 0000000..b4f6a8a --- /dev/null +++ b/clodex/schemas/audit_verdict.schema.json @@ -0,0 +1,28 @@ +{ + "title": "Clodex audit verdict", + "type": "object", + "additionalProperties": false, + "required": ["approved", "diff_hash", "reviewer_id", "persona", "summary", "findings", "required_fixes"], + "properties": { + "approved": {"type": "boolean"}, + "diff_hash": {"type": "string"}, + "reviewer_id": {"type": "string"}, + "persona": {"type": "string"}, + "summary": {"type": "string"}, + "findings": { + "type": "array", + "items": { + "type": "object", + "additionalProperties": false, + "required": ["severity", "file", "line", "message"], + "properties": { + "severity": {"type": "string", "enum": ["critical", "high", "medium", "low", "info"]}, + "file": {"type": ["string", "null"]}, + "line": {"type": ["integer", "null"]}, + "message": {"type": "string"} + } + } + }, + "required_fixes": {"type": "array", "items": {"type": "string"}} + } +} diff --git a/clodex/schemas/plan.schema.json b/clodex/schemas/plan.schema.json new file mode 100644 index 0000000..3728495 --- /dev/null +++ b/clodex/schemas/plan.schema.json @@ -0,0 +1,15 @@ +{ + "title": "Clodex plan", + "type": "object", + "additionalProperties": false, + "required": ["goal", "scope", "out_of_scope", "implementation_spec", "acceptance_criteria", "risks", "test_commands"], + "properties": { + "goal": {"type": "string"}, + "scope": {"type": "array", "items": {"type": "string"}}, + "out_of_scope": {"type": "array", "items": {"type": "string"}}, + "implementation_spec": {"type": "array", "items": {"type": "string"}}, + "acceptance_criteria": {"type": "array", "items": {"type": "string"}}, + "risks": {"type": "array", "items": {"type": "string"}}, + "test_commands": {"type": "array", "items": {"type": "string"}} + } +} diff --git a/clodex/workflow.py b/clodex/workflow.py index f8c179e..ec65ecc 100644 --- a/clodex/workflow.py +++ b/clodex/workflow.py @@ -12,6 +12,7 @@ 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 .schemas import SchemaValidationError, check as check_schema from .state import StateStore from .trace import TraceWriter from .workspace import DirtyWorkspaceError, WorkspaceManager, WorkspaceRef @@ -320,11 +321,21 @@ def _run_json_with_retry( if not result.ok: raise RuntimeError(f"{label} failed with exit code {result.returncode}: {result.stderr.strip()}") try: - return extract_json_object(result.stdout) + data = extract_json_object(result.output) + if command.schema_name: + check_schema(data, command.schema_name) + return data except AgentEnvelopeError as exc: last_error = str(exc) if _attempt == 1: raise RuntimeError(f"{label} returned an error: {last_error}") from exc + except SchemaValidationError as exc: + last_error = str(exc) + current_prompt = ( + prompt + + "\n\nYour previous response did not match the required JSON schema. " + + f"Return only a corrected JSON object. Problems: {last_error}" + ) except ValueError as exc: last_error = str(exc) current_prompt = ( diff --git a/package.json b/package.json index 43ce3c1..5008330 100644 --- a/package.json +++ b/package.json @@ -26,6 +26,7 @@ }, "files": [ "clodex/**/*.py", + "clodex/schemas/*.json", "clodex/_vendor/VENDORED.md", "clodex/_vendor/yaml/LICENSE", "npm/*.js", diff --git a/pyproject.toml b/pyproject.toml index aa92816..edd7a8c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -22,3 +22,6 @@ version = {attr = "clodex.__version__"} include = ["clodex*"] # pip installs depend on the real PyYAML; the vendored copy is for npm installs only. exclude = ["clodex._vendor*"] + +[tool.setuptools.package-data] +clodex = ["schemas/*.json"] diff --git a/skills/clodex-workflow/SKILL.md b/skills/clodex-workflow/SKILL.md index 5b4ee7d..2d5c7fb 100644 --- a/skills/clodex-workflow/SKILL.md +++ b/skills/clodex-workflow/SKILL.md @@ -21,7 +21,8 @@ Harness commands: Defaults: -- Claude planner/auditor: `claude -p --model opus --effort max --permission-mode plan`. +- Claude planner: `claude -p --model opus --effort max --permission-mode plan` (JSON via `--json-schema`). +- Claude auditor: the same with `--effort high`. Set `claude.plan.*` / `claude.audit.*` in `CLODEX.md` to change either. - Codex engineer/auditor: `codex exec` (builds and audits) with `gpt-6.1-sol` and `model_reasoning_effort="xhigh"`. - Builds default to `.clodex/workspaces//`; use `--workspace local` only when in-place changes are intentional. - Approval profiles are `ci`, `local`, and `auto_review`; `ci` is the deterministic default. diff --git a/tests/support.py b/tests/support.py index 1f68a2b..9d2296b 100644 --- a/tests/support.py +++ b/tests/support.py @@ -62,10 +62,14 @@ def __init__( sleep_seconds: float = 0, include_clodex: bool = False, envelope_error_once: bool = False, + no_structured_output: bool = False, + schema_violation_once: bool = False, ): 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.sleep_seconds = sleep_seconds self.include_clodex = include_clodex @@ -166,11 +170,27 @@ def audit_verdict(approved, summary, fixes): if flags.get('--permission-mode', ['plan'])[-1] not in ('acceptEdits', 'auto', 'bypassPermissions', 'manual', 'dontAsk', 'plan'): fail('error: invalid --permission-mode') as_json = flags.get('--output-format', ['text'])[-1] == 'json' + schema_arg = flags.get('--json-schema', [None])[-1] + if schema_arg is not None: + # The real flag takes inline JSON only (not a path) and its validator + # does not know the draft 2020-12 `$schema` URI. + try: + schema_doc = json.loads(schema_arg) + except ValueError: + fail('Error: --json-schema is not valid JSON') + if '$schema' in schema_doc: + fail('Error: --json-schema is not a valid JSON Schema: no schema with key or ref ' + repr(schema_doc['$schema'])) def emit(text, is_error=False): if as_json: # The real `claude -p --output-format json` wraps the answer in an envelope. - print(json.dumps({{'type': 'result', 'subtype': 'success', 'is_error': is_error, 'result': text, 'total_cost_usd': 0, 'session_id': 'fake'}})) + envelope = {{'type': 'result', 'subtype': 'success', 'is_error': is_error, 'result': text, 'total_cost_usd': 0, 'session_id': 'fake'}} + if schema_arg is not None and not is_error and {str(self.no_structured_output)!r} != 'True': + try: + envelope['structured_output'] = json.loads(text) + except ValueError: + pass + print(json.dumps(envelope)) else: print(text) raise SystemExit(0) @@ -183,6 +203,10 @@ def emit(text, is_error=False): if {str(self.malformed_once)!r} == 'True' and not marker.exists(): marker.write_text('seen') emit('not json') + violation_marker = Path('.fake-claude-schema-violation') + if {str(self.schema_violation_once)!r} == 'True' and 'adversarial auditor' not in stdin and not violation_marker.exists(): + violation_marker.write_text('seen') + emit(json.dumps({{'goal': 'only a goal'}})) if 'adversarial auditor' in stdin: reject_marker = Path('.fake-claude-reject') if {str(self.reject_once)!r} == 'True' and not reject_marker.exists(): @@ -210,6 +234,11 @@ def emit(text, is_error=False): fail('error: invalid approval_policy ' + value) elif key != 'model': fail('error: unknown config key ' + key) + if '--output-schema' in flags: + schema_file = Path(flags['--output-schema'][-1]) + if not schema_file.is_file(): + fail('error: --output-schema file not found: ' + str(schema_file)) + json.loads(schema_file.read_text(encoding='utf-8')) sandbox = (flags.get('-s') or flags.get('--sandbox') or ['read-only'])[-1] if sandbox not in ('read-only', 'workspace-write', 'danger-full-access'): fail('error: invalid sandbox ' + sandbox) diff --git a/tests/test_config_commands.py b/tests/test_config_commands.py index aa44636..f7aa806 100644 --- a/tests/test_config_commands.py +++ b/tests/test_config_commands.py @@ -12,6 +12,7 @@ from clodex.agents import AgentRunner from clodex.commands import ( AgentCommand, + claude_audit_command, claude_plan_command, codex_exec_command, codex_review_command, @@ -24,8 +25,10 @@ class ConfigCommandTests(unittest.TestCase): def test_config_and_commands_use_requested_defaults(self): with TempRepo() as repo: config = load_config(repo) - self.assertEqual(config.claude["model"], "opus") - self.assertEqual(config.claude["effort"], "max") + self.assertEqual(config.claude_role("plan")["model"], "opus") + self.assertEqual(config.claude_role("plan")["effort"], "max") + self.assertEqual(config.claude_role("audit")["model"], "opus") + self.assertEqual(config.claude_role("audit")["effort"], "high") self.assertEqual(config.codex["model"], "gpt-6.1-sol") self.assertEqual(config.codex["reasoning_effort"], "xhigh") self.assertEqual(config.workspace["backend"], "git-worktree") @@ -34,7 +37,11 @@ def test_config_and_commands_use_requested_defaults(self): self.assertTrue(config.mcp["async_tasks"]) self.assertTrue(config.tracing["enabled"]) self.assertGreaterEqual(len(config.reviewers), 2) - self.assertIn("--permission-mode", claude_plan_command(config).argv) + plan_argv = claude_plan_command(config).argv + audit_argv_claude = claude_audit_command(config).argv + self.assertEqual(plan_argv[plan_argv.index("--effort") + 1], "max") + self.assertEqual(audit_argv_claude[audit_argv_claude.index("--effort") + 1], "high") + self.assertEqual(audit_argv_claude[audit_argv_claude.index("--permission-mode") + 1], "plan") self.assertIn("model_reasoning_effort=\"xhigh\"", codex_exec_command(config, repo).argv) audit_argv = codex_review_command(config, repo).argv self.assertEqual(audit_argv[:2], ["codex", "exec"]) @@ -97,8 +104,8 @@ def test_front_matter_supports_nesting_and_keeps_defaults_for_empty_sections(sel ) config = load_config(repo) self.assertEqual(config.claude["plan"], {"model": "fable", "effort": "max"}) - self.assertEqual(config.claude["audit"]["effort"], "high") - self.assertEqual(config.claude["model"], "opus") + self.assertEqual(config.claude_role("plan")["model"], "fable") + self.assertEqual(config.claude_role("audit")["effort"], "high") self.assertEqual(config.codex["reasoning_effort"], "high") self.assertEqual(config.codex["model"], "gpt-6.1-sol") self.assertTrue(config.mcp["async_tasks"]) diff --git a/tests/test_structured_output.py b/tests/test_structured_output.py new file mode 100644 index 0000000..e89b27c --- /dev/null +++ b/tests/test_structured_output.py @@ -0,0 +1,231 @@ +"""Tests: structured agent output, JSON schemas and per-role model settings.""" + +from __future__ import annotations + +import json +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +from clodex.agents import AgentRunner +from clodex.commands import ( + AgentCommand, + claude_audit_command, + claude_plan_command, + codex_exec_command, + codex_review_command, + inline_schema_supported, +) +from clodex.config import load_config +from clodex.schemas import SCHEMA_NAMES, load_schema, schema_path, validate +from clodex.workflow import ClodexWorkflow +from tests.support import ROOT, FakeCliPath, TempRepo + + +def write_contract(repo: Path, front_matter: str) -> None: + (repo / "CLODEX.md").write_text(f"---\n{front_matter}---\nbody\n", encoding="utf-8") + + +def flag_value(argv: list[str], flag: str) -> str: + return argv[argv.index(flag) + 1] + + +def assert_strict(testcase: unittest.TestCase, schema: dict, path: str = "$") -> None: + """OpenAI strict mode: every property required, no extra properties, recursively.""" + if schema.get("type") == "object": + testcase.assertIs(schema.get("additionalProperties"), False, path) + testcase.assertEqual(set(schema.get("required", [])), set(schema.get("properties", {})), path) + for name, sub in schema.get("properties", {}).items(): + assert_strict(testcase, sub, f"{path}.{name}") + if schema.get("type") == "array": + assert_strict(testcase, schema["items"], f"{path}[]") + + +COMPLETE_PLAN = { + "goal": "g", + "scope": ["s"], + "out_of_scope": [], + "implementation_spec": ["step"], + "acceptance_criteria": ["ok"], + "risks": [], + "test_commands": ["pytest"], +} + + +class SchemaTests(unittest.TestCase): + def test_schema_files_load_and_are_strict_mode_compatible(self): + for name in SCHEMA_NAMES: + schema = load_schema(name) + self.assertTrue(schema_path(name).is_file(), name) + # Claude's validator rejects the draft 2020-12 `$schema` URI. + self.assertNotIn("$schema", schema, name) + assert_strict(self, schema) + + def test_plan_validation(self): + schema = load_schema("plan") + self.assertEqual(validate(COMPLETE_PLAN, schema), []) + incomplete = {key: value for key, value in COMPLETE_PLAN.items() if key != "risks"} + self.assertIn("$.risks: missing required property", validate(incomplete, schema)) + wrong_type = {**COMPLETE_PLAN, "scope": "not a list"} + self.assertTrue(any("$.scope" in error for error in validate(wrong_type, schema))) + + def test_verdict_validation_allows_null_location_and_checks_enum_and_bool(self): + schema = load_schema("audit_verdict") + verdict = { + "approved": True, + "diff_hash": "h", + "reviewer_id": "r", + "persona": "p", + "summary": "s", + "findings": [{"severity": "low", "file": None, "line": None, "message": "m"}], + "required_fixes": [], + } + self.assertEqual(validate(verdict, schema), []) + bad_severity = {**verdict, "findings": [{"severity": "huge", "file": "f", "line": 3, "message": "m"}]} + self.assertTrue(any("severity" in error for error in validate(bad_severity, schema))) + self.assertTrue(any("approved" in error for error in validate({**verdict, "approved": "yes"}, schema))) + self.assertTrue(any("line" in error for error in validate({**verdict, "findings": [{"severity": "low", "file": None, "line": True, "message": "m"}]}, schema))) + + def test_package_ships_schema_files(self): + package = json.loads((ROOT / "package.json").read_text(encoding="utf-8")) + self.assertIn("clodex/schemas/*.json", package["files"]) + self.assertIn("schemas/*.json", (ROOT / "pyproject.toml").read_text(encoding="utf-8")) + + +class RoleConfigTests(unittest.TestCase): + def test_legacy_flat_claude_keys_drive_both_roles_and_role_keys_win(self): + with TempRepo() as repo: + write_contract(repo, "claude:\n model: sonnet\n effort: low\n") + config = load_config(repo) + for role in ("plan", "audit"): + self.assertEqual(config.claude_role(role)["model"], "sonnet") + self.assertEqual(config.claude_role(role)["effort"], "low") + write_contract(repo, "claude:\n model: sonnet\n effort: low\n audit:\n effort: medium\n") + config = load_config(repo) + self.assertEqual(config.claude_role("plan")["effort"], "low") + self.assertEqual(config.claude_role("audit")["effort"], "medium") + self.assertEqual(config.claude_role("audit")["model"], "sonnet") + + def test_codex_audit_overrides_do_not_affect_builds(self): + with TempRepo() as repo: + write_contract(repo, "codex:\n model: gpt-6-sol\n reasoning_effort: xhigh\n audit:\n model: gpt-6-luna\n reasoning_effort: high\n") + config = load_config(repo) + build = codex_exec_command(config, repo).argv + audit = codex_review_command(config, repo).argv + self.assertEqual(flag_value(build, "-m"), "gpt-6-sol") + self.assertIn('model_reasoning_effort="xhigh"', build) + self.assertEqual(flag_value(audit, "-m"), "gpt-6-luna") + self.assertIn('model_reasoning_effort="high"', audit) + + def test_fallback_model_and_budget_flags_are_optional(self): + with TempRepo() as repo: + plain = claude_plan_command(load_config(repo)).argv + self.assertNotIn("--fallback-model", plain) + self.assertNotIn("--max-budget-usd", plain) + write_contract(repo, "claude:\n fallback_model: [sonnet, haiku]\n max_budget_usd: 2.5\n") + config = load_config(repo) + for argv in (claude_plan_command(config).argv, claude_audit_command(config).argv): + self.assertEqual(flag_value(argv, "--fallback-model"), "sonnet,haiku") + self.assertEqual(flag_value(argv, "--max-budget-usd"), "2.5") + + +class CommandTests(unittest.TestCase): + def test_inline_schema_policy(self): + self.assertFalse(inline_schema_supported("auto", "C:/tools/claude.cmd")) + self.assertFalse(inline_schema_supported("auto", "C:/tools/CLAUDE.BAT")) + self.assertTrue(inline_schema_supported("auto", "C:/tools/claude.exe")) + self.assertTrue(inline_schema_supported("auto", "/usr/local/bin/claude")) + self.assertTrue(inline_schema_supported(True, "C:/tools/claude.cmd")) + self.assertTrue(inline_schema_supported("true", "C:/tools/claude.cmd")) + self.assertFalse(inline_schema_supported(False, "/usr/local/bin/claude")) + self.assertFalse(inline_schema_supported("off", "/usr/local/bin/claude")) + + def test_json_schema_flag_is_compact_valid_json_and_hidden_in_display(self): + with TempRepo() as repo: + write_contract(repo, "claude:\n json_schema: true\n") + config = load_config(repo) + plan = claude_plan_command(config) + audit = claude_audit_command(config) + self.assertEqual(json.loads(flag_value(plan.argv, "--json-schema")), load_schema("plan")) + self.assertEqual(json.loads(flag_value(audit.argv, "--json-schema")), load_schema("audit_verdict")) + self.assertNotIn("\n", flag_value(plan.argv, "--json-schema")) + self.assertIn("", plan.display()) + self.assertNotIn("additionalProperties", plan.display()) + write_contract(repo, "claude:\n json_schema: false\n") + self.assertNotIn("--json-schema", claude_plan_command(load_config(repo)).argv) + + def test_codex_audit_command_uses_output_schema_file_and_captures_last_message(self): + with TempRepo() as repo: + command = codex_review_command(load_config(repo), repo) + self.assertEqual(command.argv[-1], "-") + self.assertEqual(Path(flag_value(command.argv, "--output-schema")), schema_path("audit_verdict")) + self.assertEqual(command.schema_name, "audit_verdict") + self.assertTrue(command.capture_last_message) + self.assertFalse(codex_exec_command(load_config(repo), repo).capture_last_message) + + +class RunnerTests(unittest.TestCase): + def test_runner_reads_final_message_from_output_file_and_cleans_up(self): + created: list[str] = [] + real_mkstemp = tempfile.mkstemp + + def tracking_mkstemp(*args, **kwargs): + handle, path = real_mkstemp(*args, **kwargs) + created.append(path) + return handle, path + + with TempRepo() as repo, FakeCliPath(): + command = AgentCommand("codex-build", ["codex", "exec", "-C", str(repo), "-"], capture_last_message=True) + with mock.patch("clodex.agents.tempfile.mkstemp", side_effect=tracking_mkstemp): + result = AgentRunner(repo).run(command, "prompt") + self.assertTrue(result.ok, result.stderr) + self.assertEqual(result.last_message, "implemented") + self.assertEqual(result.output, "implemented") + self.assertEqual(len(created), 1) + self.assertFalse(Path(created[0]).exists()) + + def test_runner_without_capture_uses_stdout(self): + with TempRepo() as repo, FakeCliPath(): + command = AgentCommand("codex-build", ["codex", "exec", "-C", str(repo), "-"]) + result = AgentRunner(repo).run(command, "prompt") + self.assertIsNone(result.last_message) + self.assertEqual(result.output.strip(), "implemented") + + +class WorkflowStructuredOutputTests(unittest.TestCase): + def test_build_with_inline_json_schema_reaches_approved(self): + with TempRepo() as repo, FakeCliPath(): + write_contract(repo, "claude:\n json_schema: true\n") + result = ClodexWorkflow(repo).build("implement fixture", workspace_backend="local") + self.assertEqual(result.status, "approved") + self.assertEqual(result.data["reviewers"]["claude-plan"]["approved"], True) + self.assertEqual(result.data["reviewers"]["codex-architecture"]["approved"], True) + + def test_result_string_is_used_when_structured_output_is_absent(self): + with TempRepo() as repo, FakeCliPath(no_structured_output=True): + write_contract(repo, "claude:\n json_schema: true\n") + result = ClodexWorkflow(repo).plan("plan fixture") + self.assertEqual(result.status, "planned") + self.assertEqual(result.data["goal"], "test goal") + + def test_schema_violation_is_retried_once_then_succeeds(self): + with TempRepo() as repo, FakeCliPath(schema_violation_once=True): + result = ClodexWorkflow(repo).plan("plan fixture") + self.assertEqual(result.status, "planned") + plan = json.loads((Path(result.artifacts_dir) / "01-claude-plan.json").read_text(encoding="utf-8")) + self.assertEqual(sorted(plan), sorted(COMPLETE_PLAN)) + + def test_dry_run_lists_distinct_plan_and_audit_commands(self): + with TempRepo() as repo: + write_contract(repo, "claude:\n json_schema: true\n") + commands = ClodexWorkflow(repo).dry_run_commands() + self.assertIn("--effort max", commands["claude_plan"]) + self.assertIn("--effort high", commands["claude_audit"]) + self.assertIn("", commands["claude_plan"]) + self.assertIn("", commands["claude_audit"]) + self.assertIn("--output-schema", commands["codex_audit"]) + + +if __name__ == "__main__": + unittest.main()