From dce4316cd6e653af56bfba829c5fa9127a2ae06f Mon Sep 17 00:00:00 2001 From: Minglong Pan <56749246+minglong51@users.noreply.github.com> Date: Sat, 26 Sep 2026 14:45:22 -0700 Subject: [PATCH] feat: qualify denied Codex actions and cover observer boundaries --- .github/workflows/smoke.yml | 3 + docs/designs/provider-capabilities.json | 2 +- docs/designs/provider-executables.json | 4 +- scripts/codex_qualification.py | 215 +++++++++-- scripts/provider_capture.py | 6 +- scripts/provider_probe.py | 4 +- tests/test_codex_analysis.py | 280 ++++++++++++++ tests/test_codex_observer.py | 436 +++++++++++++++++++++ tests/test_codex_qualification.py | 486 ++++++++++++++++++++++++ tests/test_provider_evidence.py | 2 +- tests/test_provider_probe.py | 6 +- 11 files changed, 1407 insertions(+), 37 deletions(-) create mode 100644 tests/test_codex_analysis.py create mode 100644 tests/test_codex_observer.py create mode 100644 tests/test_codex_qualification.py diff --git a/.github/workflows/smoke.yml b/.github/workflows/smoke.yml index b790054..70dddd2 100644 --- a/.github/workflows/smoke.yml +++ b/.github/workflows/smoke.yml @@ -123,6 +123,9 @@ jobs: - name: Exercise provider probe contracts run: PYTHONDONTWRITEBYTECODE=1 python3 -m unittest -v tests.test_provider_probe + - name: Exercise Codex protocol and qualification boundaries + run: PYTHONDONTWRITEBYTECODE=1 python3 -B -m unittest -v tests.test_codex_analysis tests.test_codex_observer tests.test_codex_qualification + - name: Verify provider evidence artifacts run: PYTHONDONTWRITEBYTECODE=1 python3 scripts/provider_evidence.py --check diff --git a/docs/designs/provider-capabilities.json b/docs/designs/provider-capabilities.json index 0e1ec39..bae255c 100644 --- a/docs/designs/provider-capabilities.json +++ b/docs/designs/provider-capabilities.json @@ -3,7 +3,7 @@ "evidence_mode": "synthetic", "human_summary": "### Provider capability evidence\n\nEvidence mode: `synthetic`. Required cells: 24/24. v0.6 gate: `experimental_preview`.\n\n| Provider | Version | Source | Support | Freshness | Pass | Fail | Unknown |\n|---|---|---|---|---|---:|---:|---:|\n| claude | synthetic | unknown:none | attach_only_unknown | unknown | 0 | 0 | 8 |\n| codex | synthetic | unknown:none | attach_only_unknown | unknown | 0 | 0 | 8 |\n| kimi | synthetic | unknown:none | attach_only_unknown | unknown | 0 | 0 | 8 |\n\n| Provider | Capability | Status | Binding (task/session/request) | Open/close | Duplicate | Stale session | Reconnect/recovery | Latency ms | Capture | Evidence |\n|---|---|---|---|---|---|---|---|---|---|---|\n| claude | approval_request | unknown | unknown/unknown/unknown | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-claude-approval-request |\n| claude | duplicate | unknown | unknown/unknown/unknown | not_applicable/not_applicable | unknown | not_applicable | not_applicable/not_applicable | | complete | evidence-claude-duplicate |\n| claude | heartbeat | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-claude-heartbeat |\n| claude | input_request | unknown | unknown/unknown/unknown | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-claude-input-request |\n| claude | reconnect | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | not_applicable | unknown/not_applicable | | complete | evidence-claude-reconnect |\n| claude | request_recovery | unknown | unknown/unknown/unknown | not_applicable/not_applicable | not_applicable | not_applicable | not_applicable/unknown | | complete | evidence-claude-request-recovery |\n| claude | stale_session | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | unknown | not_applicable/not_applicable | | complete | evidence-claude-stale-session |\n| claude | turn_state | unknown | unknown/unknown/not_applicable | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-claude-turn-state |\n| codex | approval_request | unknown | unknown/unknown/unknown | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-codex-approval-request |\n| codex | duplicate | unknown | unknown/unknown/unknown | not_applicable/not_applicable | unknown | not_applicable | not_applicable/not_applicable | | complete | evidence-codex-duplicate |\n| codex | heartbeat | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-codex-heartbeat |\n| codex | input_request | unknown | unknown/unknown/unknown | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-codex-input-request |\n| codex | reconnect | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | not_applicable | unknown/not_applicable | | complete | evidence-codex-reconnect |\n| codex | request_recovery | unknown | unknown/unknown/unknown | not_applicable/not_applicable | not_applicable | not_applicable | not_applicable/unknown | | complete | evidence-codex-request-recovery |\n| codex | stale_session | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | unknown | not_applicable/not_applicable | | complete | evidence-codex-stale-session |\n| codex | turn_state | unknown | unknown/unknown/not_applicable | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-codex-turn-state |\n| kimi | approval_request | unknown | unknown/unknown/unknown | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-kimi-approval-request |\n| kimi | duplicate | unknown | unknown/unknown/unknown | not_applicable/not_applicable | unknown | not_applicable | not_applicable/not_applicable | | complete | evidence-kimi-duplicate |\n| kimi | heartbeat | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-kimi-heartbeat |\n| kimi | input_request | unknown | unknown/unknown/unknown | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-kimi-input-request |\n| kimi | reconnect | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | not_applicable | unknown/not_applicable | | complete | evidence-kimi-reconnect |\n| kimi | request_recovery | unknown | unknown/unknown/unknown | not_applicable/not_applicable | not_applicable | not_applicable | not_applicable/unknown | | complete | evidence-kimi-request-recovery |\n| kimi | stale_session | unknown | unknown/unknown/not_applicable | not_applicable/not_applicable | not_applicable | unknown | not_applicable/not_applicable | | complete | evidence-kimi-stale-session |\n| kimi | turn_state | unknown | unknown/unknown/not_applicable | unknown/unknown | not_applicable | not_applicable | not_applicable/not_applicable | | complete | evidence-kimi-turn-state |\n", "present_cells": 24, - "provider_executables_sha256": "7ffcab616d117407e40a53af720cfaa5f289cc1cd9f004295fa8c63a492e5120", + "provider_executables_sha256": "76801e770c441995cf5b8741a7a296bb83e2e9ecf087bf5012386b303e17d957", "providers": [ { "authoritative": false, diff --git a/docs/designs/provider-executables.json b/docs/designs/provider-executables.json index dc81e43..cbf2ee0 100644 --- a/docs/designs/provider-executables.json +++ b/docs/designs/provider-executables.json @@ -2,9 +2,9 @@ "platform": "darwin-arm64", "providers": [ { - "executable_sha256": "9f7c2260251765a18d0b35198669dacc1912f6e8129a3b01f6b58d93365ff1f1", + "executable_sha256": "d8cb1e5c79684cc12a8bfc813e3a2073406921b6245744b3009be3ab5651d21e", "provider": "claude", - "provider_version": "2.1.245", + "provider_version": "2.1.283", "source_interface": "claude-native-hook" }, { diff --git a/scripts/codex_qualification.py b/scripts/codex_qualification.py index 04bc388..9a15437 100644 --- a/scripts/codex_qualification.py +++ b/scripts/codex_qualification.py @@ -5,7 +5,9 @@ import io import json import os +import re import select +import shlex import signal import subprocess import sys @@ -28,6 +30,12 @@ CAPTURE_ID = "capture-codex-input-recovery" INPUT_METHOD = "item/tool/requestUserInput" PERMISSIONS_METHOD = "item/permissions/requestApproval" +COMMAND_METHOD = "item/commandExecution/requestApproval" +FILE_METHOD = "item/fileChange/requestApproval" +SCENARIOS = { + "input": INPUT_METHOD, "permissions": PERMISSIONS_METHOD, + "command": COMMAND_METHOD, "file": FILE_METHOD, +} PROMPT = ( "This is a synthetic protocol probe. Call request_user_input exactly once, " "with one question asking whether to choose Alpha or Beta and those two options. " @@ -265,10 +273,18 @@ def close(self) -> None: class Broker: def __init__( self, transport: JsonLineTransport, binding: TaskBinding, path: Path, - *, scenario_name: str = "input", + *, scenario_name: str = "input", workspace: Path | None = None, ) -> None: - require(scenario_name in {"input", "permissions"}, "scenario-invalid") - self.request_method = INPUT_METHOD if scenario_name == "input" else PERMISSIONS_METHOD + require(scenario_name in SCENARIOS, "scenario-invalid") + require(scenario_name not in {"command", "file"} or workspace is not None, + "scenario-workspace-required") + self.scenario_name = scenario_name + self.request_method = SCENARIOS[scenario_name] + self.workspace = workspace + self.tool_item_id: str | None = None + self.tool_request_id: str | int | None = None + self.denial_sent = False + self.tool_declined = False self.transport = transport self.binding = binding self.path = path @@ -287,6 +303,10 @@ def observe(self, stream: str, message: dict[str, Any]) -> None: self.observer.ok("observe", stream=stream, observed_ns=observed_ns, message=message) def send(self, message: dict[str, Any]) -> None: + if self.tool_request_id is not None and message.get("id") == self.tool_request_id: + require(message == {"id": self.tool_request_id, "result": {"decision": "decline"}} + and not self.denial_sent, "invalid-denial-response") + self.denial_sent = True self.transport.send(message) self.observe("I", message) @@ -296,11 +316,27 @@ def receive(self, deadline: float) -> dict[str, Any]: require("jsonrpc" not in message and type(params) is dict, "protocol-envelope-invalid") if "method" in message and "id" in message: require(message["method"] == self.request_method, "unexpected-provider-request") + if self.scenario_name in {"command", "file"}: + require(self.tool_request_id is None, "duplicate-approval-request") + action_request(params, self.workspace, self.scenario_name, self.tool_item_id) + self.tool_request_id = message["id"] if message.get("method") in {"item/started", "item/completed"}: item = params.get("item") - require(type(item) is dict and item.get("type") in { - "userMessage", "agentMessage", "reasoning", "plan", - }, "provider-side-effect-attempt") + require(type(item) is dict, "provider-item-invalid") + if item.get("type") not in {"userMessage", "agentMessage", "reasoning", "plan"}: + require(self.scenario_name in {"command", "file"}, "provider-side-effect-attempt") + action_item(item, self.workspace, self.scenario_name) + if message["method"] == "item/started": + require(self.tool_item_id is None and item.get("status") == "inProgress", + "unexpected-action-start") + self.tool_item_id = item["id"] + else: + require(item["id"] == self.tool_item_id and self.denial_sent + and not self.tool_declined and item.get("status") == "declined", + "action-not-declined") + self.tool_declined = True + require(message.get("method") != "item/commandExecution/outputDelta", + "unexpected-command-output") require(message.get("method") != "warning", "provider-warning") self.observe("O", message) return message @@ -376,8 +412,9 @@ def recovery(broker: Broker, thread_id: str, request: dict[str, Any]) -> dict[st return {key: True for key in ( "stale_delivery_rejected", "duplicate_delivery_noop", "freshness_expired", "observer_process_restarted", - "pending_input_recovered" if broker.request_method == INPUT_METHOD - else "pending_permission_approval_recovered", + {INPUT_METHOD: "pending_input_recovered", PERMISSIONS_METHOD: "pending_permission_approval_recovered", + COMMAND_METHOD: "pending_command_approval_recovered", FILE_METHOD: "pending_file_approval_recovered"}[ + broker.request_method], "replay_stayed_unknown", "fresh_source_read_required", "provider_process_unchanged", )} @@ -402,13 +439,94 @@ def permission_request(params: dict[str, Any], workspace: Path) -> None: )), "permission-request-invalid") +def action_target(workspace: Path, scenario_name: str) -> Path: + return workspace.parent / ("denied-command" if scenario_name == "command" else "denied-file.txt") + + +def command_matches(command: Any, target: Path) -> bool: + if type(command) is not str or len(command) > 4096: + return False + try: + words = shlex.split(command) + if len(words) == 3 and words[0] in {"/bin/sh", "/bin/bash", "/bin/zsh"} and words[1] in {"-c", "-lc"}: + words = shlex.split(words[2]) + except ValueError: + return False + return words == ["/usr/bin/touch", str(target)] + + +def action_item(item: dict[str, Any], workspace: Path, scenario_name: str) -> None: + target = action_target(workspace, scenario_name) + require(type(item.get("id")) is str and 0 < len(item["id"]) <= 256, "action-identity-invalid") + if scenario_name == "command": + require(item.get("type") == "commandExecution" and item.get("cwd") == str(workspace) + and command_matches(item.get("command"), target) + and all(item.get(field) is None for field in ( + "exitCode", "processId", "pluginId", "scriptPath", + )) and (item.get("durationMs") is None or type(item["durationMs"]) is int and item["durationMs"] == 0), + "unexpected-command-execution") + require(item.get("aggregatedOutput") is None or item.get("aggregatedOutput") == "", + "unexpected-command-output") + else: + changes = item.get("changes") + require(item.get("type") == "fileChange" and type(changes) is list and len(changes) == 1, + "unexpected-file-change") + change = changes[0] + require(type(change) is dict and set(change) == {"path", "kind", "diff"} + and change["path"] == str(target) and change["kind"] == {"type": "add"} + and type(change["diff"]) is str + and change["diff"] in {"CLILANE_PROBE_SENTINEL\n", "+CLILANE_PROBE_SENTINEL\n"}, + "unexpected-file-change") + require(not os.path.lexists(target), "denied-action-created-target") + + +def action_request( + params: dict[str, Any], workspace: Path, scenario_name: str, item_id: str | None, +) -> None: + require(item_id is not None and params.get("itemId") == item_id + and type(params.get("startedAtMs")) is int and 0 < params["startedAtMs"] < 1 << 63, + "action-request-invalid") + target = action_target(workspace, scenario_name) + if scenario_name == "command": + require(set(params) <= { + "additionalPermissions", "approvalId", "availableDecisions", "command", "commandActions", + "cwd", "environmentId", "itemId", "kind", "networkApprovalContext", + "proposedExecpolicyAmendment", "proposedNetworkPolicyAmendments", "reason", "startedAtMs", + "threadId", "turnId", + } and params.get("cwd") == str(workspace) and command_matches(params.get("command"), target) + and params.get("kind", "command") == "command" + and (params.get("environmentId") is None or params["environmentId"] == "local") + and params.get("networkApprovalContext") is None, + "unexpected-command-request") + else: + require(set(params) <= {"grantRoot", "itemId", "reason", "startedAtMs", "threadId", "turnId"} + and (params.get("grantRoot") is None or params["grantRoot"] == str(target.parent)), + "unexpected-file-request") + require(not os.path.lexists(target), "denied-action-created-target") + + def scenario(broker: Broker, workspace: Path) -> dict[str, bool]: permissions = broker.request_method == PERMISSIONS_METHOD + action = broker.request_method in {COMMAND_METHOD, FILE_METHOD} prompt = PERMISSIONS_PROMPT if permissions else PROMPT + if action: + target = action_target(workspace, broker.scenario_name) + require(not os.path.lexists(target), "action-target-exists") + operation = ( + f"Call exec_command exactly once with cmd={json.dumps('/usr/bin/touch ' + shlex.quote(str(target)))}, " + "sandbox_permissions=require_escalated and a short justification. " + if broker.scenario_name == "command" else + f"Call apply_patch exactly once to add the absolute path {target} containing exactly " + "CLILANE_PROBE_SENTINEL followed by one newline. " + ) + prompt = "This is a synthetic approval protocol probe. " + operation + ( + "Request approval; it will be denied. Do not use another tool, run a preliminary command, " + "or retry after denial. After the denial, reply exactly CLILANE_PROBE_OK and end the turn." + ) policy: str | dict[str, Any] = {"granular": { - "mcp_elicitations": False, "rules": False, "sandbox_approval": False, - "skill_approval": False, "request_permissions": True, - }} if permissions else "never" + "mcp_elicitations": False, "rules": False, "sandbox_approval": action, + "skill_approval": False, "request_permissions": permissions, + }} if permissions or action else "never" initialize(broker) skills(broker, workspace, isolated=True) broker.send({"id": "probe-thread", "method": "thread/start", "params": { @@ -426,7 +544,7 @@ def scenario(broker: Broker, workspace: Path) -> dict[str, bool]: "threadId": thread_id, "input": [{"type": "text", "text": prompt}], "approvalPolicy": policy, "approvalsReviewer": "user", "sandboxPolicy": {"type": "readOnly", "networkAccess": False}, - "collaborationMode": {"mode": "default" if permissions else "plan", "settings": { + "collaborationMode": {"mode": "default" if permissions or action else "plan", "settings": { "model": model, "reasoning_effort": "low", "developer_instructions": prompt, }}, }}) @@ -467,12 +585,12 @@ def scenario(broker: Broker, workspace: Path) -> dict[str, bool]: "unexpected-provider-request") if permissions: permission_request(params, workspace) - else: + elif not action: require(params.get("isBlocking") is True and type(params.get("questions")) is list and len(params["questions"]) == 1, "unexpected-input-request") recovered = recovery(broker, thread_id, message) - broker.send({"id": message["id"], "result": { + broker.send({"id": message["id"], "result": {"decision": "decline"} if action else { "permissions": {}, "scope": "turn", } if permissions else {"answers": {}}}) next_read = time.monotonic() + 1 @@ -488,26 +606,69 @@ def scenario(broker: Broker, workspace: Path) -> dict[str, bool]: recovered["permission_approval_resolved" if permissions else "input_request_resolved"] = True if permissions: recovered["permission_grant_empty"] = True + if action: + require(broker.tool_declined and not os.path.lexists(action_target(workspace, broker.scenario_name)), + "action-denial-not-proven") + recovered.pop("input_request_resolved") + recovered[f"{broker.scenario_name}_approval_resolved"] = True + recovered["action_declined_without_target_creation"] = True return recovered -def verify_journal(root: Path, binding: TaskBinding, report: dict[str, Any]) -> None: +def verify_diagnostics(stderr: bytes, scenario_name: str) -> None: + if not stderr: + return + expected = { + "command": b'error=exec_command failed: CreateProcess { message: "Rejected(\\"rejected by user\\")" }', + "file": b"error=patch rejected by user", + }.get(scenario_name) + require(expected is not None and re.fullmatch( + rb"\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d(?:\.\d{1,9})?Z ERROR codex_core::tools::router: " + + re.escape(expected) + rb"\n", stderr, + ) is not None, "provider-stderr-output") + + +def verify_journal( + root: Path, binding: TaskBinding, report: dict[str, Any], *, scenario_name: str = "input", +) -> None: capture = _read_private(root / "captures" / f"{CAPTURE_ID}.capture", PER_CAPTURE_BYTES) require(hashlib.sha256(capture).hexdigest() == report["source_capture_sha256"], "capture-changed") buffers = {"I": bytearray(), "O": bytearray()} source: dict[str, list[dict[str, Any]]] = {"I": [], "O": []} - for stream, _observed_ns, payload in iter_capture_frames(io.BytesIO(capture)): + observed: list[tuple[str, int, dict[str, Any]]] = [] + stderr = bytearray() + stderr_times: list[int] = [] + for stream, observed_ns, payload in iter_capture_frames(io.BytesIO(capture)): if stream == "E": - require(not payload, "provider-stderr-output") + stderr.extend(payload) + if payload: + stderr_times.append(observed_ns) continue buffer = buffers[stream] buffer.extend(payload) while b"\n" in buffer: end = buffer.index(b"\n") - source[stream].append(strict_message(bytes(buffer[:end]))) + message = strict_message(bytes(buffer[:end])) + source[stream].append(message) + observed.append((stream, observed_ns, message)) del buffer[:end + 1] require(not any(buffers.values()), "capture-message-truncated") + verify_diagnostics(bytes(stderr), scenario_name) + if stderr: + requests = [message for stream, _, message in observed + if stream == "O" and message.get("method") == SCENARIOS[scenario_name] + and "id" in message] + require(len(requests) == 1, "denial-diagnostic-unbound") + request_id = requests[0]["id"] + declines = [observed_ns for stream, observed_ns, message in observed + if stream == "I" and type(message.get("id")) is type(request_id) + and message == {"id": request_id, "result": {"decision": "decline"}}] + completed = [observed_ns for stream, observed_ns, message in observed + if stream == "O" and message.get("method") == "turn/completed"] + require(len(declines) == len(completed) == 1 + and all(declines[0] <= moment < completed[0] for moment in stderr_times), + "denial-diagnostic-outside-turn") rows = _read_private(root / "journal" / "source.jsonl", 256 * 1024).splitlines() require(strict_message(rows[0]) == {"schema_version": 1, "task_generation": binding.generation}, "journal-binding-mismatch") @@ -535,7 +696,7 @@ def run( root: Path, executable: Path, auth_file: Path, cancellation: Cancellation | None = None, *, scenario_name: str = "input", ) -> dict[str, Any]: - require(scenario_name in {"input", "permissions"}, "scenario-invalid") + require(scenario_name in SCENARIOS, "scenario-invalid") binding = task_binding() require(root.is_absolute() and root.resolve() == root, "run-root-not-canonical") _private_directory(root, empty=True) @@ -550,6 +711,8 @@ def run( ) + CONFIG + ( "request_permissions_tool = " + ("true" if scenario_name == "permissions" else "false") + "\n" ) + if scenario_name in {"command", "file"}: + isolated_config = isolated_config.replace("shell_tool = false\n", "shell_tool = true\n") _write_private_bytes(root / "discovery-config" / "config.toml", isolated_config.encode()) paths = preflight(root, pinned, root / "discovery-config", "capture-discovery", isolated=False) config = isolated_config + "".join( @@ -585,7 +748,7 @@ def run( spec = ProbeConfig("codex", root / "captures", CAPTURE_ID, pinned.path, root / "workspace") transport = JsonLineTransport(spec, pinned, environment=environment(root / "config", root / "tmp")) broker = Broker(transport, binding, root / "journal" / "source.jsonl", - scenario_name=scenario_name) + scenario_name=scenario_name, workspace=spec.workspace) checks = scenario(broker, spec.workspace) transport.finish() finished = True @@ -595,11 +758,13 @@ def run( require(report["analysis_complete"] and report["provider_warning_count"] == 0 and report["turn_count"] == 1 and report["thread_read_count"] >= 2, "capture-analysis-incomplete") - if scenario_name == "permissions": + if scenario_name != "input": require(any(event["event"] == "thread_state" and "waitingOnApproval" in event["flags"] for event in report["events"]), "approval-state-not-observed") - verify_journal(root, binding, report) + verify_journal(root, binding, report, scenario_name=scenario_name) checks["journal_matches_source"] = True + if scenario_name in {"command", "file"}: + checks["provider_diagnostics_match_denial"] = True return { "schema_version": 1, "scope": f"phase0-{scenario_name}-observer-restart", "qualification": "unqualified", "provider": "codex", "provider_version": "0.155.1", @@ -638,9 +803,9 @@ def main(argv: Sequence[str] | None = None) -> int: ) parser.add_argument("--run-root", required=True) parser.add_argument("--provider-executable", required=True) - parser.add_argument("--scenario", choices=("input", "permissions"), default="input", - help="Input cancellation (default), or experimental permission approval " - "with an empty grant; neither scenario executes provider commands.") + parser.add_argument("--scenario", choices=tuple(SCENARIOS), default="input", + help="Input cancellation (default), empty permission grant, " + "or a declined synthetic command/file approval.") parser.add_argument("--auth-file", required=True, help="Private Codex subscription credentials valid for at least three minutes; " "the probe removes its isolated copy after the run.") diff --git a/scripts/provider_capture.py b/scripts/provider_capture.py index 1bf14cf..19f5e19 100644 --- a/scripts/provider_capture.py +++ b/scripts/provider_capture.py @@ -155,10 +155,10 @@ class ProviderLifecycle: "claude": ProviderSpec( name="claude", executable="claude", - version="2.1.245", + version="2.1.283", version_outputs=( - b"2.1.245 (Claude Code)\n", - b"2.1.245 (Claude Code)\r\n", + b"2.1.283 (Claude Code)\n", + b"2.1.283 (Claude Code)\r\n", ), source_interface="claude-native-hook", mode="stdin", diff --git a/scripts/provider_probe.py b/scripts/provider_probe.py index 7f18b8d..30f65fd 100644 --- a/scripts/provider_probe.py +++ b/scripts/provider_probe.py @@ -29,7 +29,7 @@ "codex": ROOT / "scripts/collect_codex_evidence.py", "kimi": ROOT / "scripts/collect_kimi_evidence.py", } -PROVIDER_VERSIONS = {"claude": "2.1.245", "codex": "0.155.1", "kimi": "0.38.0"} +PROVIDER_VERSIONS = {"claude": "2.1.283", "codex": "0.155.1", "kimi": "0.38.0"} PROBE_PROMPT = ( "Reply with exactly CLILANE_PROBE_OK and no other text. Do not call tools." ) @@ -74,7 +74,7 @@ READ_CHUNK_BYTES = 16 * 1024 SHA256_PATTERN = re.compile(r"[0-9a-f]{64}") SOURCE_INTERFACES = { - "claude": ("2.1.245", "claude-native-hook"), + "claude": ("2.1.283", "claude-native-hook"), "codex": ("0.155.1", "codex-app-server"), "kimi": ("0.38.0", "kimi-acp"), } diff --git a/tests/test_codex_analysis.py b/tests/test_codex_analysis.py new file mode 100644 index 0000000..523bfa0 --- /dev/null +++ b/tests/test_codex_analysis.py @@ -0,0 +1,280 @@ +from __future__ import annotations + +import copy +import importlib.util +import json +import sys +import unittest +from pathlib import Path +from typing import Any +from unittest import mock + + +ROOT = Path(__file__).resolve().parents[1] +SPEC = importlib.util.spec_from_file_location("codex_analysis", ROOT / "scripts/codex_analysis.py") +if SPEC is None or SPEC.loader is None: + raise RuntimeError("cannot load codex_analysis") +codex_analysis = importlib.util.module_from_spec(SPEC) +sys.modules[SPEC.name] = codex_analysis +SPEC.loader.exec_module(codex_analysis) + + +class CodexAnalysisTests(unittest.TestCase): + def setUp(self) -> None: + self.analysis = codex_analysis.CodexAnalysis() + self.thread = {"id": "private-thread-sentinel", "sessionId": "private-session-sentinel"} + self.observed_ns = 0 + + def feed(self, stream: str, message: dict[str, Any]) -> None: + self.observed_ns += 1_000_000 + self.analysis.feed(stream, self.observed_ns, (json.dumps(message) + "\n").encode()) + + def bind_thread(self) -> None: + self.feed("I", {"id": "thread-start", "method": "thread/start", "params": {}}) + self.feed("O", {"id": "thread-start", "result": {"thread": self.thread}}) + self.feed("O", {"method": "thread/started", "params": {"thread": self.thread}}) + + def start_turn(self, turn_id: str = "private-turn-sentinel", *, notification_first: bool = False) -> None: + self.feed("I", {"id": "start-" + turn_id, "method": "turn/start", "params": {"threadId": self.thread["id"]}}) + turn = {"id": turn_id, "status": "inProgress"} + response = {"id": "start-" + turn_id, "result": {"turn": turn}} + notification = {"method": "turn/started", "params": {"threadId": self.thread["id"], "turn": turn}} + for message in ([notification, response] if notification_first else [response, notification]): + self.feed("O", message) + + def complete_turn(self, turn_id: str = "private-turn-sentinel", status: str = "completed") -> None: + self.feed("O", {"method": "turn/completed", "params": {"threadId": self.thread["id"], "turn": {"id": turn_id, "status": status}}}) + + def request(self, request_id: str | int = "private-request-sentinel", method: str = "item/tool/requestUserInput", **extra: Any) -> dict[str, Any]: + params = {"threadId": self.thread["id"], "turnId": "private-turn-sentinel", "itemId": "private-item-sentinel", **extra} + if method == "item/tool/requestUserInput": + params.setdefault("isBlocking", True) + return {"id": request_id, "method": method, "params": params} + + def resolve(self, request_id: str | int = "private-request-sentinel") -> None: + self.feed("O", {"method": "serverRequest/resolved", "params": {"threadId": self.thread["id"], "requestId": request_id}}) + + def test_completed_turns_accept_both_start_response_orders(self) -> None: + self.bind_thread() + for number, status in enumerate(("completed", "interrupted", "failed")): + turn_id = f"turn-{number}" + self.start_turn(turn_id, notification_first=number % 2 == 0) + self.complete_turn(turn_id, status) + result = self.analysis.finish() + self.assertTrue(result["analysis_complete"]) + self.assertEqual(result["turn_count"], 3) + self.assertEqual([event["state"] for event in result["events"]], ["inProgress", "completed", "inProgress", "interrupted", "inProgress", "failed"]) + self.assertEqual(result["qualification"], "unqualified") + self.assertEqual(result["task_binding"], "unknown") + self.assertEqual(result["source_health"], "unknown") + + def test_input_and_approval_lifecycles_require_server_resolution(self) -> None: + cases = [("item/tool/requestUserInput", True), ("item/tool/requestUserInput", False), ("item/commandExecution/requestApproval", True), ("item/fileChange/requestApproval", True), ("item/permissions/requestApproval", True)] + for method, blocking in cases: + with self.subTest(method=method, blocking=blocking): + self.setUp() + self.bind_thread() + self.start_turn() + self.feed("O", self.request(method=method, isBlocking=blocking, prompt="private-prompt-sentinel")) + self.feed("I", {"id": "private-request-sentinel", "result": {"answer": "private-answer-sentinel"}}) + self.complete_turn() + self.assertFalse(self.analysis.finish()["analysis_complete"]) + self.assertEqual(self.analysis.finish()["unresolved_request_count"], 1) + self.resolve() + result = self.analysis.finish() + self.assertTrue(result["analysis_complete"]) + opened, resolved = [event for event in result["events"] if event["request_id"]] + self.assertEqual(opened["request_id"], resolved["request_id"]) + self.assertEqual(opened["blocking"], blocking) + self.assertEqual(opened["request_kind"], "input" if "requestUserInput" in method else "approval") + self.assertNotIn("private-", json.dumps(result)) + + def test_string_and_integer_request_ids_are_distinct(self) -> None: + self.bind_thread() + self.start_turn() + self.feed("O", self.request(7)) + self.feed("O", self.request("7")) + self.resolve(7) + self.assertEqual(self.analysis.finish()["unresolved_request_count"], 1) + self.resolve("7") + self.complete_turn() + self.assertTrue(self.analysis.finish()["analysis_complete"]) + self.assertEqual({event["request_id"] for event in self.analysis.events if event["request_id"]}, {"synthetic-codex-request-1", "synthetic-codex-request-2"}) + + def test_exact_duplicates_are_idempotent_across_lifecycle(self) -> None: + self.bind_thread() + self.start_turn() + request = self.request() + self.feed("O", request) + self.feed("O", copy.deepcopy(request)) + self.feed("I", {"id": "thread-start", "method": "thread/start", "params": {}}) + self.feed("O", {"id": "thread-start", "result": {"thread": self.thread}}) + self.resolve() + self.resolve() + self.complete_turn() + self.complete_turn() + result = self.analysis.finish() + self.assertTrue(result["analysis_complete"]) + self.assertEqual(result["duplicate_count"], 5) + self.assertEqual(len(result["events"]), 4) + + def test_conflicting_duplicate_identities_fail_closed(self) -> None: + cases = [ + ("I", {"id": "thread-start", "method": "initialize", "params": {}}, "conflicting-client-request"), + ("O", {"id": "thread-start", "result": {"thread": {"id": "other", "sessionId": "other"}}}, "conflicting-response"), + ("O", self.request(prompt="changed"), "conflicting-server-request"), + ] + for stream, message, error in cases: + with self.subTest(error=error): + self.setUp() + self.bind_thread() + self.start_turn() + self.feed("O", self.request()) + with self.assertRaisesRegex(codex_analysis.AnalysisError, f"^{error}$"): + self.feed(stream, message) + + def test_stale_or_unbound_thread_turn_and_request_are_rejected(self) -> None: + cases = [ + ("O", self.request(threadId="old-thread"), "thread-binding-mismatch"), + ("O", self.request(turnId="old-turn"), "turn-binding-mismatch"), + ("O", {"id": "unknown", "result": {}}, "unbound-response"), + ("I", {"id": "unknown", "result": {}}, "unbound-request-response"), + ("O", {"method": "serverRequest/resolved", "params": {"threadId": self.thread["id"], "requestId": "unknown"}}, "unbound-request-resolution"), + ("O", {"method": "thread/started", "params": {"thread": {**self.thread, "sessionId": "old-session"}}}, "thread-binding-mismatch"), + ] + for stream, message, error in cases: + with self.subTest(error=error): + self.setUp() + self.bind_thread() + self.start_turn() + with self.assertRaisesRegex(codex_analysis.AnalysisError, f"^{error}$"): + self.feed(stream, message) + + def test_overlapping_and_reused_turns_are_rejected(self) -> None: + self.bind_thread() + self.start_turn() + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^overlapping-turns$"): + self.start_turn("other-turn") + self.complete_turn() + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^turn-binding-mismatch$"): + self.feed("O", self.request()) + self.feed("I", {"id": "new-start", "method": "turn/start", "params": {"threadId": self.thread["id"]}}) + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^reused-turn-identity$"): + self.feed("O", {"id": "new-start", "result": {"turn": {"id": "private-turn-sentinel", "status": "inProgress"}}}) + + def test_completion_before_start_and_conflicting_terminal_status_fail(self) -> None: + self.bind_thread() + self.feed("I", {"id": "start", "method": "turn/start", "params": {"threadId": self.thread["id"]}}) + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^turn-order-mismatch$"): + self.complete_turn() + self.setUp() + self.bind_thread() + self.start_turn() + self.complete_turn() + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^conflicting-turn-completion$"): + self.complete_turn(status="failed") + + def test_thread_reads_are_correlated_and_cannot_change_binding(self) -> None: + self.bind_thread() + self.start_turn() + self.complete_turn() + request = {"id": "read", "method": "thread/read", "params": {"threadId": self.thread["id"], "includeTurns": False}} + self.feed("I", request) + self.assertFalse(self.analysis.finish()["analysis_complete"]) + self.feed("O", {"id": "read", "result": {"thread": self.thread}}) + self.assertEqual(self.analysis.finish()["thread_read_count"], 1) + self.assertTrue(self.analysis.finish()["analysis_complete"]) + self.feed("I", {**request, "id": "read-2"}) + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^thread-binding-mismatch$"): + self.feed("O", {"id": "read-2", "result": {"thread": {**self.thread, "sessionId": "stale"}}}) + + def test_thread_read_parameters_are_strict(self) -> None: + for extras in ({"includeTurns": 1}, {"includeTurns": None}, {"extra": True}): + with self.subTest(extras=extras): + self.setUp() + self.bind_thread() + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^invalid-thread-read$"): + self.feed("I", {"id": "read", "method": "thread/read", "params": {"threadId": self.thread["id"], **extras}}) + + def test_strict_json_envelope_and_number_boundaries(self) -> None: + cases = [(b'{"id":1,"id":2,"result":{}}\n', "duplicate-json-key"), (b'{"id":1,"result":{"x":0,"x":1}}\n', "duplicate-json-key"), (b'{"id":1,"result":{"x":NaN}}\n', "invalid-json-number"), (b'{"id":1,"result":{"x":1e999}}\n', "invalid-json-number"), (b'\xff\n', "invalid-json"), (b'[]\n', "invalid-object"), (b'{"id":1,"result":{},"error":{}}\n', "invalid-envelope"), (b'{"jsonrpc":"2.0","id":1,"result":{}}\n', "invalid-envelope"), (b'{"method":"initialize","result":{}}\n', "invalid-envelope")] + for raw, error in cases: + with self.subTest(raw=raw): + with self.assertRaisesRegex(codex_analysis.AnalysisError, f"^{error}$"): + codex_analysis.CodexAnalysis().feed("I", 0, raw) + + def test_request_identity_types_and_lengths_are_strict(self) -> None: + for identity in (None, True, False, [], {}, 1.5, "", "x" * 257, 1 << 63, -(1 << 63) - 1): + with self.subTest(identity=identity): + self.setUp() + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^invalid-identity$"): + self.feed("I", {"id": identity, "method": "initialize", "params": {}}) + + def test_input_blocking_marker_must_be_boolean(self) -> None: + for value in (None, 0, 1, "true", []): + with self.subTest(value=value): + self.setUp() + self.bind_thread() + self.start_turn() + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^invalid-request-blocking$"): + self.feed("O", self.request(isBlocking=value)) + + def test_thread_flags_are_bounded_and_order_independent(self) -> None: + self.bind_thread() + flags = ["waitingOnUserInput", "waitingOnApproval"] + for value in (flags, list(reversed(flags))): + self.feed("O", {"method": "thread/status/changed", "params": {"threadId": self.thread["id"], "status": {"type": "active", "activeFlags": value}}}) + self.assertEqual(self.analysis.finish()["duplicate_count"], 1) + self.assertEqual(self.analysis.events[0]["flags"], sorted(flags)) + for status in ({"type": "active"}, {"type": "idle", "activeFlags": flags}, {"type": "active", "activeFlags": ["unknown"]}, {"type": "active", "activeFlags": [flags[0], flags[0]]}): + with self.subTest(status=status): + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^invalid-thread-flags$"): + self.feed("O", {"method": "thread/status/changed", "params": {"threadId": self.thread["id"], "status": status}}) + + def test_partial_lines_stderr_and_payloads_do_not_leak(self) -> None: + self.bind_thread() + self.start_turn() + raw = (json.dumps({"method": "item/agentMessage/delta", "params": {"threadId": self.thread["id"], "turnId": "private-turn-sentinel", "delta": "private-content-\u2603"}}, ensure_ascii=False) + "\n").encode() + split = raw.index("\u2603".encode()) + 1 + self.analysis.feed("O", 9_000_000, raw[:split]) + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^truncated-json-line$"): + self.analysis.finish() + self.analysis.feed("O", 10_000_000, raw[split:]) + self.analysis.feed("E", 11_000_000, b"private-stderr-sentinel\xff") + self.complete_turn() + result = self.analysis.finish() + self.assertTrue(result["analysis_complete"]) + self.assertEqual(result["ignored_notification_count"], 1) + self.assertNotIn("private-", json.dumps(result)) + + def test_unhandled_messages_prevent_complete_analysis(self) -> None: + self.bind_thread() + self.start_turn() + self.complete_turn() + self.feed("O", {"method": "unknown/newFeature", "params": {}}) + self.assertFalse(self.analysis.finish()["analysis_complete"]) + self.assertEqual(self.analysis.finish()["unhandled_message_count"], 1) + + def test_protocol_limits_reject_excess_before_unbounded_growth(self) -> None: + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^line-limit$"): + self.analysis.feed("O", 0, b"x" * (codex_analysis.MAX_LINE_BYTES + 1)) + self.setUp() + with mock.patch.object(codex_analysis, "MAX_MESSAGES", 1): + self.feed("I", {"method": "initialized"}) + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^message-limit$"): + self.feed("I", {"method": "initialized"}) + self.setUp() + with mock.patch.object(codex_analysis, "MAX_IDENTITIES", 1): + self.feed("I", {"id": "one", "method": "initialize"}) + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^identity-limit$"): + self.feed("I", {"id": "two", "method": "initialize"}) + self.setUp() + self.bind_thread() + with mock.patch.object(codex_analysis, "MAX_EVENTS", 1): + self.start_turn() + with self.assertRaisesRegex(codex_analysis.AnalysisError, "^event-limit$"): + self.complete_turn() + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_codex_observer.py b/tests/test_codex_observer.py new file mode 100644 index 0000000..5522e58 --- /dev/null +++ b/tests/test_codex_observer.py @@ -0,0 +1,436 @@ +from __future__ import annotations + +import hashlib +import io +import json +import os +import stat +import sys +import tempfile +import unittest +from contextlib import redirect_stderr, redirect_stdout +from pathlib import Path +from typing import Any +from unittest import mock + + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +try: + import codex_observer +finally: + sys.path.pop(0) + + +class ObserverFixture(unittest.TestCase): + def setUp(self) -> None: + temporary = tempfile.TemporaryDirectory(prefix="codex-observer-test-") + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name).resolve() + self.root.chmod(0o700) + self.binding = codex_observer.TaskBinding("a" * 64) + self.now = 10_000_000_000 + self.observed_ns = 0 + self.observer = codex_observer.Observer(self.binding, clock=lambda: self.now) + self.thread = {"id": "private-thread-sentinel", "sessionId": "private-session-sentinel"} + + def command(self, op: str, **values: Any) -> dict[str, Any]: + return {"op": op, "task_generation": self.binding.generation, "instance": self.observer.instance, **values} + + def observe(self, stream: str, message: dict[str, Any]) -> dict[str, Any]: + self.observed_ns += 1_000_000 + return self.observer.handle(self.command("observe", stream=stream, observed_ns=self.observed_ns, message=message)) + + def bind_thread(self) -> None: + self.observe("I", {"id": "thread-start", "method": "thread/start", "params": {}}) + self.observe("O", {"id": "thread-start", "result": {"thread": self.thread}}) + + def read_thread(self, request_id: str = "read-live") -> None: + self.observe("I", {"id": request_id, "method": "thread/read", "params": {"threadId": self.thread["id"], "includeTurns": False}}) + self.observe("O", {"id": request_id, "result": {"thread": self.thread}}) + + def make_ready(self) -> dict[str, Any]: + self.bind_thread() + self.read_thread() + return self.observer.handle(self.command("ready", recovered_count=0)) + + def frames(self, *, read: bool = False, request: bool = False) -> list[tuple[str, dict[str, Any]]]: + result = [("I", {"id": "thread-start", "method": "thread/start", "params": {}}), ("O", {"id": "thread-start", "result": {"thread": self.thread}})] + if request: + turn = {"id": "private-turn-sentinel", "status": "inProgress"} + result.extend([("I", {"id": "turn-start", "method": "turn/start", "params": {"threadId": self.thread["id"]}}), ("O", {"id": "turn-start", "result": {"turn": turn}}), ("O", {"method": "turn/started", "params": {"threadId": self.thread["id"], "turn": turn}}), ("O", {"id": "private-request-sentinel", "method": "item/permissions/requestApproval", "params": {"threadId": self.thread["id"], "turnId": turn["id"], "itemId": "private-item-sentinel", "reason": "private-prompt-sentinel"}})]) + if read: + result.extend([("I", {"id": "read-replayed", "method": "thread/read", "params": {"threadId": self.thread["id"], "includeTurns": False}}), ("O", {"id": "read-replayed", "result": {"thread": self.thread}})]) + return result + + def journal(self, frames: list[tuple[str, dict[str, Any]]], name: str = "journal.jsonl") -> tuple[Path, str]: + path = self.root / name + journal = codex_observer.Journal(path, self.binding) + try: + for number, (stream, message) in enumerate(frames, 1): + journal.append(stream, number * 1_000_000, message) + digest = journal.digest.hexdigest() + finally: + journal.close() + return path, digest + + def replay(self, frames: list[tuple[str, dict[str, Any]]]) -> None: + path, digest = self.journal(frames) + self.observer.replay(path, digest) + self.observed_ns = len(frames) * 1_000_000 + + def private_file(self, raw: bytes, name: str = "fixture.jsonl") -> Path: + path = self.root / name + path.write_bytes(raw) + path.chmod(0o600) + return path + + +class ObserverFreshnessTests(ObserverFixture): + def test_bound_live_thread_read_is_required_before_ready(self) -> None: + self.bind_thread() + with self.assertRaisesRegex(codex_observer.ObserverError, "^fresh-source-proof-required$"): + self.observer.handle(self.command("ready", recovered_count=0)) + self.observe("I", {"id": "read-live", "method": "thread/read", "params": {"threadId": self.thread["id"]}}) + with self.assertRaisesRegex(codex_observer.ObserverError, "^fresh-source-proof-required$"): + self.observer.handle(self.command("ready", recovered_count=0)) + self.observe("O", {"id": "read-live", "result": {"thread": self.thread}}) + report = self.observer.handle(self.command("ready", recovered_count=0)) + self.assertEqual((report["health"], report["state"]), ("healthy", "ready")) + self.assertEqual(report["fresh_until_monotonic_ns"], self.now + codex_observer.FRESHNESS_NS) + self.assertEqual(report["qualification"], "unqualified") + + def test_replay_and_duplicate_read_cannot_supply_new_live_proof(self) -> None: + self.replay(self.frames(read=True)) + report = self.observer.report() + self.assertEqual((report["health"], report["state"]), ("unknown", "recovering")) + self.assertEqual(report["thread_read_count"], 1) + self.assertEqual(report["replayed_message_count"], 4) + self.assertIsNone(report["fresh_until_monotonic_ns"]) + self.read_thread("read-replayed") + self.assertEqual(self.observer.report()["duplicate_count"], 2) + with self.assertRaisesRegex(codex_observer.ObserverError, "^fresh-source-proof-required$"): + self.observer.handle(self.command("ready", recovered_count=0)) + self.read_thread("read-new") + report = self.observer.handle(self.command("ready", recovered_count=0)) + self.assertEqual(report["health"], "healthy") + self.assertEqual(report["thread_read_count"], 2) + + def test_live_response_to_a_replayed_request_is_not_live_proof(self) -> None: + self.replay(self.frames(read=True)[:-1]) + self.observe("O", {"id": "read-replayed", "result": {"thread": self.thread}}) + self.assertEqual(self.observer.report()["thread_read_count"], 1) + with self.assertRaisesRegex(codex_observer.ObserverError, "^fresh-source-proof-required$"): + self.observer.handle(self.command("ready", recovered_count=0)) + + def test_recovered_requests_require_exact_count_and_stay_sanitized(self) -> None: + self.replay(self.frames(request=True)) + self.read_thread() + for count in (0, 2, True, "1"): + with self.subTest(count=count): + with self.assertRaisesRegex(codex_observer.ObserverError, "^recovery-count-mismatch$"): + self.observer.handle(self.command("ready", recovered_count=count)) + report = self.observer.handle(self.command("ready", recovered_count=1)) + self.assertEqual(report["unresolved_requests"], ["synthetic-codex-request-1"]) + self.assertNotIn("private-", json.dumps(report)) + self.observe("I", {"id": "private-request-sentinel", "result": {"permissions": {}}}) + self.assertEqual(self.observer.report()["unresolved_requests"], ["synthetic-codex-request-1"]) + report = self.observe("O", {"method": "serverRequest/resolved", "params": {"threadId": self.thread["id"], "requestId": "private-request-sentinel"}}) + self.assertEqual(report["unresolved_requests"], []) + + def test_invalid_response_and_out_of_order_observation_are_atomic(self) -> None: + self.bind_thread() + self.observe("I", {"id": "read-live", "method": "thread/read", "params": {"threadId": self.thread["id"]}}) + before = self.observer.report() + with self.assertRaisesRegex(codex_observer.AnalysisError, "^thread-binding-mismatch$"): + self.observe("O", {"id": "read-live", "result": {"thread": {**self.thread, "sessionId": "stale-session"}}}) + self.assertEqual(self.observer.report(), before) + with self.assertRaisesRegex(codex_observer.ObserverError, "^observation-order-mismatch$"): + self.observer.handle(self.command("observe", stream="O", observed_ns=0, message={"id": "read-live", "result": {"thread": self.thread}})) + self.assertEqual(self.observer.report(), before) + self.observe("O", {"id": "read-live", "result": {"thread": self.thread}}) + self.assertEqual(self.observer.handle(self.command("ready", recovered_count=0))["health"], "healthy") + + def test_health_expires_at_exact_deadline_and_cannot_be_revived(self) -> None: + ready = self.make_ready() + deadline = ready["fresh_until_monotonic_ns"] + self.now = deadline - 1 + self.assertEqual(self.observer.report()["health"], "healthy") + self.now = deadline + report = self.observer.handle(self.command("status")) + self.assertEqual((report["health"], report["state"]), ("unknown", "expired")) + self.assertIsNone(report["fresh_until_monotonic_ns"]) + for command in (self.command("ready", recovered_count=0), self.command("observe", stream="I", observed_ns=20_000_000, message={"method": "initialized"})): + with self.subTest(op=command["op"]): + with self.assertRaisesRegex(codex_observer.ObserverError, "^observer-expired$"): + self.observer.handle(command) + self.now -= 1 + self.assertEqual(self.observer.report()["state"], "expired") + + def test_only_a_new_correlated_read_refreshes_ready_health(self) -> None: + deadline = self.make_ready()["fresh_until_monotonic_ns"] + self.now += 1_000_000_000 + self.read_thread("read-live") + self.assertEqual(self.observer.report()["fresh_until_monotonic_ns"], deadline) + self.observe("O", {"method": "thread/status/changed", "params": {"threadId": self.thread["id"], "status": {"type": "idle"}}}) + self.assertEqual(self.observer.report()["fresh_until_monotonic_ns"], deadline) + self.read_thread("read-fresh") + self.assertEqual(self.observer.report()["fresh_until_monotonic_ns"], self.now + codex_observer.FRESHNESS_NS) + + def test_stale_proof_and_unhandled_source_cannot_mark_ready(self) -> None: + self.bind_thread() + self.read_thread() + self.now += codex_observer.FRESHNESS_NS + with self.assertRaisesRegex(codex_observer.ObserverError, "^fresh-source-proof-required$"): + self.observer.handle(self.command("ready", recovered_count=0)) + self.read_thread("read-fresh") + self.observe("O", {"method": "unknown/newFeature", "params": {}}) + with self.assertRaisesRegex(codex_observer.ObserverError, "^fresh-source-proof-required$"): + self.observer.handle(self.command("ready", recovered_count=0)) + + def test_stale_task_and_observer_commands_do_not_mutate_state(self) -> None: + self.bind_thread() + before = self.observer.report() + for field, value, error in (("task_generation", "b" * 64, "stale-task-generation"), ("instance", "old-instance", "stale-observer-instance")): + with self.subTest(field=field): + command = self.command("observe", stream="I", observed_ns=3_000_000, message={"method": "initialized"}) + command[field] = value + with self.assertRaisesRegex(codex_observer.ObserverError, f"^{error}$"): + self.observer.handle(command) + self.assertEqual(self.observer.report(), before) + replacement = codex_observer.Observer(self.binding, clock=lambda: self.now) + self.assertNotEqual(replacement.instance, self.observer.instance) + with self.assertRaisesRegex(codex_observer.ObserverError, "^stale-observer-instance$"): + replacement.handle(self.command("status")) + + def test_commands_and_observation_frames_reject_invalid_shapes(self) -> None: + for command in (self.command("status", extra=True), self.command("unknown"), self.command("ready"), self.command("observe", stream="I", observed_ns=0)): + with self.subTest(command=command): + with self.assertRaisesRegex(codex_observer.ObserverError, "^invalid-command$"): + self.observer.handle(command) + for stream, observed, message in (("E", 0, {}), (None, 0, {}), ("O", True, {}), ("O", -1, {}), ("O", 1 << 63, {}), ("O", 0, []), ("O", 0, {"payload": "x" * 65536})): + with self.subTest(stream=stream, observed=observed): + with self.assertRaisesRegex(codex_observer.ObserverError, "^invalid-observation$"): + self.observer.handle(self.command("observe", stream=stream, observed_ns=observed, message=message)) + self.assertEqual(self.observer.report()["message_count"], 0) + + +class JournalBoundaryTests(ObserverFixture): + def test_journal_is_private_exclusive_and_digest_binds_all_bytes(self) -> None: + path, digest = self.journal(self.frames()) + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + self.assertEqual(digest, hashlib.sha256(path.read_bytes()).hexdigest()) + original = path.read_bytes() + with self.assertRaises(FileExistsError): + codex_observer.Journal(path, self.binding) + self.assertEqual(path.read_bytes(), original) + self.observer.replay(path, digest) + self.assertEqual(self.observer.report()["replayed_message_count"], 2) + self.assertEqual(self.observer.report()["health"], "unknown") + + def test_journal_integrity_and_task_binding_fail_before_replay(self) -> None: + path, digest = self.journal(self.frames()) + with self.assertRaisesRegex(codex_observer.ObserverError, "^journal-integrity-mismatch$"): + self.observer.replay(path, "0" * 64) + self.assertEqual(self.observer.report()["message_count"], 0) + other = codex_observer.Observer(codex_observer.TaskBinding("b" * 64)) + with self.assertRaisesRegex(codex_observer.ObserverError, "^journal-binding-mismatch$"): + other.replay(path, digest) + self.assertEqual(other.report()["message_count"], 0) + raw = path.read_bytes().rstrip(b"\n") + path.write_bytes(raw) + with self.assertRaisesRegex(codex_observer.ObserverError, "^journal-integrity-mismatch$"): + self.observer.replay(path, hashlib.sha256(raw).hexdigest()) + + def test_failed_replay_is_atomic_and_valid_replay_can_follow(self) -> None: + path, _digest = self.journal(self.frames()) + valid = path.read_bytes() + malformed = valid + json.dumps({"stream": "O", "observed_ns": 3_000_000, "message": {"id": "unbound", "result": {}}}).encode() + b"\n" + path.write_bytes(malformed) + with self.assertRaisesRegex(codex_observer.AnalysisError, "^unbound-response$"): + self.observer.replay(path, hashlib.sha256(malformed).hexdigest()) + self.assertEqual(self.observer.report()["message_count"], 0) + self.assertEqual(self.observer.report()["replayed_message_count"], 0) + path.write_bytes(valid) + self.observer.replay(path, hashlib.sha256(valid).hexdigest()) + self.assertEqual(self.observer.report()["message_count"], 2) + with self.assertRaisesRegex(codex_observer.ObserverError, "^replay-order-mismatch$"): + self.observer.replay(path, hashlib.sha256(valid).hexdigest()) + + def test_journal_headers_and_frame_fields_are_exact(self) -> None: + header = {"schema_version": 1, "task_generation": self.binding.generation} + cases = [(dict(header, schema_version=True), "journal-binding-mismatch"), (dict(header, extra=True), "journal-binding-mismatch"), (header, "journal-frame-invalid")] + for number, (value, error) in enumerate(cases): + with self.subTest(error=error, number=number): + raw = json.dumps(value).encode() + b"\n" + if error == "journal-frame-invalid": + raw += b'{"stream":"I","observed_ns":0,"message":{},"extra":true}\n' + path = self.private_file(raw, f"invalid-{number}.jsonl") + with self.assertRaisesRegex(codex_observer.ObserverError, f"^{error}$"): + self.observer.replay(path, hashlib.sha256(raw).hexdigest()) + self.assertEqual(self.observer.report()["message_count"], 0) + + def test_private_reader_rejects_modes_links_empty_and_oversized_files(self) -> None: + path = self.private_file(b"synthetic private contents") + for mode in (0o644, 0o400, 0o666): + with self.subTest(mode=mode): + path.chmod(mode) + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-file-invalid$"): + codex_observer._read_private(path, 100) + path.chmod(0o600) + hardlink = self.root / "hardlink" + os.link(path, hardlink) + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-file-invalid$"): + codex_observer._read_private(path, 100) + hardlink.unlink() + symlink = self.root / "symlink" + symlink.symlink_to(path) + with self.assertRaises(OSError): + codex_observer._read_private(symlink, 100) + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-file-invalid$"): + codex_observer._read_private(path, 2) + path.write_bytes(b"") + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-file-invalid$"): + codex_observer._read_private(path, 100) + + def test_private_reader_rejects_nonregular_files_without_blocking(self) -> None: + fifo = self.root / "fifo" + os.mkfifo(fifo, 0o600) + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-file-invalid$"): + codex_observer._read_private(fifo, 100) + + def test_private_reader_rejects_mutation_during_read(self) -> None: + path = self.private_file(b"original") + original_read = os.read + changed = False + + def mutate(descriptor: int, count: int) -> bytes: + nonlocal changed + result = original_read(descriptor, count) + if not changed: + changed = True + path.write_bytes(b"replacement contents") + return result + + with mock.patch.object(codex_observer.os, "read", side_effect=mutate): + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-file-changed$"): + codex_observer._read_private(path, 100) + + def test_private_directory_requires_canonical_path_mode_and_owner(self) -> None: + path = self.private_file(b"synthetic") + self.root.chmod(0o755) + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-directory-required$"): + codex_observer._read_private(path, 100) + self.root.chmod(0o700) + link = self.root / "directory-link" + link.symlink_to(self.root, target_is_directory=True) + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-directory-required$"): + codex_observer._read_private(link / path.name, 100) + with mock.patch.object(codex_observer.os, "getuid", return_value=os.getuid() + 1): + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-directory-required$"): + codex_observer._read_private(path, 100) + + def test_journal_budgets_and_failed_creation_do_not_leave_partial_files(self) -> None: + path = self.root / "journal.jsonl" + with mock.patch.object(codex_observer, "MAX_JOURNAL_BYTES", 1): + with self.assertRaisesRegex(codex_observer.ObserverError, "^journal-byte-limit$"): + codex_observer.Journal(path, self.binding) + self.assertFalse(path.exists()) + journal = codex_observer.Journal(path, self.binding) + try: + with mock.patch.object(codex_observer, "MAX_COMMANDS", 1): + journal.append("I", 0, {"method": "initialized"}) + with self.assertRaisesRegex(codex_observer.ObserverError, "^journal-message-limit$"): + journal.append("I", 1, {"method": "initialized"}) + before = path.read_bytes() + with mock.patch.object(codex_observer, "MAX_JOURNAL_BYTES", len(before)): + with self.assertRaisesRegex(codex_observer.ObserverError, "^journal-byte-limit$"): + journal.append("I", 2, {"method": "initialized"}) + self.assertEqual(path.read_bytes(), before) + finally: + journal.close() + with mock.patch.object(codex_observer, "MAX_COMMANDS", 0): + with self.assertRaisesRegex(codex_observer.ObserverError, "^journal-message-limit$"): + self.observer.replay(path, hashlib.sha256(path.read_bytes()).hexdigest()) + + +class TaskBindingTests(ObserverFixture): + def environment_and_record(self) -> tuple[dict[str, str], Path, dict[str, Any]]: + state = self.root / "state" + state.mkdir(mode=0o700) + registry = state / "registry" + registry.mkdir(mode=0o700) + socket = str(self.root / "synthetic.sock") + task = "01234567-89ab-cdef-0123-456789abcdef" + bucket = registry / hashlib.sha256(socket.encode()).hexdigest()[:16] + bucket.mkdir(mode=0o700) + path = bucket / (task + ".json") + record = {"schema_version": 1, "phase": "live", "socket": socket, "id": task, "session": "agt_" + task.replace("-", "")[:12], "created": "synthetic-generation-one"} + path.write_text(json.dumps(record), encoding="utf-8") + path.chmod(0o600) + return {"CLILANE_STATE_HOME": str(state), "CLILANE_TMUX_SOCKET": socket, "CLILANE_TASK_ID": task}, path, record + + def test_generation_is_stable_for_exact_record_and_changes_with_new_task_lifetime(self) -> None: + environment, path, record = self.environment_and_record() + binding = codex_observer.TaskBinding.from_environment(environment) + self.assertEqual(binding, codex_observer.TaskBinding.from_environment(environment)) + self.assertRegex(binding.generation, "^[0-9a-f]{64}$") + record["created"] = "synthetic-generation-two" + path.write_text(json.dumps(record), encoding="utf-8") + self.assertNotEqual(binding, codex_observer.TaskBinding.from_environment(environment)) + self.assertNotIn(environment["CLILANE_TASK_ID"], binding.generation) + + def test_task_record_must_match_live_socket_id_and_session(self) -> None: + environment, path, record = self.environment_and_record() + for field, value in (("schema_version", True), ("phase", "done"), ("socket", "/another/socket"), ("id", "wrong"), ("session", "agt_wrong"), ("created", ""), ("created", "invalid\0value")): + with self.subTest(field=field, value=value): + path.write_text(json.dumps({**record, field: value}), encoding="utf-8") + with self.assertRaisesRegex(codex_observer.ObserverError, "^task-record-mismatch$"): + codex_observer.TaskBinding.from_environment(environment) + + def test_invalid_environment_missing_record_and_public_record_are_rejected(self) -> None: + environment, path, _record = self.environment_and_record() + for field, value in (("CLILANE_TMUX_SOCKET", "relative"), ("CLILANE_TASK_ID", "invalid"), ("CLILANE_TASK_ID", "01234567-89AB-cdef-0123-456789abcdef")): + with self.subTest(field=field): + with self.assertRaisesRegex(codex_observer.ObserverError, "^task-environment-invalid$"): + codex_observer.TaskBinding.from_environment({**environment, field: value}) + path.chmod(0o644) + with self.assertRaisesRegex(codex_observer.ObserverError, "^private-file-invalid$"): + codex_observer.TaskBinding.from_environment(environment) + path.unlink() + with self.assertRaisesRegex(codex_observer.ObserverError, "^task-record-not-found-for-absolute-socket$"): + codex_observer.TaskBinding.from_environment(environment) + + +class CommandBoundaryTests(ObserverFixture): + def commands(self, raw: bytes) -> list[dict[str, Any]]: + path = self.private_file(raw, "commands") + with path.open("rb") as stream: + return list(codex_observer._commands(stream.fileno())) + + def test_command_stream_accepts_exact_lines_and_rejects_truncation(self) -> None: + self.assertEqual(self.commands(b'{"op":"status"}\n{"op":"ready"}\n'), [{"op": "status"}, {"op": "ready"}]) + with self.assertRaisesRegex(codex_observer.ObserverError, "^truncated-command$"): + self.commands(b'{"op":"status"}') + with self.assertRaises(codex_observer.ProbeError): + self.commands(b'{"op":"status","op":"ready"}\n') + + def test_command_count_line_input_and_time_limits(self) -> None: + for constant, limit, raw, error in (("MAX_COMMANDS", 1, b'{}\n{}\n', "command-limit"), ("MAX_COMMAND_BYTES", 1, b'{}\n', "command-limit"), ("MAX_INPUT_BYTES", 2, b'{}\n', "input-byte-limit"), ("LIFETIME_SECONDS", 0, b'{}\n', "observer-time-limit")): + with self.subTest(constant=constant): + with mock.patch.object(codex_observer, constant, limit): + with self.assertRaisesRegex(codex_observer.ObserverError, f"^{error}$"): + self.commands(raw) + + def test_cli_failure_does_not_echo_private_argument_or_filesystem_error(self) -> None: + stdout, stderr = io.StringIO(), io.StringIO() + with mock.patch.object(codex_observer.TaskBinding, "from_environment", side_effect=OSError("private-path-sentinel")): + with redirect_stdout(stdout), redirect_stderr(stderr): + result = codex_observer.main(["--journal", "private-journal-sentinel", "--journal-sha256", "0" * 64]) + self.assertEqual(result, 1) + self.assertEqual(stdout.getvalue(), "") + self.assertEqual(stderr.getvalue(), "codex_observer.py: observer-failed\n") + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_codex_qualification.py b/tests/test_codex_qualification.py new file mode 100644 index 0000000..d70ef0d --- /dev/null +++ b/tests/test_codex_qualification.py @@ -0,0 +1,486 @@ +from __future__ import annotations + +import copy +import base64 +import hashlib +import json +import os +import shlex +import signal +import sys +import tempfile +import unittest +from pathlib import Path +from typing import Any +from unittest import mock + + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / "scripts")) +try: + import codex_qualification as qualification +finally: + sys.path.pop(0) + + +class SyntheticTransport: + def __init__(self, messages: list[dict[str, Any]] | None = None) -> None: + self.messages = copy.deepcopy(messages or []) + self.sent: list[dict[str, Any]] = [] + + def send(self, message: dict[str, Any]) -> None: + self.sent.append(copy.deepcopy(message)) + + def receive(self, deadline: float) -> dict[str, Any]: + if not self.messages: + raise qualification.ProbeError("synthetic-messages-exhausted") + return self.messages.pop(0) + + +class QualificationFixture(unittest.TestCase): + def setUp(self) -> None: + temporary = tempfile.TemporaryDirectory(prefix="codex-qualification-test-") + self.addCleanup(temporary.cleanup) + self.root = Path(temporary.name).resolve() + self.root.chmod(0o700) + self.workspace = self.root / "workspace" + self.workspace.mkdir(mode=0o700) + self.binding = qualification.TaskBinding("a" * 64) + self.broker_count = 0 + + def target(self, scenario_name: str) -> Path: + return self.root / ("denied-command" if scenario_name == "command" else "denied-file.txt") + + def item(self, scenario_name: str, status: str = "inProgress") -> dict[str, Any]: + if scenario_name == "command": + return {"id": "synthetic-item", "type": "commandExecution", "cwd": str(self.workspace), "command": "/usr/bin/touch " + shlex.quote(str(self.target(scenario_name))), "status": status, "aggregatedOutput": None, "exitCode": None, "processId": None, "durationMs": None} + return {"id": "synthetic-item", "type": "fileChange", "status": status, "changes": [{"path": str(self.target(scenario_name)), "kind": {"type": "add"}, "diff": "CLILANE_PROBE_SENTINEL\n"}]} + + def request(self, scenario_name: str) -> dict[str, Any]: + params: dict[str, Any] = {"threadId": "synthetic-thread", "turnId": "synthetic-turn", "itemId": "synthetic-item", "startedAtMs": 1} + if scenario_name == "command": + params.update(cwd=str(self.workspace), command=self.item(scenario_name)["command"]) + elif scenario_name == "file": + params["grantRoot"] = str(self.root) + elif scenario_name == "permissions": + params.update(cwd=str(self.workspace), permissions={"network": {"enabled": True}}) + else: + params.update(isBlocking=True, questions=[{"id": "choice", "question": "Alpha or Beta?", "options": [{"label": "Alpha"}, {"label": "Beta"}]}]) + return {"id": "synthetic-request", "method": qualification.SCENARIOS[scenario_name], "params": params} + + def notification(self, method: str, item: dict[str, Any]) -> dict[str, Any]: + return {"method": method, "params": {"threadId": "synthetic-thread", "turnId": "synthetic-turn", "item": item}} + + def broker(self, scenario_name: str, messages: list[dict[str, Any]] | None = None) -> tuple[qualification.Broker, SyntheticTransport]: + self.broker_count += 1 + transport = SyntheticTransport(messages) + observer = mock.Mock() + observer.ok.return_value = {"unresolved_requests": [], "unhandled_message_count": 0} + with mock.patch.object(qualification, "ObserverClient", return_value=observer): + broker = qualification.Broker(transport, self.binding, self.root / f"journal-{self.broker_count}.jsonl", scenario_name=scenario_name, workspace=self.workspace) + self.addCleanup(broker.close) + return broker, transport + + def lifecycle(self, scenario_name: str) -> list[dict[str, Any]]: + result = [{"id": "probe-init", "result": {}}, {"id": "probe-skills", "result": {"data": [{"cwd": str(self.workspace), "errors": [], "skills": []}]}}, {"id": "probe-thread", "result": {"thread": {"id": "synthetic-thread", "sessionId": "synthetic-session"}, "model": "synthetic-model"}}, {"id": "probe-turn", "result": {"turn": {"id": "synthetic-turn", "status": "inProgress"}}}] + if scenario_name in {"command", "file"}: + result.append(self.notification("item/started", self.item(scenario_name))) + result.append(self.request(scenario_name)) + if scenario_name in {"command", "file"}: + result.append(self.notification("item/completed", self.item(scenario_name, "declined"))) + result.extend([{"method": "serverRequest/resolved", "params": {"threadId": "synthetic-thread", "requestId": "synthetic-request"}}, {"method": "turn/completed", "params": {"threadId": "synthetic-thread", "turn": {"id": "synthetic-turn", "status": "completed"}}}, {"method": "thread/status/changed", "params": {"threadId": "synthetic-thread", "status": {"type": "idle"}}}]) + return result + + +class ActionValidationTests(QualificationFixture): + def test_only_exact_touch_command_and_allowed_shell_wrappers_match(self) -> None: + target = self.target("command") + command = "/usr/bin/touch " + shlex.quote(str(target)) + for value in (command, *[f"{shell} {flag} {shlex.quote(command)}" for shell in ("/bin/sh", "/bin/bash", "/bin/zsh") for flag in ("-c", "-lc")]): + with self.subTest(command=value): + self.assertTrue(qualification.command_matches(value, target)) + for value in (None, [], "x" * 4097, "'unterminated", command + " && /usr/bin/true", command + "; /usr/bin/true", command + "\n/usr/bin/true", command + " | /usr/bin/true", command + " > /tmp/other", command + " /tmp/other", command.replace("/usr/bin/touch", "touch"), "/usr/bin/touch /tmp/other", f"/bin/zsh -c {shlex.quote(command)} ignored", f"/usr/bin/env /bin/sh -c {shlex.quote(command)}"): + with self.subTest(command=value): + self.assertFalse(qualification.command_matches(value, target)) + + def test_command_item_rejects_wrong_cwd_execution_metadata_and_output(self) -> None: + for changes, error in (({"cwd": str(self.root)}, "unexpected-command-execution"), ({"command": "/usr/bin/true"}, "unexpected-command-execution"), ({"type": "fileChange"}, "unexpected-command-execution"), ({"exitCode": 0}, "unexpected-command-execution"), ({"processId": "123"}, "unexpected-command-execution"), ({"pluginId": "plugin"}, "unexpected-command-execution"), ({"scriptPath": "/tmp/script"}, "unexpected-command-execution"), ({"durationMs": True}, "unexpected-command-execution"), ({"durationMs": 1}, "unexpected-command-execution"), ({"aggregatedOutput": "executed output"}, "unexpected-command-output")): + with self.subTest(changes=changes): + with self.assertRaisesRegex(qualification.ProbeError, f"^{error}$"): + qualification.action_item({**self.item("command"), **changes}, self.workspace, "command") + with self.assertRaisesRegex(qualification.ProbeError, "^unexpected-command-output$"): + qualification.action_item({**self.item("command", "declined"), "aggregatedOutput": "executed output"}, self.workspace, "command") + + def test_file_item_requires_one_exact_addition(self) -> None: + valid = self.item("file") + qualification.action_item(valid, self.workspace, "file") + for changes in ([], valid["changes"] * 2, [{**valid["changes"][0], "path": str(self.root / "other")}], [{**valid["changes"][0], "kind": {"type": "update"}}], [{**valid["changes"][0], "diff": "different contents\n"}], [{**valid["changes"][0], "extra": True}], [None]): + with self.subTest(changes=changes): + with self.assertRaisesRegex(qualification.ProbeError, "^unexpected-file-change$"): + qualification.action_item({**valid, "changes": changes}, self.workspace, "file") + for identity in (None, "", "x" * 257, True): + with self.subTest(identity=identity): + with self.assertRaisesRegex(qualification.ProbeError, "^action-identity-invalid$"): + qualification.action_item({**valid, "id": identity}, self.workspace, "file") + + def test_existing_or_dangling_symlink_target_rejected_before_action(self) -> None: + for scenario_name in ("command", "file"): + for symlink in (False, True): + with self.subTest(scenario=scenario_name, symlink=symlink): + target = self.target(scenario_name) + if symlink: + target.symlink_to(self.root / "missing-target") + else: + target.write_text("existing sentinel", encoding="utf-8") + try: + with self.assertRaisesRegex(qualification.ProbeError, "^denied-action-created-target$"): + qualification.action_item(self.item(scenario_name), self.workspace, scenario_name) + with self.assertRaisesRegex(qualification.ProbeError, "^denied-action-created-target$"): + qualification.action_request(self.request(scenario_name)["params"], self.workspace, scenario_name, "synthetic-item") + broker, transport = self.broker(scenario_name) + with self.assertRaisesRegex(qualification.ProbeError, "^action-target-exists$"): + qualification.scenario(broker, self.workspace) + self.assertEqual(transport.sent, []) + finally: + target.unlink() + + def test_approval_request_requires_matching_item_valid_time_and_local_scope(self) -> None: + for scenario_name in ("command", "file"): + params = self.request(scenario_name)["params"] + qualification.action_request(params, self.workspace, scenario_name, "synthetic-item") + for changes in ({"itemId": "other"}, {"startedAtMs": True}, {"startedAtMs": 0}, {"startedAtMs": 1 << 63}): + with self.subTest(scenario=scenario_name, changes=changes): + with self.assertRaisesRegex(qualification.ProbeError, "^action-request-invalid$"): + qualification.action_request({**params, **changes}, self.workspace, scenario_name, "synthetic-item") + with self.assertRaisesRegex(qualification.ProbeError, "^action-request-invalid$"): + qualification.action_request(params, self.workspace, scenario_name, None) + for changes in ({"cwd": str(self.root)}, {"command": "/usr/bin/true"}, {"environmentId": "remote"}, {"networkApprovalContext": {}}, {"kind": "network"}, {"unexpected": True}): + with self.subTest(changes=changes): + with self.assertRaisesRegex(qualification.ProbeError, "^unexpected-command-request$"): + qualification.action_request({**self.request("command")["params"], **changes}, self.workspace, "command", "synthetic-item") + for changes in ({"grantRoot": "/"}, {"path": "/other"}): + with self.subTest(changes=changes): + with self.assertRaisesRegex(qualification.ProbeError, "^unexpected-file-request$"): + qualification.action_request({**self.request("file")["params"], **changes}, self.workspace, "file", "synthetic-item") + + +class BrokerDenialTests(QualificationFixture): + def test_command_prompt_quotes_target_in_root_with_spaces_and_shell_metacharacters(self) -> None: + self.root = self.root / "private run;literal" + self.root.mkdir(mode=0o700) + self.workspace = self.root / "workspace" + self.workspace.mkdir(mode=0o700) + broker, transport = self.broker("command", self.lifecycle("command")) + with mock.patch.object(qualification, "recovery", return_value={}): + result = qualification.scenario(broker, self.workspace) + turn = next(message for message in transport.sent if message.get("method") == "turn/start") + prompt = turn["params"]["input"][0]["text"] + command, _end = json.JSONDecoder().raw_decode(prompt.split("cmd=", 1)[1]) + self.assertEqual(shlex.split(command), ["/usr/bin/touch", str(self.target("command"))]) + self.assertEqual([message for message in transport.sent if "result" in message], [{"id": "synthetic-request", "result": {"decision": "decline"}}]) + self.assertTrue(result["action_declined_without_target_creation"]) + self.assertFalse(os.path.lexists(self.target("command"))) + + def test_command_and_file_scenarios_send_exactly_one_decline_and_never_grant(self) -> None: + for scenario_name in ("command", "file"): + with self.subTest(scenario=scenario_name): + broker, transport = self.broker(scenario_name, self.lifecycle(scenario_name)) + with mock.patch.object(qualification, "recovery", return_value={"synthetic_recovery": True}): + result = qualification.scenario(broker, self.workspace) + self.assertEqual([message for message in transport.sent if "result" in message], [{"id": "synthetic-request", "result": {"decision": "decline"}}]) + self.assertTrue(result[scenario_name + "_approval_resolved"]) + self.assertTrue(result["action_declined_without_target_creation"]) + self.assertTrue(broker.tool_declined) + self.assertFalse(os.path.lexists(self.target(scenario_name))) + thread = next(message for message in transport.sent if message.get("method") == "thread/start") + self.assertEqual(thread["params"]["sandbox"], "read-only") + self.assertEqual(thread["params"]["approvalPolicy"]["granular"], {"mcp_elicitations": False, "rules": False, "sandbox_approval": True, "skill_approval": False, "request_permissions": False}) + + def test_broker_blocks_grant_mutation_and_repeated_denial_before_transport(self) -> None: + broker, transport = self.broker("command", [self.notification("item/started", self.item("command")), self.request("command")]) + broker.receive(0) + broker.receive(0) + for result in ({"decision": "accept"}, {"decision": "acceptForSession"}, {"decision": "decline", "extra": True}, {"permissions": {"network": {"enabled": True}}}): + with self.subTest(result=result): + with self.assertRaisesRegex(qualification.ProbeError, "^invalid-denial-response$"): + broker.send({"id": "synthetic-request", "result": result}) + self.assertEqual(transport.sent, []) + decline = {"id": "synthetic-request", "result": {"decision": "decline"}} + broker.send(decline) + with self.assertRaisesRegex(qualification.ProbeError, "^invalid-denial-response$"): + broker.send(decline) + self.assertEqual(transport.sent, [decline]) + + def test_duplicate_or_differently_bound_approval_rejected(self) -> None: + for scenario_name in ("command", "file"): + with self.subTest(scenario=scenario_name): + start = self.notification("item/started", self.item(scenario_name)) + request = self.request(scenario_name) + broker, _transport = self.broker(scenario_name, [start, request, request]) + broker.receive(0) + broker.receive(0) + with self.assertRaisesRegex(qualification.ProbeError, "^duplicate-approval-request$"): + broker.receive(0) + wrong = copy.deepcopy(request) + wrong["params"]["itemId"] = "other-item" + broker, _transport = self.broker(scenario_name, [start, wrong]) + broker.receive(0) + with self.assertRaisesRegex(qualification.ProbeError, "^action-request-invalid$"): + broker.receive(0) + + def test_terminal_action_requires_prior_denial_exact_item_and_declined_status(self) -> None: + for scenario_name in ("command", "file"): + for status, identity, denial in (("declined", "synthetic-item", False), ("completed", "synthetic-item", True), ("failed", "synthetic-item", True), ("declined", "other-item", True)): + with self.subTest(scenario=scenario_name, status=status, identity=identity, denial=denial): + item = {**self.item(scenario_name, status), "id": identity} + broker, _transport = self.broker(scenario_name, [self.notification("item/started", self.item(scenario_name)), self.request(scenario_name), self.notification("item/completed", item)]) + broker.receive(0) + broker.receive(0) + if denial: + broker.send({"id": "synthetic-request", "result": {"decision": "decline"}}) + with self.assertRaisesRegex(qualification.ProbeError, "^action-not-declined$"): + broker.receive(0) + + def test_extra_action_start_output_delta_and_provider_warning_are_rejected(self) -> None: + cases = [(self.notification("item/started", self.item("command")), "unexpected-action-start"), ({"method": "item/commandExecution/outputDelta", "params": {"delta": "private-output"}}, "unexpected-command-output"), ({"method": "warning", "params": {"message": "private-warning"}}, "provider-warning")] + for message, error in cases: + with self.subTest(error=error): + broker, _transport = self.broker("command", [self.notification("item/started", self.item("command")), message]) + broker.receive(0) + with self.assertRaisesRegex(qualification.ProbeError, f"^{error}$"): + broker.receive(0) + + def test_input_and_permission_scenarios_cannot_execute_actions(self) -> None: + for scenario_name in ("input", "permissions"): + with self.subTest(scenario=scenario_name): + broker, _transport = self.broker(scenario_name, [self.notification("item/started", self.item("command"))]) + with self.assertRaisesRegex(qualification.ProbeError, "^provider-side-effect-attempt$"): + broker.receive(0) + + def test_action_scenario_requires_completed_turn_after_decline(self) -> None: + for status in ("failed", "interrupted", "inProgress"): + with self.subTest(status=status): + messages = self.lifecycle("command") + next(message for message in messages if message.get("method") == "turn/completed")["params"]["turn"]["status"] = status + broker, _transport = self.broker("command", messages) + with mock.patch.object(qualification, "recovery", return_value={}): + with self.assertRaisesRegex(qualification.ProbeError, "^turn-not-completed$"): + qualification.scenario(broker, self.workspace) + + def test_existing_input_and_permission_scenarios_send_empty_responses(self) -> None: + for scenario_name, expected in (("input", {"answers": {}}), ("permissions", {"permissions": {}, "scope": "turn"})): + with self.subTest(scenario=scenario_name): + broker, transport = self.broker(scenario_name, self.lifecycle(scenario_name)) + with mock.patch.object(qualification, "recovery", return_value={"synthetic_recovery": True}): + result = qualification.scenario(broker, self.workspace) + self.assertEqual([message for message in transport.sent if "result" in message], [{"id": "synthetic-request", "result": expected}]) + self.assertTrue(result["input_request_resolved" if scenario_name == "input" else "permission_approval_resolved"]) + + def test_input_rejects_nonblocking_or_multiple_questions_before_response(self) -> None: + for changes in ({"isBlocking": False}, {"isBlocking": 1}, {"questions": []}, {"questions": [{}, {}]}, {"questions": None}): + with self.subTest(changes=changes): + messages = self.lifecycle("input") + messages[4]["params"].update(changes) + broker, transport = self.broker("input", messages) + with mock.patch.object(qualification, "recovery") as recover: + with self.assertRaisesRegex(qualification.ProbeError, "^unexpected-input-request$"): + qualification.scenario(broker, self.workspace) + recover.assert_not_called() + self.assertFalse(any("result" in message for message in transport.sent)) + + def test_permission_scope_is_network_only_and_strictly_typed(self) -> None: + params = self.request("permissions")["params"] + qualification.permission_request(params, self.workspace) + for value in ({}, {"network": {"enabled": 1}}, {"network": {"enabled": False}}, {"network": {"enabled": True, "extra": True}}, {"network": {"enabled": True}, "fileSystem": {}}, {"network": {"enabled": True}, "other": True}): + with self.subTest(permissions=value): + with self.assertRaisesRegex(qualification.ProbeError, "^unexpected-permissions$"): + qualification.permission_request({**params, "permissions": value}, self.workspace) + for changes in ({"cwd": "/other"}, {"environmentId": "remote"}, {"startedAtMs": True}, {"reason": "x" * 4097}, {"extra": True}): + with self.subTest(changes=changes): + with self.assertRaisesRegex(qualification.ProbeError, "^permission-request-invalid$"): + qualification.permission_request({**params, **changes}, self.workspace) + + +class DiagnosticAndJournalTests(QualificationFixture): + def diagnostic(self, scenario_name: str) -> bytes: + suffix = b'error=exec_command failed: CreateProcess { message: "Rejected(\\"rejected by user\\")" }' if scenario_name == "command" else b"error=patch rejected by user" + return b"2026-09-26T12:34:56.123456Z ERROR codex_core::tools::router: " + suffix + b"\n" + + def test_only_one_exact_diagnostic_for_matching_action_scenario_is_allowed(self) -> None: + for scenario_name in qualification.SCENARIOS: + qualification.verify_diagnostics(b"", scenario_name) + for scenario_name in ("command", "file"): + diagnostic = self.diagnostic(scenario_name) + qualification.verify_diagnostics(diagnostic, scenario_name) + for value in (diagnostic * 2, diagnostic + b"extra private output\n", b"extra private output\n" + diagnostic, diagnostic.rstrip(b"\n"), diagnostic.replace(b" ERROR ", b" WARN "), diagnostic.replace(b"codex_core::tools::router", b"other::router"), diagnostic.replace(b"rejected by user", b"execution failed"), self.diagnostic("file" if scenario_name == "command" else "command")): + with self.subTest(scenario=scenario_name, stderr=value): + with self.assertRaisesRegex(qualification.ProbeError, "^provider-stderr-output$"): + qualification.verify_diagnostics(value, scenario_name) + for scenario_name in ("input", "permissions"): + with self.subTest(scenario=scenario_name): + with self.assertRaisesRegex(qualification.ProbeError, "^provider-stderr-output$"): + qualification.verify_diagnostics(self.diagnostic("command"), scenario_name) + + def journal_inputs(self, scenario_name: str) -> tuple[bytes, bytes, list[tuple[str, int, bytes]], dict[str, Any]]: + messages = [("I", {"method": "initialized"}), ("O", {"method": "thread/status/changed", "params": {"threadId": "synthetic-thread", "status": {"type": "idle"}}})] + capture = b"synthetic-capture-bytes" + frames = [(stream, number, (json.dumps(message) + "\n").encode()) for number, (stream, message) in enumerate(messages)] + rows = [{"schema_version": 1, "task_generation": self.binding.generation}] + [{"stream": stream, "observed_ns": number, "message": message} for number, (stream, message) in enumerate(messages)] + journal = b"".join((json.dumps(row) + "\n").encode() for row in rows) + return capture, journal, frames, {"source_capture_sha256": hashlib.sha256(capture).hexdigest()} + + def verify_frames(self, frames: list[tuple[str, int, bytes]], scenario_name: str) -> None: + capture = b"synthetic ordered capture" + rows = [{"schema_version": 1, "task_generation": self.binding.generation}] + rows.extend({"stream": stream, "observed_ns": moment, "message": json.loads(raw)} for stream, moment, raw in frames if stream != "E") + journal = b"".join((json.dumps(row) + "\n").encode() for row in rows) + with mock.patch.object(qualification, "_read_private", side_effect=[capture, journal]), mock.patch.object(qualification, "iter_capture_frames", return_value=iter(sorted(frames, key=lambda frame: frame[1]))): + qualification.verify_journal(self.root, self.binding, {"source_capture_sha256": hashlib.sha256(capture).hexdigest()}, scenario_name=scenario_name) + + def denial_frames(self, scenario_name: str) -> list[tuple[str, int, bytes]]: + messages = [("O", 1, self.request(scenario_name)), ("I", 3, {"id": "synthetic-request", "result": {"decision": "decline"}}), ("O", 7, {"method": "turn/completed", "params": {"threadId": "synthetic-thread", "turn": {"id": "synthetic-turn", "status": "completed"}}})] + return [(stream, moment, (json.dumps(message) + "\n").encode()) for stream, moment, message in messages] + + def test_denial_diagnostic_must_be_within_exact_decline_completion_window(self) -> None: + for scenario_name in ("command", "file"): + frames = self.denial_frames(scenario_name) + diagnostic = self.diagnostic(scenario_name) + for moment in (3, 6): + with self.subTest(scenario=scenario_name, accepted=moment): + self.verify_frames(frames + [("E", moment, diagnostic)], scenario_name) + for moment in (0, 2, 7, 8): + with self.subTest(scenario=scenario_name, rejected=moment): + with self.assertRaisesRegex(qualification.ProbeError, "^denial-diagnostic-outside-turn$"): + self.verify_frames(frames + [("E", moment, diagnostic)], scenario_name) + midpoint = len(diagnostic) // 2 + self.verify_frames(frames + [("E", 3, diagnostic[:midpoint]), ("E", 6, diagnostic[midpoint:]), ("E", 8, b"")], scenario_name) + with self.assertRaisesRegex(qualification.ProbeError, "^denial-diagnostic-outside-turn$"): + self.verify_frames(frames + [("E", 3, diagnostic[:midpoint]), ("E", 7, diagnostic[midpoint:])], scenario_name) + + def test_denial_diagnostic_requires_one_matching_request_decline_and_completion(self) -> None: + frames = self.denial_frames("command") + diagnostic = ("E", 4, self.diagnostic("command")) + for changed, error in ((frames[1:], "denial-diagnostic-unbound"), (frames + [frames[0]], "denial-diagnostic-unbound"), ([frames[0], frames[2]], "denial-diagnostic-outside-turn"), (frames + [frames[1]], "denial-diagnostic-outside-turn"), (frames[:2], "denial-diagnostic-outside-turn"), (frames + [frames[2]], "denial-diagnostic-outside-turn")): + with self.subTest(error=error, frame_count=len(changed)): + with self.assertRaisesRegex(qualification.ProbeError, f"^{error}$"): + self.verify_frames(changed + [diagnostic], "command") + for identity in ("wrong-request", 7, True): + with self.subTest(identity=identity): + changed = list(frames) + changed[1] = ("I", 3, (json.dumps({"id": identity, "result": {"decision": "decline"}}) + "\n").encode()) + with self.assertRaisesRegex(qualification.ProbeError, "^denial-diagnostic-outside-turn$"): + self.verify_frames(changed + [diagnostic], "command") + changed = list(frames) + request = self.request("command") + request["id"] = 1 + changed[0] = ("O", 1, (json.dumps(request) + "\n").encode()) + changed[1] = ("I", 3, b'{"id":true,"result":{"decision":"decline"}}\n') + with self.assertRaisesRegex(qualification.ProbeError, "^denial-diagnostic-outside-turn$"): + self.verify_frames(changed + [diagnostic], "command") + + def test_journal_verification_binds_capture_hash_task_messages_and_scenario(self) -> None: + capture, journal, frames, report = self.journal_inputs("input") + with mock.patch.object(qualification, "_read_private", side_effect=[capture, journal]), mock.patch.object(qualification, "iter_capture_frames", return_value=iter(frames)): + qualification.verify_journal(self.root, self.binding, report) + with mock.patch.object(qualification, "_read_private", return_value=capture): + with self.assertRaisesRegex(qualification.ProbeError, "^capture-changed$"): + qualification.verify_journal(self.root, self.binding, {"source_capture_sha256": "0" * 64}) + wrong = journal.replace(self.binding.generation.encode(), b"b" * 64) + with mock.patch.object(qualification, "_read_private", side_effect=[capture, wrong]), mock.patch.object(qualification, "iter_capture_frames", return_value=iter(frames)): + with self.assertRaisesRegex(qualification.ProbeError, "^journal-binding-mismatch$"): + qualification.verify_journal(self.root, self.binding, report) + wrong = journal.replace(b"initialized", b"different") + with mock.patch.object(qualification, "_read_private", side_effect=[capture, wrong]), mock.patch.object(qualification, "iter_capture_frames", return_value=iter(frames)): + with self.assertRaisesRegex(qualification.ProbeError, "^journal-source-mismatch$"): + qualification.verify_journal(self.root, self.binding, report) + with mock.patch.object(qualification, "_read_private", side_effect=[capture, journal]), mock.patch.object(qualification, "iter_capture_frames", return_value=iter(frames + [("E", 2, self.diagnostic("file"))])): + with self.assertRaisesRegex(qualification.ProbeError, "^provider-stderr-output$"): + qualification.verify_journal(self.root, self.binding, report, scenario_name="command") + + def test_journal_rejects_truncated_protocol_line(self) -> None: + capture, journal, frames, report = self.journal_inputs("input") + frames[-1] = (frames[-1][0], frames[-1][1], frames[-1][2].rstrip(b"\n")) + with mock.patch.object(qualification, "_read_private", side_effect=[capture, journal]), mock.patch.object(qualification, "iter_capture_frames", return_value=iter(frames)): + with self.assertRaisesRegex(qualification.ProbeError, "^capture-message-truncated$"): + qualification.verify_journal(self.root, self.binding, report) + + +class CleanupAndIsolationTests(QualificationFixture): + def test_run_failure_cleans_credential_copy_and_preserves_replacement_inode(self) -> None: + claims = base64.urlsafe_b64encode(b'{"exp":4102444800}').decode().rstrip("=") + synthetic = json.dumps({"tokens": {"access_token": "synthetic." + claims + ".synthetic"}}).encode() + pinned = qualification.PinnedExecutable(Path("/synthetic/executable"), "b" * 64, (1, 1, 1, 1)) + for replacement in (False, True): + with self.subTest(replacement=replacement): + root = self.root / ("replacement-run" if replacement else "failed-run") + root.mkdir(mode=0o700) + + def fail_transport(*args: Any, **kwargs: Any) -> None: + path = root / "config/auth.json" + self.assertEqual(path.read_bytes(), synthetic) + if replacement: + path.rename(root / "displaced-copy") + path.write_bytes(b"synthetic replacement") + raise qualification.ProbeError("synthetic-spawn-failed") + + with mock.patch.object(qualification, "task_binding", return_value=self.binding), mock.patch.object(qualification, "load_executable_manifest", return_value={}), mock.patch.object(qualification, "pin_executable", return_value=pinned), mock.patch.object(qualification, "preflight", return_value=[]), mock.patch.object(qualification, "_read_private", return_value=synthetic), mock.patch.object(qualification, "JsonLineTransport", side_effect=fail_transport): + with self.assertRaisesRegex(qualification.ProbeError, "^credential-copy-replaced$" if replacement else "^synthetic-spawn-failed$"): + qualification.run(root, pinned.path, Path("/synthetic/input")) + if replacement: + self.assertEqual((root / "config/auth.json").read_bytes(), b"synthetic replacement") + else: + self.assertFalse((root / "config/auth.json").exists()) + + def test_credential_cleanup_removes_only_original_inode_and_is_idempotent(self) -> None: + path = self.root / "synthetic-credential-copy" + path.write_bytes(b"synthetic contents") + identity = path.stat() + qualification.remove_credentials(path, identity) + self.assertFalse(path.exists()) + qualification.remove_credentials(path, identity) + path.write_bytes(b"first") + identity = path.stat() + displaced = self.root / "displaced" + path.rename(displaced) + path.write_bytes(b"replacement") + with self.assertRaisesRegex(qualification.ProbeError, "^credential-copy-replaced$"): + qualification.remove_credentials(path, identity) + self.assertEqual(path.read_bytes(), b"replacement") + self.assertEqual(displaced.read_bytes(), b"first") + + def test_cancellation_cleans_copy_and_restores_handlers_without_real_signals(self) -> None: + with mock.patch.object(qualification.signal, "signal", return_value="previous") as install: + cancellation = qualification.Cancellation() + cleanup = mock.Mock() + cancellation.cleanup = cleanup + with self.assertRaisesRegex(qualification.ProbeError, "^qualification-cancelled$"): + cancellation.handle(signal.SIGTERM, None) + cleanup.assert_called_once_with() + cancellation.close() + self.assertEqual(install.call_count, 9) + for number in (signal.SIGTERM, signal.SIGHUP, signal.SIGINT): + self.assertIn(mock.call(number, signal.SIG_IGN), install.call_args_list) + self.assertIn(mock.call(number, "previous"), install.call_args_list) + + def test_environment_drops_unlisted_values_without_mutating_parent(self) -> None: + environment = {"HOME": "/synthetic/home", "PATH": "/usr/bin:/bin", "LANG": "C", "OPENAI_API_KEY": "synthetic-denied", "CLILANE_PRIVATE_VALUE": "synthetic-denied"} + with mock.patch.dict(os.environ, environment, clear=True): + before = dict(os.environ) + actual = qualification.environment(self.root / "config", self.root / "tmp") + self.assertEqual(dict(os.environ), before) + self.assertEqual(actual["CODEX_HOME"], str(self.root / "config")) + self.assertEqual(actual["TMPDIR"], str(self.root / "tmp")) + self.assertNotIn("OPENAI_API_KEY", actual) + self.assertNotIn("CLILANE_PRIVATE_VALUE", actual) + + def test_skills_preflight_refuses_enabled_skills_in_isolation(self) -> None: + valid = {"id": "probe-skills", "result": {"data": [{"cwd": str(self.workspace), "errors": [], "skills": [{"enabled": False, "path": "/synthetic/skill"}]}]}} + transport = SyntheticTransport([valid]) + self.assertEqual(qualification.skills(transport, self.workspace, isolated=True), ["/synthetic/skill"]) + valid["result"]["data"][0]["skills"][0]["enabled"] = True + with self.assertRaisesRegex(qualification.ProbeError, "^host-skills-enabled$"): + qualification.skills(SyntheticTransport([valid]), self.workspace, isolated=True) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_provider_evidence.py b/tests/test_provider_evidence.py index 2066004..b781152 100644 --- a/tests/test_provider_evidence.py +++ b/tests/test_provider_evidence.py @@ -720,7 +720,7 @@ def test_executable_manifest_pins_exact_provider_binaries(self) -> None: self.assertEqual( expected, { - "claude": "9f7c2260251765a18d0b35198669dacc1912f6e8129a3b01f6b58d93365ff1f1", + "claude": "d8cb1e5c79684cc12a8bfc813e3a2073406921b6245744b3009be3ab5651d21e", "codex": "8eaf1ad12fe6bf89b1710330f58900014322c7c5af677e43be116d8ac5fc0a9e", "kimi": "92bf3b4b6643e7c4cc12c82e5680cc5b54a5a6768a301de815e5e9a02d2184bb", }, diff --git a/tests/test_provider_probe.py b/tests/test_provider_probe.py index fcd57cb..9e12dc9 100644 --- a/tests/test_provider_probe.py +++ b/tests/test_provider_probe.py @@ -242,7 +242,7 @@ def run_hook(settings, event_name, value): fail(87) if sys.argv[1:] == ["--version"]: - sys.stdout.write("2.1.245 (Claude Code)\\n") + sys.stdout.write("2.1.283 (Claude Code)\\n") fail(0) arguments = sys.argv[1:] if arguments != {expected_arguments!r}: @@ -294,7 +294,7 @@ def fake_claude_with_descendant(path: Path, process_group_path: Path) -> Path: import time if sys.argv[1:] == ["--version"]: - sys.stdout.write("2.1.245 (Claude Code)\\n") + sys.stdout.write("2.1.283 (Claude Code)\\n") raise SystemExit(0) child = subprocess.Popen( [sys.executable, "-c", "import signal,time; signal.signal(signal.SIGTERM, signal.SIG_IGN); time.sleep(60)"], @@ -455,7 +455,7 @@ def test_claude_uses_exact_bounded_cli_and_three_private_hooks(self) -> None: self.config("claude", executable, "capture-claude-success") ) self.assertEqual(result.provider, "claude") - self.assertEqual(result.provider_version, "2.1.245") + self.assertEqual(result.provider_version, "2.1.283") self.assertEqual(result.message_count, 3) for suffix in provider_probe.CLAUDE_CAPTURE_SUFFIXES: receipt = self.receipt(f"capture-claude-success-{suffix}")