diff --git a/CHANGELOG.md b/CHANGELOG.md index ed28c74..9a1769b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -4,6 +4,14 @@ All notable changes to Skivolve are documented in this file. The format follows ## [Unreleased] +### Changed + +- Refreshed the Codex app-server runtime lock from 0.144.3 to 0.146.0, including the executable, bundled tools, and generated protocol schema. + +### Fixed + +- Accepted bounded Codex collaboration lifecycles, matched wait and send-input completions to their starts, recorded child status and token evidence without retaining unbounded messages, kept child-agent traffic from mutating the main result, and cleaned deeply nested unreadable runtime directories without following symlinks. + ## [0.5.0] - 2026-07-29 ### Changed diff --git a/codex-app-server-lock.json b/codex-app-server-lock.json index bc5a3a2..1f03b0b 100644 --- a/codex-app-server-lock.json +++ b/codex-app-server-lock.json @@ -1,6 +1,6 @@ { - "codex_cli_version": "codex-cli 0.144.3", - "executable_sha256": "37e6f5953f191b04f7b62cb07dae90f51d0947ad89f0355665b421fbde28700b", + "codex_cli_version": "codex-cli 0.146.0", + "executable_sha256": "2e863156ed35ecc5253b1e2f907a9143077b9f7cb51942070c61996471ff6e04", "models": { "gpt-5.6-luna": { "reasoning_efforts": [ @@ -24,7 +24,7 @@ }, "protocol": { "bundle": "codex_app_server_protocol.v2.schemas.json", - "canonical_bytes": 308970, + "canonical_bytes": 331046, "canonicalization": "json-sort-keys-compact-ascii-v1", "generate_argv": [ "app-server", @@ -33,17 +33,17 @@ "--out", "{output_dir}" ], - "sha256": "f5e8d20f3a8f9bb5e5b23ab0c5aa6bde7b12e7e0713606c5d0132651a4959d37" + "sha256": "e554a74bd59d38d16acb1744750b2999156ee3d65d0fe906b22ab52edf17fbbc" }, "runtime_bundle": { "canonicalization": "json-sort-keys-compact-ascii-v1", "files": { - "bin/codex": "37e6f5953f191b04f7b62cb07dae90f51d0947ad89f0355665b421fbde28700b", - "codex-path/rg": "ebeaf56f8a25e102e9419933423738b3a2a613a444fd749d695e15eba53f71f2", + "bin/codex": "2e863156ed35ecc5253b1e2f907a9143077b9f7cb51942070c61996471ff6e04", + "codex-path/rg": "e62198eb19b136b88c330af83647b5a962cb99b6b1f066758568f12de1974849", "codex-resources/bwrap": "77360cb751ccedc5971391444ac86a8a33c15b04d6b4a6fe45f5d25496e62c4c", "codex-resources/zsh/bin/zsh": "67faaaa89242c4a332e16e508a1977cffc24bf7fca31d4411cdfd101f3831ef3" }, - "sha256": "dd0959589aaf01d8e9f18147255d885ca8eb9e5b4373575225da98698498bcdf" + "sha256": "134da631c1d438ae8c28fe3ac24aabf9bc38828a1892038cb650ab0102b4188c" }, "schema_version": 1 } diff --git a/skivolve/codex_app_server.py b/skivolve/codex_app_server.py index cd82d43..d9a82f0 100644 --- a/skivolve/codex_app_server.py +++ b/skivolve/codex_app_server.py @@ -34,6 +34,16 @@ ) +class ProviderTimeoutError(ProviderError): + """Raised when a Codex operation exceeds its configured deadline.""" + + def __init__(self, message: str, *, cleanup_confirmed: bool = False) -> None: + if type(cleanup_confirmed) is not bool: + raise TypeError("cleanup_confirmed must be a bool") + super().__init__(message) + self.cleanup_confirmed = cleanup_confirmed + + _MAX_FRAME_BYTES = 8 * 1024 * 1024 _MAX_MESSAGES = 20_000 _MAX_PAGES = 64 @@ -41,6 +51,8 @@ _MAX_MODELS = 4_096 _MAX_SKILLS = 1_024 _MAX_COMPLETED_MESSAGES = 64 +_MAX_COLLAB_THREADS = 32 +_MAX_COLLAB_TURNS = 128 _MAX_RETAINED_TEXT_BYTES = 8 * 1024 * 1024 _MAX_RATE_LIMIT_BUCKETS = 32 _MAX_AUTH_BYTES = 1024 * 1024 @@ -122,6 +134,7 @@ _ALLOWED_ITEM_TYPES = frozenset( { "agentMessage", + "collabAgentToolCall", "commandExecution", "contextCompaction", "fileChange", @@ -185,6 +198,12 @@ def as_json(self) -> dict[str, int | str]: return asdict(self) +@dataclass(frozen=True) +class _WorkspaceRuntime: + parent_descriptor: int + directories: tuple[tuple[str, int], ...] + + class _RecoveryProbe(Protocol): def confirm_unit_clean( self, @@ -474,7 +493,7 @@ def _require_exact_keys( def _remaining(deadline: float, label: str) -> float: remaining = deadline - time.monotonic() if remaining <= 0: - raise ProviderError(f"{label} timed out") + raise ProviderTimeoutError(f"{label} timed out") return remaining @@ -1228,11 +1247,36 @@ def __init__( self._expected_codex_home = expected_codex_home self._on_dispatched = on_dispatched self._thread_id: str | None = None + self._session_id: str | None = None self._announced_thread_id: str | None = None self._turn_id: str | None = None self._announced_turn_id: str | None = None self._turn_completed: dict[str, Any] | None = None self._completed_messages: list[dict[str, str | None]] = [] + self._collab_thread_ids: set[str] = set() + self._validated_collab_thread_ids: set[str] = set() + self._collab_parent_ids: dict[str, str] = {} + self._collab_depths: dict[str, int] = {} + self._collab_turn_ids: dict[str, set[str]] = {} + self._active_collab_thread_ids: set[str] = set() + self._seen_collab_turn_ids: set[tuple[str, str]] = set() + self._collab_turn_usage: dict[tuple[str, str], dict[str, int]] = {} + self._collab_turn_statuses = { + "completed": 0, + "failed": 0, + "interrupted": 0, + } + self._pending_spawn_items: dict[ + tuple[str, str], tuple[str | None, bytes | None] + ] = {} + self._terminal_collab_history: dict[tuple[str, str], bytes] = {} + self._terminal_collab_item_ids: set[tuple[str, str]] = set() + self._completed_spawn_edges: set[tuple[str, str]] = set() + self._active_nonspawn_items: dict[ + tuple[str, str], tuple[str, tuple[str, ...], bytes | None] + ] = {} + self._unbound_collab_thread_ids: set[str] = set() + self._failed_collab_thread_ids: set[str] = set() self._retained_text_bytes = 0 self._last_usage: dict[str, int] | None = None self._rate_limits: dict[str, Any] | None = None @@ -1254,6 +1298,12 @@ def run(self, prompt: str, deadline: float) -> _TurnOutcome: after_limits = self._read_rate_limits(deadline) if self._last_usage is None: raise ProviderError("Codex turn omitted last-turn token usage") + total_usage = dict(self._last_usage) + child_usage = {key: 0 for key in self._last_usage} + for usage in self._collab_turn_usage.values(): + for key, value in usage.items(): + total_usage[key] += value + child_usage[key] += value quota = { "before": before_limits, "rolling": self._rate_limits, @@ -1262,16 +1312,22 @@ def run(self, prompt: str, deadline: float) -> _TurnOutcome: assert self._thread_id is not None and self._turn_id is not None raw = { "account": account, + "collaboration": { + "child_thread_count": len(self._validated_collab_thread_ids), + "child_turn_count": len(self._seen_collab_turn_ids), + "child_turn_statuses": dict(self._collab_turn_statuses), + "usage": child_usage, + }, "model": self._model, "reasoning_effort": self._reasoning_effort, "thread_id_sha256": _opaque_sha256(self._thread_id), "turn": turn, "turn_id_sha256": _opaque_sha256(self._turn_id), - "usage": dict(self._last_usage), + "usage": dict(total_usage), } return _TurnOutcome( final_output=final_output, - tokens=dict(self._last_usage), + tokens=total_usage, quota=quota, raw_response=raw, ) @@ -1560,7 +1616,7 @@ def _start_thread(self, deadline: float) -> None: ) if created_at is None or updated_at is None or updated_at < created_at: raise ProviderError("Codex thread timestamps are missing or inconsistent") - _require_protocol_id(thread.get("sessionId"), "thread.sessionId") + session_id = _require_protocol_id(thread.get("sessionId"), "thread.sessionId") status = _require_object(thread.get("status"), "thread.status") _require_exact_keys(status, "thread.status", required={"type"}) if status != {"type": "idle"} or thread.get("preview") != "": @@ -1594,6 +1650,7 @@ def _start_thread(self, deadline: float) -> None: thread_id = _require_protocol_id(thread.get("id"), "thread.id") if self._announced_thread_id not in {None, thread_id}: raise ProviderError("Codex thread announcement disagrees with thread/start") + self._session_id = session_id self._thread_id = thread_id def _start_turn(self, prompt: str, deadline: float) -> None: @@ -1626,10 +1683,107 @@ def _start_turn(self, prompt: str, deadline: float) -> None: def _handle_notification(self, method: str, raw_params: Any) -> None: params = _require_object(raw_params, "notification params") + if self._turn_completed is not None and ( + method.startswith(("item/", "model/", "turn/")) + or method + in { + "error", + "thread/started", + "thread/status/changed", + "thread/tokenUsage/updated", + } + ): + raise ProviderError("Codex emitted turn traffic after root completion") if method == "model/rerouted": raise ProviderError("Codex rerouted the pinned model") if method == "error": - _require_object(params.get("error"), "error notification") + _require_exact_keys( + params, + "error notification", + required={"error", "threadId", "turnId", "willRetry"}, + ) + error = _require_object(params.get("error"), "error notification.error") + _require_exact_keys( + error, + "error notification.error", + required={"message"}, + optional={"additionalDetails", "codexErrorInfo"}, + ) + _require_string(error.get("message"), "error notification.error.message") + additional_details = error.get("additionalDetails") + if additional_details is not None and not isinstance( + additional_details, str + ): + raise ProviderError( + "error notification additionalDetails must be a string or null" + ) + error_info = error.get("codexErrorInfo") + if isinstance(error_info, str): + if error_info not in { + "badRequest", + "contextWindowExceeded", + "cyberPolicy", + "internalServerError", + "other", + "sandboxError", + "serverOverloaded", + "sessionBudgetExceeded", + "threadRollbackFailed", + "unauthorized", + "usageLimitExceeded", + }: + raise ProviderError( + "error notification has unknown Codex error info" + ) + elif error_info is not None: + info = _require_object(error_info, "error notification.codexErrorInfo") + if len(info) != 1: + raise ProviderError( + "error notification has invalid Codex error info" + ) + variant, raw_info = next(iter(info.items())) + variant_info = _require_object( + raw_info, f"error notification.codexErrorInfo.{variant}" + ) + if variant in { + "httpConnectionFailed", + "responseStreamConnectionFailed", + "responseStreamDisconnected", + "responseTooManyFailedAttempts", + }: + _require_exact_keys( + variant_info, + f"error notification.codexErrorInfo.{variant}", + required=set(), + optional={"httpStatusCode"}, + ) + _optional_bounded_integer( + variant_info.get("httpStatusCode"), + "error notification HTTP status", + maximum=65535, + ) + elif variant == "activeTurnNotSteerable": + _require_exact_keys( + variant_info, + "error notification active-turn error info", + required={"turnKind"}, + ) + if variant_info.get("turnKind") not in {"review", "compact"}: + raise ProviderError( + "error notification has unknown active turn kind" + ) + else: + raise ProviderError( + "error notification has unknown Codex error info" + ) + main_turn = self._matches_turn(params) + if not main_turn and not self._matches_collab_turn(params): + raise ProviderError("Codex error notification changed turn scope") + will_retry = params.get("willRetry") + if type(will_retry) is not bool: + raise ProviderError("Codex error notification has invalid retry state") + if will_retry or not main_turn: + return raise ProviderError("Codex reported a turn error") if method == "account/rateLimits/updated": update = _sanitize_rate_snapshot( @@ -1676,6 +1830,187 @@ def _handle_notification(self, method: str, raw_params: Any) -> None: announced = _require_protocol_id( thread.get("id"), "thread/started.thread.id" ) + if self._thread_id is not None and announced != self._thread_id: + _require_exact_keys( + thread, + "child thread", + required={ + "cliVersion", + "createdAt", + "cwd", + "ephemeral", + "id", + "modelProvider", + "preview", + "sessionId", + "source", + "status", + "turns", + "updatedAt", + }, + optional={ + "agentNickname", + "agentRole", + "canAcceptDirectInput", + "extra", + "forkedFromId", + "gitInfo", + "historyMode", + "isPinned", + "name", + "parentThreadId", + "path", + "recencyAt", + "threadSource", + }, + ) + for field in ("agentNickname", "agentRole", "name"): + value = thread.get(field) + if value is not None and not isinstance(value, str): + raise ProviderError( + f"child thread.{field} must be a string or null" + ) + direct_input = thread.get("canAcceptDirectInput") + if direct_input is not None and type(direct_input) is not bool: + raise ProviderError( + "child thread.canAcceptDirectInput must be a bool or null" + ) + if "isPinned" in thread and type(thread["isPinned"]) is not bool: + raise ProviderError("child thread.isPinned must be a bool") + if thread.get("extra") is not None: + _require_object(thread["extra"], "child thread.extra") + git_info = thread.get("gitInfo") + if git_info is not None: + git_info = _require_object(git_info, "child thread.gitInfo") + _require_exact_keys( + git_info, + "child thread.gitInfo", + required=set(), + optional={"branch", "originUrl", "sha"}, + ) + for field in ("branch", "originUrl", "sha"): + value = git_info.get(field) + if value is not None and not isinstance(value, str): + raise ProviderError( + f"child thread.gitInfo.{field} must be a string or null" + ) + _optional_bounded_integer( + thread.get("recencyAt"), "child thread.recencyAt" + ) + if announced in self._validated_collab_thread_ids: + raise ProviderError("Codex announced a child thread more than once") + created_at = _optional_bounded_integer( + thread.get("createdAt"), "child thread.createdAt" + ) + updated_at = _optional_bounded_integer( + thread.get("updatedAt"), "child thread.updatedAt" + ) + session_id = _require_protocol_id( + thread.get("sessionId"), "child thread.sessionId" + ) + parent_id = _require_protocol_id( + thread.get("parentThreadId"), "child thread.parentThreadId" + ) + source = _require_object(thread.get("source"), "child thread.source") + _require_exact_keys( + source, "child thread.source", required={"subAgent"} + ) + subagent = _require_object( + source.get("subAgent"), "child thread.source.subAgent" + ) + _require_exact_keys( + subagent, + "child thread.source.subAgent", + required={"thread_spawn"}, + ) + spawn = _require_object( + subagent.get("thread_spawn"), + "child thread.source.subAgent.thread_spawn", + ) + _require_exact_keys( + spawn, + "child thread.source.subAgent.thread_spawn", + required={"depth", "parent_thread_id"}, + optional={"agent_nickname", "agent_path", "agent_role"}, + ) + for field in ("agent_nickname", "agent_path", "agent_role"): + value = spawn.get(field) + if value is not None and not isinstance(value, str): + raise ProviderError( + f"child thread.source {field} must be a string or null" + ) + source_parent_id = _require_protocol_id( + spawn.get("parent_thread_id"), + "child thread.source parent thread id", + ) + bound_parent_id = self._collab_parent_ids.get(announced) + parent_depth = ( + 0 + if parent_id == self._thread_id + else self._collab_depths.get(parent_id) + ) + depth = _optional_bounded_integer( + spawn.get("depth"), + "child thread.source depth", + minimum=1, + maximum=_MAX_COLLAB_THREADS, + ) + self._validate_thread_status( + _require_object(thread.get("status"), "child thread.status"), + "child thread.status", + ) + preview = thread.get("preview") + if not isinstance(preview, str): + raise ProviderError("child thread.preview must be a string") + if len(preview.encode("utf-8")) > _MAX_RETAINED_TEXT_BYTES: + raise ProviderError("child thread.preview exceeds the byte limit") + if ( + created_at is None + or updated_at is None + or updated_at < created_at + or depth is None + or parent_depth is None + or depth != parent_depth + 1 + or self._session_id is None + or session_id != self._session_id + or parent_id != source_parent_id + or (bound_parent_id is not None and bound_parent_id != parent_id) + or ( + parent_id != self._thread_id + and parent_id not in self._validated_collab_thread_ids + ) + or thread.get("modelProvider") != "openai" + or thread.get("cliVersion") != self._locked_thread_cli_version + or thread.get("cwd") != str(self._workspace) + or thread.get("ephemeral") is not True + or thread.get("historyMode") != "paginated" + or thread.get("path") is not None + or thread.get("forkedFromId") is not None + or thread.get("threadSource") != "skill-eval" + or thread.get("turns") != [] + ): + raise ProviderError( + "Codex child thread provenance differs from the root request" + ) + if not ( + any( + sender == parent_id and pending[0] in {None, announced} + for ( + sender, + _item_id, + ), pending in self._pending_spawn_items.items() + ) + or (parent_id, announced) in self._completed_spawn_edges + ): + raise ProviderError( + "Codex child thread lacked parent-owned spawn history" + ) + if not self._claim_collab_thread_scope(announced): + raise ProviderError("Codex thread announcement changed scope") + self._validated_collab_thread_ids.add(announced) + self._collab_parent_ids[announced] = parent_id + self._collab_depths[announced] = depth + return if self._announced_thread_id is not None: raise ProviderError("Codex announced more than one thread") self._announced_thread_id = announced @@ -1684,26 +2019,69 @@ def _handle_notification(self, method: str, raw_params: Any) -> None: thread_id = _require_protocol_id( params.get("threadId"), "turn/started.threadId" ) - if self._thread_id is None or thread_id != self._thread_id: - raise ProviderError("Codex turn announcement changed thread scope") turn = _require_object(params.get("turn"), "turn/started.turn") announced = _require_protocol_id(turn.get("id"), "turn/started.turn.id") + if self._thread_id is None: + raise ProviderError("Codex turn announcement changed thread scope") + if thread_id != self._thread_id: + if thread_id not in self._validated_collab_thread_ids: + raise ProviderError( + "Codex child turn preceded its provenance announcement" + ) + turns = self._collab_turn_ids.setdefault(thread_id, set()) + turn_scope = (thread_id, announced) + if turn_scope in self._seen_collab_turn_ids: + raise ProviderError("Codex announced a child turn more than once") + if len(self._seen_collab_turn_ids) >= _MAX_COLLAB_TURNS: + raise ProviderError("collaboration turn count exceeds the limit") + turns.add(announced) + self._seen_collab_turn_ids.add(turn_scope) + return if self._announced_turn_id is not None: raise ProviderError("Codex announced more than one turn") self._announced_turn_id = announced return if method == "thread/tokenUsage/updated": - if not self._matches_turn(params): + main_turn = self._matches_turn(params) + if not main_turn and not self._matches_collab_turn(params): raise ProviderError("Codex token usage targeted an unknown turn") usage = _require_object(params.get("tokenUsage"), "tokenUsage") last = _require_object(usage.get("last"), "tokenUsage.last") - self._last_usage = self._validate_usage(last) + validated = self._validate_usage(last) + if main_turn: + self._last_usage = validated + else: + thread_id = _require_protocol_id( + params.get("threadId"), "tokenUsage.threadId" + ) + turn_id = _require_protocol_id( + params.get("turnId"), "tokenUsage.turnId" + ) + self._collab_turn_usage[(thread_id, turn_id)] = validated return if method == "item/completed": - if not self._matches_turn(params): + main_turn = self._matches_turn(params) + if not main_turn and not self._matches_collab_turn(params): raise ProviderError("Codex item completion targeted an unknown turn") + if ( + _optional_bounded_integer( + params.get("completedAtMs"), + "item/completed.completedAtMs", + minimum=-(2**63), + ) + is None + ): + raise ProviderError("item/completed.completedAtMs is required") item = _require_object(params.get("item"), "item/completed.item") - self._validate_item(item) + self._validate_item( + item, + owner_thread_id=_require_protocol_id( + params.get("threadId"), "item/completed.threadId" + ), + lifecycle="completed", + ) + if not main_turn: + return if item.get("type") == "agentMessage": if len(self._completed_messages) >= _MAX_COMPLETED_MESSAGES: raise ProviderError( @@ -1723,29 +2101,71 @@ def _handle_notification(self, method: str, raw_params: Any) -> None: ) return if method == "item/started": - if not self._matches_turn(params): + if not self._matches_turn(params) and not self._matches_collab_turn(params): raise ProviderError("Codex item start targeted an unknown turn") + if ( + _optional_bounded_integer( + params.get("startedAtMs"), + "item/started.startedAtMs", + minimum=-(2**63), + ) + is None + ): + raise ProviderError("item/started.startedAtMs is required") self._validate_item( - _require_object(params.get("item"), "item/started.item") + _require_object(params.get("item"), "item/started.item"), + owner_thread_id=_require_protocol_id( + params.get("threadId"), "item/started.threadId" + ), + lifecycle="started", ) return if method == "turn/completed": completed_thread_id = _require_protocol_id( params.get("threadId"), "turn/completed.threadId" ) - if self._thread_id is None or completed_thread_id != self._thread_id: - raise ProviderError("Codex completed an unknown thread") turn = _require_object(params.get("turn"), "turn/completed.turn") completed_turn_id = _require_protocol_id( turn.get("id"), "turn/completed.turn.id" ) + encoded_size = len(_canonical_json(turn, "turn/completed.turn")) + if encoded_size > _MAX_RETAINED_TEXT_BYTES: + raise ProviderError("completed turn exceeds the retained-byte limit") + if self._thread_id is None: + raise ProviderError("Codex completed an unknown thread") + if completed_thread_id != self._thread_id: + if completed_turn_id not in self._collab_turn_ids.get( + completed_thread_id, set() + ): + raise ProviderError("Codex completed an unknown child turn") + child_status = self._validate_completed_turn(turn, completed_thread_id) + if ( + completed_thread_id, + completed_turn_id, + ) not in self._collab_turn_usage: + raise ProviderError("Codex child turn omitted token usage") + self._collab_turn_ids[completed_thread_id].remove(completed_turn_id) + self._active_collab_thread_ids.discard(completed_thread_id) + self._collab_turn_statuses[child_status] += 1 + return if self._turn_id is None or completed_turn_id != self._turn_id: raise ProviderError("Codex completed an unknown turn") if self._turn_completed is not None: raise ProviderError("Codex completed the same turn more than once") - encoded_size = len(_canonical_json(turn, "turn/completed.turn")) - if encoded_size > _MAX_RETAINED_TEXT_BYTES: - raise ProviderError("completed turn exceeds the retained-byte limit") + self._validate_completed_turn(turn, completed_thread_id) + children_with_turns = { + thread_id for thread_id, _turn_id in self._seen_collab_turn_ids + } + if ( + self._pending_spawn_items + or self._active_nonspawn_items + or any(self._collab_turn_ids.values()) + or self._active_collab_thread_ids + or not self._collab_thread_ids <= children_with_turns + ): + raise ProviderError( + "Codex completed root turn with outstanding child work" + ) self._turn_completed = turn return if method in _IGNORED_NOTIFICATIONS: @@ -1755,11 +2175,448 @@ def _handle_notification(self, method: str, raw_params: Any) -> None: raise ProviderError(f"prohibited Codex notification: {method}") raise ProviderError("unexpected Codex notification") - @staticmethod - def _validate_item(item: dict[str, Any]) -> None: + def _validate_item( + self, + item: dict[str, Any], + *, + owner_thread_id: str | None = None, + lifecycle: str | None = None, + ) -> None: item_type = _require_string(item.get("type"), "thread item type") if item_type not in _ALLOWED_ITEM_TYPES: raise ProviderError("prohibited or unknown Codex item type") + if item_type == "collabAgentToolCall": + self._validate_collab_item( + item, + owner_thread_id=owner_thread_id, + lifecycle=lifecycle, + ) + elif item_type == "agentMessage": + _require_exact_keys( + item, + "agent message", + required={"id", "text", "type"}, + optional={"memoryCitation", "phase"}, + ) + _require_protocol_id(item.get("id"), "agent message id") + if not isinstance(item.get("text"), str): + raise ProviderError("agent message text must be a string") + self._validate_message_phase(item.get("phase")) + + def _validate_collab_item( + self, + item: dict[str, Any], + *, + owner_thread_id: str | None, + lifecycle: str | None, + ) -> None: + _require_exact_keys( + item, + "collaboration item", + required={ + "agentsStates", + "id", + "receiverThreadIds", + "senderThreadId", + "status", + "tool", + "type", + }, + optional={"model", "prompt", "reasoningEffort"}, + ) + item_id = _require_protocol_id(item.get("id"), "collaboration item id") + sender = _require_protocol_id( + item.get("senderThreadId"), "collaboration sender thread id" + ) + if self._thread_id is None or ( + sender != self._thread_id and sender not in self._collab_thread_ids + ): + raise ProviderError("collaboration item changed sender thread scope") + if owner_thread_id is not None and sender != owner_thread_id: + raise ProviderError("collaboration item sender disagrees with its envelope") + tool = _require_string(item.get("tool"), "collaboration tool") + if tool not in {"spawnAgent", "sendInput", "resumeAgent", "wait", "closeAgent"}: + raise ProviderError("collaboration tool is unknown") + raw_receivers = _require_list( + item.get("receiverThreadIds"), + "collaboration receiver thread ids", + maximum=_MAX_COLLAB_THREADS, + ) + receivers = [ + _require_protocol_id(value, "collaboration receiver thread id") + for value in raw_receivers + ] + if len(receivers) != len(set(receivers)) or ( + self._thread_id in receivers + and (sender == self._thread_id or tool == "spawnAgent") + ): + raise ProviderError("collaboration receiver thread ids are invalid") + if any(receiver in self._failed_collab_thread_ids for receiver in receivers): + raise ProviderError("failed spawn child thread ID was reused") + status = _require_string(item.get("status"), "collaboration item status") + if status not in {"inProgress", "completed", "failed"}: + raise ProviderError("collaboration item status is unknown") + if lifecycle == "started" and status != "inProgress": + raise ProviderError("started collaboration item is not in progress") + if lifecycle in {"completed", "snapshot"} and status == "inProgress": + raise ProviderError("completed collaboration item is still in progress") + item_scope = (sender, item_id) + if lifecycle != "snapshot" and item_scope in self._terminal_collab_item_ids: + raise ProviderError("collaboration item ID was reused after termination") + model = item.get("model") + prompt = item.get("prompt") + reasoning_effort = item.get("reasoningEffort") + for field, value in ( + ("model", model), + ("prompt", prompt), + ("reasoningEffort", reasoning_effort), + ): + if value is not None and not isinstance(value, str): + raise ProviderError(f"collaboration {field} must be a string or null") + prompt_digest = ( + hashlib.sha256(prompt.encode("utf-8")).digest() + if prompt is not None + else None + ) + if tool == "sendInput" and prompt_digest is None: + raise ProviderError("sendInput prompt must be a string") + spawn_placeholder = tool == "spawnAgent" and ( + lifecycle == "started" + or ( + lifecycle in {"completed", "snapshot"} + and status == "failed" + and not receivers + ) + ) + if spawn_placeholder: + if model not in {"", self._model} or reasoning_effort not in { + "medium", + self._reasoning_effort, + }: + raise ProviderError("spawnAgent placeholder metadata is invalid") + elif tool == "spawnAgent" and lifecycle == "completed": + if model != self._model or reasoning_effort != self._reasoning_effort: + raise ProviderError("terminal spawnAgent metadata is invalid") + elif reasoning_effort == "": + raise ProviderError( + "collaboration reasoningEffort must be a non-empty string or null" + ) + elif model not in {None, self._model}: + raise ProviderError("collaboration model differs from the pinned model") + elif reasoning_effort not in {None, self._reasoning_effort}: + raise ProviderError( + "collaboration reasoningEffort differs from the pinned reasoning effort" + ) + states = _require_object(item.get("agentsStates"), "collaboration agent states") + if len(states) > _MAX_COLLAB_THREADS: + raise ProviderError("collaboration agent-state count exceeds the limit") + agent_statuses: dict[str, str] = {} + for raw_thread_id, raw_state in states.items(): + thread_id = _require_protocol_id( + raw_thread_id, "collaboration agent-state thread id" + ) + if thread_id not in receivers: + raise ProviderError("collaboration state targeted an unknown receiver") + state = _require_object(raw_state, "collaboration agent state") + _require_exact_keys( + state, + "collaboration agent state", + required={"status"}, + optional={"message"}, + ) + agent_status = _require_string( + state.get("status"), "collaboration agent status" + ) + if agent_status not in { + "pendingInit", + "running", + "interrupted", + "completed", + "errored", + "shutdown", + "notFound", + }: + raise ProviderError("collaboration agent status is unknown") + agent_statuses[thread_id] = agent_status + message = state.get("message") + if message is not None: + if ( + len( + _require_string(message, "collaboration agent message").encode( + "utf-8" + ) + ) + > _MAX_RETAINED_TEXT_BYTES + ): + raise ProviderError("collaboration agent message exceeds the limit") + if ( + tool == "wait" + and lifecycle in {"completed", "snapshot"} + and set(agent_statuses) != set(receivers) + ): + raise ProviderError("wait completion states disagree with its receivers") + terminal_item = hashlib.sha256( + _canonical_json( + { + "agent_statuses": agent_statuses, + "model": model, + "prompt_sha256": ( + prompt_digest.hex() if prompt_digest is not None else None + ), + "reasoning_effort": reasoning_effort, + "receivers": receivers, + "status": status, + "tool": tool, + }, + "collaboration terminal item", + ) + ).digest() + if ( + lifecycle == "snapshot" + and self._terminal_collab_history.get(item_scope) != terminal_item + ): + raise ProviderError( + "collaboration snapshot lacked matching live terminal history" + ) + active_nonspawn_item = self._active_nonspawn_items.get(item_scope) + if tool != "spawnAgent" and lifecycle == "started": + if active_nonspawn_item is not None: + raise ProviderError("collaboration item was already in progress") + elif tool != "spawnAgent" and lifecycle == "completed": + if active_nonspawn_item is None: + raise ProviderError("collaboration item completed without a start") + expected_tool, expected_receivers, expected_prompt = active_nonspawn_item + if expected_tool != tool: + raise ProviderError( + "terminal collaboration item disagreed with its start" + ) + if tool == "wait": + if not set(receivers).issubset(expected_receivers): + raise ProviderError( + "terminal collaboration item disagreed with its start" + ) + elif tuple(receivers) != expected_receivers: + raise ProviderError( + "terminal collaboration item disagreed with its start" + ) + if tool == "sendInput" and prompt_digest != expected_prompt: + raise ProviderError( + "terminal sendInput prompt disagreed with its start" + ) + if tool == "spawnAgent": + if lifecycle == "started" and receivers: + raise ProviderError("spawnAgent start must not have receivers") + if status == "completed" and len(receivers) != 1: + raise ProviderError( + "completed spawnAgent must have exactly one receiver" + ) + if len(receivers) > 1: + raise ProviderError("spawnAgent must have at most one receiver") + pending_key = (sender, item_id) + pending = pending_key in self._pending_spawn_items + expected_receiver, expected_prompt_digest = self._pending_spawn_items.get( + pending_key, (None, None) + ) + if status == "inProgress" and pending: + raise ProviderError("spawnAgent item was already in progress") + if ( + lifecycle == "completed" + and pending + and prompt_digest != expected_prompt_digest + ): + raise ProviderError( + "terminal spawnAgent prompt disagreed with its start" + ) + if expected_receiver is not None and receivers != [expected_receiver]: + raise ProviderError( + "spawnAgent receiver did not match its pending child thread" + ) + unbound_slots = sum( + receiver is None + for receiver, _prompt_digest in self._pending_spawn_items.values() + ) + receiver = receivers[0] if receivers else None + failed_reservation = None + if ( + status == "failed" + and receiver is None + and expected_receiver is None + and self._unbound_collab_thread_ids + ): + if unbound_slots != 1 or len(self._unbound_collab_thread_ids) != 1: + raise ProviderError( + "receiver-less spawn failure had ambiguous child scope" + ) + failed_reservation = next(iter(self._unbound_collab_thread_ids)) + if ( + lifecycle == "snapshot" + and status == "completed" + and receiver not in self._collab_thread_ids + ): + raise ProviderError( + "spawnAgent snapshot lacked matching live terminal history" + ) + if lifecycle != "snapshot" and status != "failed": + for claimed_receiver in receivers: + parent_id = self._collab_parent_ids.get(claimed_receiver) + if parent_id is not None and parent_id != sender: + raise ProviderError( + "spawnAgent receiver parent disagrees with its spawning sender" + ) + if ( + status == "failed" + and receiver is not None + and ( + receiver in self._validated_collab_thread_ids + or receiver in self._active_collab_thread_ids + or self._collab_turn_ids.get(receiver) + ) + ): + raise ProviderError("failed spawn retained an active child") + if ( + lifecycle != "snapshot" + and status != "inProgress" + and pending + and expected_receiver is None + ): + if status == "failed" and receiver is None: + pass + elif receiver in self._unbound_collab_thread_ids: + self._bind_unbound_collab_thread(receiver, sender) + elif receiver is not None and receiver in self._collab_thread_ids: + raise ProviderError("spawnAgent receiver was already claimed") + elif len(self._unbound_collab_thread_ids) >= unbound_slots: + raise ProviderError( + "spawnAgent receiver did not match a pending child thread" + ) + elif not pending and status != "inProgress" and lifecycle != "snapshot": + raise ProviderError("spawnAgent completed without a pending child") + new_receivers = set(receivers) - self._collab_thread_ids + if len(self._collab_thread_ids) + len(new_receivers) > _MAX_COLLAB_THREADS: + raise ProviderError("collaboration thread count exceeds the limit") + if lifecycle == "snapshot": + pass + elif status == "inProgress": + if any(receiver in self._collab_thread_ids for receiver in receivers): + raise ProviderError("spawnAgent receiver was already claimed") + if len(self._pending_spawn_items) >= _MAX_COLLAB_THREADS: + raise ProviderError("pending spawn count exceeds the limit") + self._pending_spawn_items[pending_key] = (receiver, prompt_digest) + else: + self._pending_spawn_items.pop(pending_key, None) + if lifecycle == "snapshot": + pass + elif status == "failed": + failed_receivers = set(receivers) + if failed_reservation is not None: + failed_receivers.add(failed_reservation) + for failed_receiver in failed_receivers: + self._failed_collab_thread_ids.add(failed_receiver) + self._unbound_collab_thread_ids.discard(failed_receiver) + self._collab_thread_ids.discard(failed_receiver) + self._active_collab_thread_ids.discard(failed_receiver) + self._collab_parent_ids.pop(failed_receiver, None) + else: + for claimed_receiver in receivers: + self._collab_parent_ids[claimed_receiver] = sender + self._collab_thread_ids.update(receivers) + if status == "completed": + self._completed_spawn_edges.add((sender, receivers[0])) + elif any( + receiver != self._thread_id and receiver not in self._collab_thread_ids + for receiver in receivers + ): + raise ProviderError("collaboration item targeted an unknown child thread") + if lifecycle != "snapshot": + for thread_id, agent_status in agent_statuses.items(): + if ( + thread_id == self._thread_id + or thread_id not in self._collab_thread_ids + ): + continue + if agent_status in {"pendingInit", "running"}: + child_has_completed_turn = any( + seen_thread_id == thread_id + for seen_thread_id, _turn_id in self._seen_collab_turn_ids + ) and not self._collab_turn_ids.get(thread_id) + if not ( + tool == "spawnAgent" + and lifecycle == "completed" + and child_has_completed_turn + ): + self._active_collab_thread_ids.add(thread_id) + else: + self._active_collab_thread_ids.discard(thread_id) + if ( + lifecycle == "completed" + and tool in {"sendInput", "resumeAgent"} + and status == "completed" + ): + self._active_collab_thread_ids.update( + receiver + for receiver in receivers + if receiver != self._thread_id + and receiver in self._collab_thread_ids + and receiver not in agent_statuses + ) + if ( + lifecycle == "completed" + and tool == "closeAgent" + and status == "completed" + ): + closed = set(receivers) + while True: + descendants = { + thread_id + for thread_id, parent_id in self._collab_parent_ids.items() + if parent_id in closed + } + if descendants <= closed: + break + closed.update(descendants) + self._active_collab_thread_ids.difference_update(closed) + if lifecycle != "snapshot" and status in {"completed", "failed"}: + if len(self._terminal_collab_item_ids) >= _MAX_MESSAGES: + raise ProviderError( + "terminal collaboration item count exceeds the limit" + ) + self._terminal_collab_item_ids.add(item_scope) + self._terminal_collab_history[item_scope] = terminal_item + if lifecycle == "started" and tool != "spawnAgent": + if ( + item_scope not in self._active_nonspawn_items + and len(self._active_nonspawn_items) >= _MAX_MESSAGES + ): + raise ProviderError("active collaboration item count exceeds the limit") + self._active_nonspawn_items[item_scope] = ( + tool, + tuple(receivers), + prompt_digest, + ) + elif lifecycle == "completed" and tool != "spawnAgent": + self._active_nonspawn_items.pop(item_scope, None) + + def _validate_completed_turn( + self, turn: dict[str, Any], owner_thread_id: str + ) -> str: + status = _require_string(turn.get("status"), "completed turn status") + if status not in {"completed", "interrupted", "failed"}: + raise ProviderError("completed turn has a non-terminal status") + items = _require_list( + turn.get("items"), "completed turn items", maximum=_MAX_MESSAGES + ) + items_view = _require_string(turn.get("itemsView"), "completed turn itemsView") + if items_view not in {"full", "notLoaded", "summary"}: + raise ProviderError("completed turn returned an unsupported item view") + if items_view == "notLoaded" and items: + raise ProviderError("Codex not-loaded turn items must be empty") + for index, raw_item in enumerate(items): + self._validate_item( + _require_object(raw_item, f"turn.items[{index}]"), + owner_thread_id=owner_thread_id, + lifecycle="snapshot", + ) + return status @staticmethod def _validate_message_phase(value: Any) -> str | None: @@ -1779,27 +2636,74 @@ def _matches_turn(self, params: dict[str, Any]) -> bool: and turn_id == self._turn_id ) + def _claim_collab_thread_scope(self, thread_id: str) -> bool: + if thread_id in self._failed_collab_thread_ids: + raise ProviderError("failed spawn child thread ID was reused") + if thread_id in self._collab_thread_ids: + return True + open_spawn_slots = sum( + receiver is None + for receiver, _prompt_digest in self._pending_spawn_items.values() + ) + if len(self._unbound_collab_thread_ids) >= open_spawn_slots: + return False + if len(self._collab_thread_ids) >= _MAX_COLLAB_THREADS: + raise ProviderError("collaboration thread count exceeds the limit") + self._unbound_collab_thread_ids.add(thread_id) + self._collab_thread_ids.add(thread_id) + return True + + def _bind_unbound_collab_thread(self, thread_id: str, sender: str) -> None: + parent_id = self._collab_parent_ids.get(thread_id) + if parent_id is not None and parent_id != sender: + raise ProviderError( + "spawnAgent receiver parent disagrees with its spawning sender" + ) + self._collab_parent_ids[thread_id] = sender + self._unbound_collab_thread_ids.remove(thread_id) + + def _matches_collab_turn(self, params: dict[str, Any]) -> bool: + thread_id = _require_protocol_id( + params.get("threadId"), "collaboration notification.threadId" + ) + turn_id = _require_protocol_id( + params.get("turnId"), "collaboration notification.turnId" + ) + return turn_id in self._collab_turn_ids.get(thread_id, set()) + def _validate_ignored_notification_scope( self, method: str, params: dict[str, Any] ) -> None: - if method.startswith("item/"): - if not self._matches_turn(params): - raise ProviderError(f"{method} omitted or changed turn scope") - return - if method.startswith("turn/"): - if not self._matches_turn(params): + if method.startswith(("item/", "turn/", "model/")): + if not self._matches_turn(params) and not self._matches_collab_turn(params): raise ProviderError(f"{method} omitted or changed turn scope") return if method == "thread/status/changed": + _require_exact_keys( + params, + "thread/status/changed", + required={"status", "threadId"}, + ) thread_id = _require_protocol_id( params.get("threadId"), "thread/status/changed.threadId" ) - if self._thread_id is None or thread_id != self._thread_id: + status_type = self._validate_thread_status( + _require_object(params.get("status"), "thread/status/changed.status"), + "thread/status/changed.status", + ) + if self._thread_id is None: raise ProviderError("thread/status/changed changed thread scope") - return - if method.startswith("model/"): - if not self._matches_turn(params): - raise ProviderError(f"{method} omitted or changed turn scope") + if ( + thread_id != self._thread_id + and thread_id not in self._collab_thread_ids + ): + if not self._claim_collab_thread_scope(thread_id): + raise ProviderError("thread/status/changed changed thread scope") + if thread_id != self._thread_id: + if status_type == "active": + self._active_collab_thread_ids.add(thread_id) + else: + self._active_collab_thread_ids.discard(thread_id) return thread_id = params.get("threadId") turn_id = params.get("turnId") @@ -1810,6 +2714,32 @@ def _validate_ignored_notification_scope( if _require_protocol_id(turn_id, f"{method}.turnId") != self._turn_id: raise ProviderError(f"{method} targeted an unknown turn") + @staticmethod + def _validate_thread_status(status: dict[str, Any], label: str) -> str: + status_type = _require_string(status.get("type"), f"{label}.type") + if status_type == "active": + _require_exact_keys( + status, + label, + required={"activeFlags", "type"}, + ) + flags = _require_list( + status.get("activeFlags"), + f"{label}.activeFlags", + maximum=2, + ) + if any( + _require_string(flag, "thread active flag") + not in {"waitingOnApproval", "waitingOnUserInput"} + for flag in flags + ): + raise ProviderError("thread status has an unknown active flag") + elif status_type in {"notLoaded", "idle", "systemError"}: + _require_exact_keys(status, label, required={"type"}) + else: + raise ProviderError("thread status is unknown") + return status_type + @staticmethod def _validate_usage(raw: dict[str, Any]) -> dict[str, int]: mapping = { @@ -1841,17 +2771,21 @@ def _finalize_turn(self) -> tuple[str, dict[str, Any]]: if turn.get("error") is not None: raise ProviderError("completed Codex turn included an error") items = _require_list(turn.get("items"), "turn.items", maximum=_MAX_MESSAGES) - items_view = turn.get("itemsView", "full") + items_view = _require_string(turn.get("itemsView"), "turn.itemsView") messages: list[dict[str, str | None]] if items_view == "notLoaded": if items: raise ProviderError("Codex not-loaded turn items must be empty") messages = list(self._completed_messages) - elif items_view == "full": + elif items_view in {"full", "summary"}: messages = [] for index, raw_item in enumerate(items): item = _require_object(raw_item, f"turn.items[{index}]") - self._validate_item(item) + self._validate_item( + item, + owner_thread_id=self._thread_id, + lifecycle="snapshot", + ) if item.get("type") == "agentMessage": messages.append( { @@ -1864,11 +2798,42 @@ def _finalize_turn(self) -> tuple[str, dict[str, Any]]: ), } ) + if items_view == "summary": + if len(items) != 1 or len(messages) != 1: + raise ProviderError( + "Codex summary turn must contain one agent message" + ) + summary_index = next( + ( + index + for index in range(len(self._completed_messages) - 1, -1, -1) + if str(self._completed_messages[index]["text"]).strip() + ), + None, + ) + if summary_index is None: + summary_index = len(self._completed_messages) - 1 + if ( + summary_index < 0 + or messages[0] != self._completed_messages[summary_index] + ): + raise ProviderError( + "Codex summary message disagrees with its completion event" + ) + if any( + message["phase"] == "final_answer" + for message in self._completed_messages[summary_index + 1 :] + ): + raise ProviderError( + "Codex summary omitted a completed final-answer message" + ) + messages = self._completed_messages[: summary_index + 1] else: raise ProviderError("Codex turn returned an unsupported item view") if not messages or not self._completed_messages: raise ProviderError("Codex turn omitted a completed final agent message") - message_ids = [message["id"] for message in messages] + id_source = self._completed_messages if items_view == "summary" else messages + message_ids = [message["id"] for message in id_source] if len(set(message_ids)) != len(message_ids): raise ProviderError("Codex turn repeated an agent-message id") if items_view == "full" and messages != self._completed_messages: @@ -2097,7 +3062,7 @@ def send(self, payload: bytes, deadline: float) -> None: while view: events = selector.select(_remaining(deadline, "Codex protocol write")) if not events: - raise ProviderError("Codex protocol write timed out") + raise ProviderTimeoutError("Codex protocol write timed out") try: written = os.write(self._stdin_fd, view) except BlockingIOError: @@ -2127,7 +3092,7 @@ def receive(self, deadline: float) -> bytes: raise ProviderError("Codex protocol frame exceeds the byte limit") events = selector.select(_remaining(deadline, "Codex protocol read")) if not events: - raise ProviderError("Codex protocol read timed out") + raise ProviderTimeoutError("Codex protocol read timed out") try: chunk = os.read(self._stdout_fd, 64 * 1024) except BlockingIOError: @@ -2434,6 +3399,7 @@ def _owner_lock( except OSError as exc: raise ProviderError(f"cannot open {label}: {exc}") from exc body_error: BaseException | None = None + serialization_timeout: ProviderTimeoutError | None = None try: identity = _validate_owner_file_descriptor(lock_path, descriptor, label) while True: @@ -2441,27 +3407,36 @@ def _owner_lock( fcntl.flock(descriptor, fcntl.LOCK_EX | fcntl.LOCK_NB) break except BlockingIOError: - _remaining(deadline, f"{label} serialization") + try: + _remaining(deadline, f"{label} serialization") + except ProviderTimeoutError as exc: + serialization_timeout = exc + break time.sleep(min(0.05, max(0.0, deadline - time.monotonic()))) - identity = _validate_owner_file_descriptor(lock_path, descriptor, label) - try: - yield identity - except BaseException as exc: - body_error = exc - raise - finally: + if serialization_timeout is None: + identity = _validate_owner_file_descriptor(lock_path, descriptor, label) try: - _validate_owner_file_descriptor(lock_path, descriptor, label) + yield identity except BaseException as exc: - if body_error is not None: - body_error.add_note(f"{label} integrity also failed: {exc}") - else: - raise + body_error = exc + raise + finally: + try: + _validate_owner_file_descriptor(lock_path, descriptor, label) + except BaseException as exc: + if body_error is not None: + body_error.add_note(f"{label} integrity also failed: {exc}") + else: + raise finally: try: fcntl.flock(descriptor, fcntl.LOCK_UN) finally: os.close(descriptor) + if serialization_timeout is not None: + raise ProviderTimeoutError( + str(serialization_timeout), cleanup_confirmed=True + ) from serialization_timeout @contextmanager @@ -3137,13 +4112,15 @@ def _remove_invocation_root(path: Path) -> None: raise ProviderError(f"cannot remove Codex invocation root: {exc}") from exc -def _prepare_workspace_runtime(workspace: Path) -> tuple[Path, ...]: +def _prepare_workspace_runtime(workspace: Path) -> _WorkspaceRuntime: paths = ( workspace / ".skill-eval-tmp", workspace / ".skill-eval-cache", workspace / ".skill-eval-home", ) created: list[Path] = [] + parent_descriptor: int | None = None + directories: list[tuple[str, int]] = [] try: for path in paths: if path.exists() or path.is_symlink(): @@ -3152,32 +4129,182 @@ def _prepare_workspace_runtime(workspace: Path) -> tuple[Path, ...]: ) path.mkdir(mode=_PRIVATE_DIRECTORY_MODE) created.append(path) + flags = os.O_RDONLY | os.O_CLOEXEC | getattr(os, "O_DIRECTORY", 0) + flags |= getattr(os, "O_NOFOLLOW", 0) + parent_descriptor = os.open(workspace, flags) + for path in paths: + descriptor = os.open(path.name, flags, dir_fd=parent_descriptor) + os.fchmod(descriptor, _PRIVATE_DIRECTORY_MODE) + directories.append((path.name, descriptor)) except BaseException: + for _name, descriptor in reversed(directories): + os.close(descriptor) + if parent_descriptor is not None: + os.close(parent_descriptor) for path in reversed(created): shutil.rmtree(path, ignore_errors=True) raise - return paths + assert parent_descriptor is not None + return _WorkspaceRuntime(parent_descriptor, tuple(directories)) -def _cleanup_workspace_runtime(paths: tuple[Path, ...]) -> None: - for path in paths: - try: - metadata = path.lstat() - except FileNotFoundError: - continue - except OSError as exc: - raise ProviderError( - f"cannot inspect workspace runtime path: {exc}" - ) from exc - try: - if stat.S_ISLNK(metadata.st_mode) or stat.S_ISREG(metadata.st_mode): - path.unlink() - elif stat.S_ISDIR(metadata.st_mode): - shutil.rmtree(path) - else: - raise ProviderError("workspace runtime path changed to an unsafe type") - except OSError as exc: - raise ProviderError(f"cannot clean workspace runtime path: {exc}") from exc +def _open_workspace_runtime_directory( + parent_descriptor: int, + name: str, + expected_identity: tuple[int, int], +) -> int: + path_flag = getattr(os, "O_PATH", None) + if path_flag is None: + raise ProviderError("workspace runtime cleanup requires Linux O_PATH") + path_descriptor = os.open( + name, + path_flag | os.O_CLOEXEC | os.O_DIRECTORY | os.O_NOFOLLOW, + dir_fd=parent_descriptor, + ) + try: + metadata = os.fstat(path_descriptor) + if ( + not stat.S_ISDIR(metadata.st_mode) + or ( + metadata.st_dev, + metadata.st_ino, + ) + != expected_identity + ): + raise ProviderError("workspace runtime entry changed to an unsafe type") + os.chmod(f"/proc/self/fd/{path_descriptor}", _PRIVATE_DIRECTORY_MODE) + descriptor = os.open( + ".", + os.O_RDONLY | os.O_CLOEXEC | os.O_DIRECTORY | os.O_NOFOLLOW, + dir_fd=path_descriptor, + ) + opened = os.fstat(descriptor) + if (opened.st_dev, opened.st_ino) != (metadata.st_dev, metadata.st_ino): + os.close(descriptor) + raise ProviderError("workspace runtime directory identity changed") + return descriptor + finally: + os.close(path_descriptor) + + +def _clear_workspace_runtime_directory(descriptor: int) -> None: + os.fchmod(descriptor, _PRIVATE_DIRECTORY_MODE) + root = os.fstat(descriptor) + stack: list[tuple[list[str], int, str | None, tuple[int, int]]] = [ + (os.listdir(descriptor), 0, None, (root.st_dev, root.st_ino)) + ] + current_descriptor = descriptor + flags = os.O_RDONLY | os.O_CLOEXEC | os.O_DIRECTORY | os.O_NOFOLLOW + try: + while stack: + names, index, name, identity = stack[-1] + if index < len(names): + child_name = names[index] + stack[-1] = (names, index + 1, name, identity) + metadata = os.stat( + child_name, + dir_fd=current_descriptor, + follow_symlinks=False, + ) + if not stat.S_ISDIR(metadata.st_mode): + os.unlink(child_name, dir_fd=current_descriptor) + continue + child_descriptor = _open_workspace_runtime_directory( + current_descriptor, + child_name, + (metadata.st_dev, metadata.st_ino), + ) + try: + child_identity = os.fstat(child_descriptor) + os.fchmod(child_descriptor, _PRIVATE_DIRECTORY_MODE) + child_names = os.listdir(child_descriptor) + except BaseException: + os.close(child_descriptor) + raise + if current_descriptor != descriptor: + os.close(current_descriptor) + current_descriptor = child_descriptor + stack.append( + ( + child_names, + 0, + child_name, + (child_identity.st_dev, child_identity.st_ino), + ) + ) + continue + + if len(stack) == 1: + break + stack.pop() + parent_identity = stack[-1][3] + parent_descriptor = ( + descriptor + if len(stack) == 1 + else os.open("..", flags, dir_fd=current_descriptor) + ) + try: + observed_parent = os.fstat(parent_descriptor) + if (observed_parent.st_dev, observed_parent.st_ino) != parent_identity: + raise ProviderError( + "workspace runtime parent directory identity changed" + ) + assert name is not None + current = os.stat( + name, + dir_fd=parent_descriptor, + follow_symlinks=False, + ) + if (current.st_dev, current.st_ino) != identity: + raise ProviderError("workspace runtime directory identity changed") + os.rmdir(name, dir_fd=parent_descriptor) + if os.fstat(current_descriptor).st_nlink != 0: + raise ProviderError( + "workspace runtime directory removal was replaced" + ) + except BaseException: + if parent_descriptor != descriptor: + os.close(parent_descriptor) + raise + os.close(current_descriptor) + current_descriptor = parent_descriptor + except BaseException: + if current_descriptor != descriptor: + os.close(current_descriptor) + raise + + +def _cleanup_workspace_runtime(runtime: _WorkspaceRuntime) -> None: + try: + for name, descriptor in runtime.directories: + identity = os.fstat(descriptor) + _clear_workspace_runtime_directory(descriptor) + try: + current = os.stat( + name, + dir_fd=runtime.parent_descriptor, + follow_symlinks=False, + ) + except FileNotFoundError: + if identity.st_nlink != 0: + raise ProviderError( + "workspace runtime directory was renamed" + ) from None + continue + if not stat.S_ISDIR(current.st_mode) or ( + current.st_dev, + current.st_ino, + ) != (identity.st_dev, identity.st_ino): + raise ProviderError("workspace runtime directory identity changed") + os.rmdir(name, dir_fd=runtime.parent_descriptor) + if os.fstat(descriptor).st_nlink != 0: + raise ProviderError("workspace runtime directory removal was replaced") + except OSError as exc: + raise ProviderError(f"cannot clean workspace runtime path: {exc}") from exc + finally: + for _name, descriptor in runtime.directories: + os.close(descriptor) + os.close(runtime.parent_descriptor) def _fsync_private_directory(path: Path) -> None: @@ -3521,6 +4648,7 @@ def run_agent(self, request: AgentRequest) -> ProviderResult: started = time.monotonic() transport: _Transport | None = None outcome: _TurnOutcome | None = None + timeout_error: ProviderTimeoutError | None = None sandbox: dict[str, Any] | None = None expected_command_sha256: str | None = None coordination_root = _validate_private_directory( @@ -3551,7 +4679,7 @@ def run_agent(self, request: AgentRequest) -> ProviderResult: poison_recovered = poison_store.recover(poison_binding) _prepare_runtime_mountpoint(mounted_root) invocation_root = _new_invocation_root(self._runtime_root) - workspace_runtime: tuple[Path, ...] | None = None + workspace_runtime: _WorkspaceRuntime | None = None tool_attestations: list[VerifiedExecutable] = [] try: try: @@ -3618,17 +4746,20 @@ def run_agent(self, request: AgentRequest) -> ProviderResult: raise ProviderError( "Codex system context retained the host skill path" ) - outcome = _AppServerProtocol( - session, - model=self._config.model, - reasoning_effort=self._config.reasoning_effort, - workspace=mounted_paths["work"], - system_context=system_context, - locked_efforts=self._lock.model_efforts[self._config.model], - locked_thread_cli_version=self._lock.thread_cli_version, - expected_codex_home=mounted_paths["codex-home"], - on_dispatched=request.on_dispatched, - ).run(request.prompt, deadline) + try: + outcome = _AppServerProtocol( + session, + model=self._config.model, + reasoning_effort=self._config.reasoning_effort, + workspace=mounted_paths["work"], + system_context=system_context, + locked_efforts=self._lock.model_efforts[self._config.model], + locked_thread_cli_version=self._lock.thread_cli_version, + expected_codex_home=mounted_paths["codex-home"], + on_dispatched=request.on_dispatched, + ).run(request.prompt, deadline) + except ProviderTimeoutError as exc: + timeout_error = exc finally: session.close() if not self._transport_is_injected: @@ -3685,6 +4816,10 @@ def run_agent(self, request: AgentRequest) -> ProviderResult: ) finally: _remove_invocation_root(invocation_root) + if timeout_error is not None: + raise ProviderTimeoutError( + str(timeout_error), cleanup_confirmed=True + ) from timeout_error assert outcome is not None and sandbox is not None return self._build_result( request, diff --git a/tests/test_codex_app_server.py b/tests/test_codex_app_server.py index 7514804..1b8cf7a 100644 --- a/tests/test_codex_app_server.py +++ b/tests/test_codex_app_server.py @@ -39,19 +39,25 @@ _CleanupPoisonStore, _JsonRpcSession, _LAUNCH_GATE_SCRIPT, + _MAX_COLLAB_THREADS, _PoisonBinding, _ProcessTransport, _SystemdRecoveryProbe, _auth_lock, + _cleanup_workspace_runtime, _command_sha256, _load_protocol_lock, _linux_process_start_time, + _prepare_workspace_runtime, + _remaining, + _remove_invocation_root, _static_config, _resolve_gate_shell, _resolve_system_tool, validate_codex_protocol_lock, _validate_cli_version_output, CodexAppServerProvider, + ProviderTimeoutError, ) from skivolve.manifest import ProviderConfig # noqa: E402 from skivolve.providers import AgentRequest, ProviderError # noqa: E402 @@ -152,6 +158,40 @@ def _model(model: str, efforts: tuple[str, ...]) -> dict[str, Any]: } +def _child_thread( + thread_id: str = "child-1", + *, + depth: int = 1, + parent_id: str = "thread-1", + session_id: str = "session-1", +) -> dict[str, Any]: + return { + "cliVersion": "0.146.0", + "createdAt": 1_700_000_001, + "cwd": "/runtime/work", + "ephemeral": True, + "historyMode": "paginated", + "id": thread_id, + "modelProvider": "openai", + "parentThreadId": parent_id, + "path": None, + "preview": "delegated task", + "sessionId": session_id, + "source": { + "subAgent": { + "thread_spawn": { + "depth": depth, + "parent_thread_id": parent_id, + } + } + }, + "status": {"activeFlags": [], "type": "active"}, + "threadSource": "skill-eval", + "turns": [], + "updatedAt": 1_700_000_001, + } + + def _rate_limits(used: int, limit_id: str = "codex") -> dict[str, Any]: snapshot = { "limitId": limit_id, @@ -249,10 +289,11 @@ def __init__( identifier_sentinel: str | None = None, omit_item_completed: bool = False, not_loaded_has_items: bool = False, + error_will_retry: bool | None = None, skill_disable_succeeds: bool = True, - thread_cli_version: str = "0.144.3", + thread_cli_version: str = "0.146.0", turn_error: dict[str, Any] | None = None, - turn_items_view: str | None = "notLoaded", + turn_items_view: str | None = "summary", ) -> None: self.workspace = workspace self.additional_item_completed = additional_item_completed @@ -265,6 +306,7 @@ def __init__( self.full_has_non_object_item = full_has_non_object_item self.omit_item_completed = omit_item_completed self.not_loaded_has_items = not_loaded_has_items + self.error_will_retry = error_will_retry self.thread_id = ( f"{identifier_sentinel}.thread" if identifier_sentinel else "thread-1" ) @@ -429,6 +471,28 @@ def send(self, payload: bytes, _deadline: float) -> None: self._queue_turn_events() def _queue_turn_events(self) -> None: + if self.error_will_retry is not None: + self.incoming.append( + _line( + { + "method": "error", + "params": { + "error": { + "additionalDetails": None, + "codexErrorInfo": { + "responseStreamDisconnected": { + "httpStatusCode": None + } + }, + "message": "SENTINEL_MUST_NOT_BE_DISCLOSED", + }, + "threadId": self.thread_id, + "turnId": self.turn_id, + "willRetry": self.error_will_retry, + }, + } + ) + ) if self.reroute: self.incoming.append( _line( @@ -535,11 +599,27 @@ def _queue_turn_events(self) -> None: full_items.append(self.additional_item_completed) if self.full_has_non_object_item: full_items.append(None) - turn_items = ( - full_items - if self.turn_items_view in {None, "full"} or self.not_loaded_has_items - else [] - ) + if self.turn_items_view == "summary": + turn_items = [ + next( + ( + item + for item in reversed(full_items) + if isinstance(item, dict) + and item.get("type") == "agentMessage" + and str(item.get("text", "")).strip() + ), + next( + item + for item in reversed(full_items) + if isinstance(item, dict) and item.get("type") == "agentMessage" + ), + ) + ] + elif self.turn_items_view == "full" or self.not_loaded_has_items: + turn_items = full_items + else: + turn_items = [] turn = { "durationMs": 25, "id": self.turn_id, @@ -600,13 +680,25 @@ def _protocol( workspace=Path("/runtime/work"), system_context="isolated evaluation", locked_efforts=("low", "medium", "high", "xhigh", "max"), - locked_thread_cli_version="0.144.3", + locked_thread_cli_version="0.146.0", expected_codex_home=Path("/runtime/codex-home"), on_dispatched=on_dispatched, ) class CodexProtocolTests(unittest.TestCase): + def test_expired_deadline_raises_typed_timeout(self) -> None: + with self.assertRaisesRegex( + ProviderTimeoutError, "operation timed out" + ) as caught: + _remaining(time.monotonic() - 1, "operation") + self.assertFalse(caught.exception.cleanup_confirmed) + confirmed = ProviderTimeoutError("operation timed out", cleanup_confirmed=True) + self.assertTrue(confirmed.cleanup_confirmed) + self.assertIsInstance(confirmed, ProviderError) + with self.assertRaisesRegex(TypeError, "cleanup_confirmed"): + ProviderTimeoutError("operation timed out", cleanup_confirmed=1) # type: ignore[arg-type] + def test_happy_path_paginates_disables_skills_merges_quota_and_uses_last_usage( self, ) -> None: @@ -619,14 +711,14 @@ def test_happy_path_paginates_disables_skills_merges_quota_and_uses_last_usage( workspace=Path("/runtime/work"), system_context="isolated evaluation", locked_efforts=("low", "medium", "high", "xhigh", "max"), - locked_thread_cli_version="0.144.3", + locked_thread_cli_version="0.146.0", expected_codex_home=Path("/runtime/codex-home"), ).run("implement request", time.monotonic() + 5) self.assertEqual(outcome.final_output, "completed fixture") self.assertEqual(outcome.tokens["input_tokens"], 10) self.assertEqual(outcome.tokens["total_tokens"], 15) - self.assertEqual(outcome.raw_response["turn"]["items_view"], "notLoaded") + self.assertEqual(outcome.raw_response["turn"]["items_view"], "summary") rolling = outcome.quota["rolling"]["rateLimits"] self.assertEqual(rolling["planType"], "pro") self.assertEqual(rolling["primary"]["usedPercent"], 40) @@ -652,6 +744,115 @@ def test_happy_path_paginates_disables_skills_merges_quota_and_uses_last_usage( self.assertEqual(turn["params"]["model"], "gpt-5.6-luna") self.assertNotIn("environments", turn["params"]) + def test_child_usage_retains_only_the_latest_turn_update(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + protocol._collab_turn_ids["child-1"] = {"child-turn"} + for input_tokens, output_tokens, total_tokens in ((3, 1, 4), (20, 8, 28)): + protocol._handle_notification( + "thread/tokenUsage/updated", + { + "threadId": "child-1", + "tokenUsage": { + "last": { + "cachedInputTokens": 4, + "inputTokens": input_tokens, + "outputTokens": output_tokens, + "reasoningOutputTokens": 3, + "totalTokens": total_tokens, + } + }, + "turnId": "child-turn", + }, + ) + + self.assertEqual( + protocol._collab_turn_usage[("child-1", "child-turn")]["total_tokens"], + 28, + ) + + def test_happy_path_adds_child_usage_to_reported_totals(self) -> None: + protocol = _protocol(ScriptedTransport(Path("/runtime/work"))) + protocol._validated_collab_thread_ids.add("child-1") + protocol._seen_collab_turn_ids.add(("child-1", "child-turn")) + protocol._collab_turn_statuses["failed"] = 1 + protocol._collab_turn_usage[("child-1", "child-turn")] = { + "cached_input_tokens": 4, + "input_tokens": 20, + "output_tokens": 8, + "reasoning_output_tokens": 3, + "total_tokens": 28, + } + outcome = protocol.run("request", time.monotonic() + 5) + + self.assertEqual( + outcome.tokens, + { + "cached_input_tokens": 6, + "input_tokens": 30, + "output_tokens": 13, + "reasoning_output_tokens": 4, + "total_tokens": 43, + }, + ) + self.assertEqual(outcome.raw_response["usage"], outcome.tokens) + self.assertEqual( + outcome.raw_response["collaboration"], + { + "child_thread_count": 1, + "child_turn_count": 1, + "child_turn_statuses": { + "completed": 0, + "failed": 1, + "interrupted": 0, + }, + "usage": { + "cached_input_tokens": 4, + "input_tokens": 20, + "output_tokens": 8, + "reasoning_output_tokens": 3, + "total_tokens": 28, + }, + }, + ) + + def test_post_completion_quota_read_rejects_new_collaboration_work(self) -> None: + class PostCompletionTransport(ScriptedTransport): + def send(self, payload: bytes, deadline: float) -> None: + message = json.loads(payload) + if ( + message.get("method") == "account/rateLimits/read" + and self.rate_reads == 1 + ): + self.incoming.append( + _line( + { + "method": "item/started", + "params": { + "item": { + "agentsStates": {}, + "id": "late-spawn", + "receiverThreadIds": [], + "senderThreadId": self.thread_id, + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + }, + "startedAtMs": 1, + "threadId": self.thread_id, + "turnId": self.turn_id, + }, + } + ) + ) + super().send(payload, deadline) + + with self.assertRaisesRegex(ProviderError, "after root completion"): + _protocol(PostCompletionTransport(Path("/runtime/work"))).run( + "request", time.monotonic() + 5 + ) + def test_thread_cli_version_mismatch_fails_before_turn_start(self) -> None: for version in ("codex-cli 0.144.1", "0.144.2"): with self.subTest(version=version): @@ -677,10 +878,355 @@ def test_reroute_fails_closed(self) -> None: workspace=Path("/runtime/work"), system_context="test", locked_efforts=("low", "medium", "high", "xhigh", "max"), - locked_thread_cli_version="0.144.3", + locked_thread_cli_version="0.146.0", expected_codex_home=Path("/runtime/codex-home"), ).run("request", time.monotonic() + 5) + def test_retryable_error_notification_does_not_interrupt_turn(self) -> None: + transport = ScriptedTransport(Path("/runtime/work"), error_will_retry=True) + + outcome = _protocol(transport).run("request", time.monotonic() + 5) + + self.assertEqual(outcome.final_output, "completed fixture") + + def test_non_retryable_error_notification_fails_closed(self) -> None: + transport = ScriptedTransport(Path("/runtime/work"), error_will_retry=False) + + with self.assertRaises(ProviderError) as caught: + _protocol(transport).run("request", time.monotonic() + 5) + self.assertEqual(str(caught.exception), "Codex reported a turn error") + self.assertNotIn("SENTINEL_MUST_NOT_BE_DISCLOSED", str(caught.exception)) + + def test_error_notification_requires_boolean_retry_state(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + for retry_state in (None, 0, 1, "true", {}): + with self.subTest(retry_state=retry_state): + with self.assertRaisesRegex(ProviderError, "invalid retry state"): + protocol._handle_notification( + "error", + { + "error": { + "additionalDetails": None, + "codexErrorInfo": None, + "message": "untrusted", + }, + "threadId": "thread-1", + "turnId": "turn-1", + "willRetry": retry_state, + }, + ) + + def test_retryable_error_notification_validates_shape_and_scope(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + valid = { + "error": {"message": "untrusted"}, + "threadId": "thread-1", + "turnId": "turn-1", + "willRetry": True, + } + invalid = { + "missing message": { + **valid, + "error": {"codexErrorInfo": None}, + }, + "invalid message": { + **valid, + "error": {**valid["error"], "message": None}, + }, + "invalid additional details": { + **valid, + "error": {**valid["error"], "additionalDetails": {}}, + }, + "unknown error code": { + **valid, + "error": {**valid["error"], "codexErrorInfo": "futureError"}, + }, + "unknown error variant": { + **valid, + "error": {**valid["error"], "codexErrorInfo": {"futureError": {}}}, + }, + "invalid HTTP status": { + **valid, + "error": { + **valid["error"], + "codexErrorInfo": { + "responseStreamDisconnected": {"httpStatusCode": 65536} + }, + }, + }, + "invalid active turn kind": { + **valid, + "error": { + **valid["error"], + "codexErrorInfo": {"activeTurnNotSteerable": {"turnKind": "exec"}}, + }, + }, + "wrong thread": {**valid, "threadId": "thread-2"}, + "wrong turn": {**valid, "turnId": "turn-2"}, + } + + for error in ( + valid["error"], + {**valid["error"], "additionalDetails": "upstream detail"}, + {**valid["error"], "codexErrorInfo": "serverOverloaded"}, + { + **valid["error"], + "codexErrorInfo": { + "responseStreamDisconnected": {"httpStatusCode": 503} + }, + }, + { + **valid["error"], + "codexErrorInfo": {"activeTurnNotSteerable": {"turnKind": "review"}}, + }, + ): + protocol._handle_notification("error", {**valid, "error": error}) + for label, params in invalid.items(): + with self.subTest(label=label): + with self.assertRaises(ProviderError): + protocol._handle_notification("error", params) + + def test_non_retryable_child_error_does_not_abort_main_turn(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + protocol._collab_thread_ids.add("child-1") + protocol._collab_turn_ids["child-1"] = {"child-turn"} + protocol._active_collab_thread_ids.add("child-1") + protocol._collab_turn_usage[("child-1", "child-turn")] = { + "cached_input_tokens": 0, + "input_tokens": 1, + "output_tokens": 0, + "reasoning_output_tokens": 0, + "total_tokens": 1, + } + + protocol._handle_notification( + "error", + { + "error": { + "additionalDetails": None, + "codexErrorInfo": None, + "message": "untrusted child failure", + }, + "threadId": "child-1", + "turnId": "child-turn", + "willRetry": False, + }, + ) + protocol._handle_notification( + "turn/completed", + { + "threadId": "child-1", + "turn": { + "id": "child-turn", + "items": [], + "itemsView": "notLoaded", + "status": "failed", + }, + }, + ) + + self.assertIsNone(protocol._turn_completed) + self.assertEqual(protocol._collab_turn_ids["child-1"], set()) + self.assertEqual(protocol._active_collab_thread_ids, set()) + self.assertEqual(protocol._collab_turn_statuses["failed"], 1) + + def test_child_completion_without_usage_keeps_turn_active(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + protocol._collab_turn_ids["child-1"] = {"child-turn"} + + with self.assertRaisesRegex(ProviderError, "child turn omitted token usage"): + protocol._handle_notification( + "turn/completed", + { + "threadId": "child-1", + "turn": { + "id": "child-turn", + "items": [], + "itemsView": "notLoaded", + "status": "completed", + }, + }, + ) + + self.assertEqual(protocol._collab_turn_ids["child-1"], {"child-turn"}) + + def test_root_completion_requires_terminal_child_work(self) -> None: + completed = { + "threadId": "main-thread", + "turn": { + "id": "main-turn", + "items": [], + "itemsView": "notLoaded", + "status": "completed", + }, + } + for state in ( + "pending-spawn", + "bound-thread", + "awaiting-turn", + "active-turn", + "resumed-thread", + "resumed-item", + "started-resume-item", + "completed-resume-without-state", + ): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + if state == "pending-spawn": + protocol._pending_spawn_items[("main-thread", "spawn-1")] = ( + None, + None, + ) + elif state == "bound-thread": + protocol._collab_thread_ids.add("child-1") + elif state == "awaiting-turn": + protocol._collab_thread_ids.add("child-1") + protocol._validated_collab_thread_ids.add("child-1") + elif state == "active-turn": + protocol._collab_thread_ids.add("child-1") + protocol._collab_turn_ids["child-1"] = {"child-turn"} + elif state == "resumed-thread": + protocol._collab_thread_ids.add("child-1") + protocol._seen_collab_turn_ids.add(("child-1", "prior-turn")) + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": "child-1", + }, + ) + elif state == "resumed-item": + protocol._collab_thread_ids.add("child-1") + protocol._seen_collab_turn_ids.add(("child-1", "prior-turn")) + protocol._validate_item( + { + "agentsStates": {"child-1": {"status": "running"}}, + "id": "resume-1", + "receiverThreadIds": ["child-1"], + "senderThreadId": "main-thread", + "status": "completed", + "tool": "resumeAgent", + "type": "collabAgentToolCall", + } + ) + elif state in {"started-resume-item", "completed-resume-without-state"}: + protocol._collab_thread_ids.add("child-1") + protocol._seen_collab_turn_ids.add(("child-1", "prior-turn")) + resume = { + "agentsStates": {}, + "id": "resume-1", + "receiverThreadIds": ["child-1"], + "senderThreadId": "main-thread", + "status": "inProgress", + "tool": "resumeAgent", + "type": "collabAgentToolCall", + } + protocol._handle_notification( + "item/started", + { + "item": resume, + "startedAtMs": 1, + "threadId": "main-thread", + "turnId": "main-turn", + }, + ) + if state == "completed-resume-without-state": + protocol._handle_notification( + "item/completed", + { + "item": {**resume, "status": "completed"}, + "completedAtMs": 2, + "threadId": "main-thread", + "turnId": "main-turn", + }, + ) + + with ( + self.subTest(state=state), + self.assertRaisesRegex(ProviderError, "outstanding child work"), + ): + protocol._handle_notification("turn/completed", completed) + self.assertIsNone(protocol._turn_completed) + + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + protocol._collab_thread_ids.add("child-1") + protocol._seen_collab_turn_ids.add(("child-1", "prior-turn")) + resume = { + "agentsStates": {}, + "id": "resume-1", + "receiverThreadIds": ["child-1"], + "senderThreadId": "main-thread", + "status": "inProgress", + "tool": "resumeAgent", + "type": "collabAgentToolCall", + } + envelope = {"threadId": "main-thread", "turnId": "main-turn"} + completed_resume = { + **resume, + "agentsStates": {"child-1": {"status": "running"}}, + "status": "completed", + } + with self.assertRaisesRegex(ProviderError, "live terminal history"): + protocol._validate_item(completed_resume, lifecycle="snapshot") + with self.assertRaisesRegex(ProviderError, "completed without a start"): + protocol._handle_notification( + "item/completed", + {**envelope, "completedAtMs": 2, "item": completed_resume}, + ) + protocol._handle_notification( + "item/started", {**envelope, "item": resume, "startedAtMs": 1} + ) + with self.assertRaisesRegex(ProviderError, "disagreed with its start"): + protocol._handle_notification( + "item/completed", + { + **envelope, + "completedAtMs": 2, + "item": { + **resume, + "prompt": "follow-up", + "status": "completed", + "tool": "sendInput", + }, + }, + ) + protocol._handle_notification( + "item/completed", + { + **envelope, + "completedAtMs": 2, + "item": { + **resume, + "agentsStates": {"child-1": {"status": "completed"}}, + "status": "completed", + }, + }, + ) + self.assertEqual(protocol._active_nonspawn_items, {}) + protocol._handle_notification("turn/completed", completed) + self.assertIsNotNone(protocol._turn_completed) + + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + protocol._collab_thread_ids.add("child-1") + protocol._validated_collab_thread_ids.add("child-1") + protocol._seen_collab_turn_ids.add(("child-1", "child-turn")) + protocol._handle_notification("turn/completed", completed) + + self.assertIsNotNone(protocol._turn_completed) + def test_missing_usage_rejects_completed_turn(self) -> None: transport = ScriptedTransport(Path("/runtime/work"), omit_usage=True) with self.assertRaisesRegex(ProviderError, "token usage"): @@ -691,20 +1237,21 @@ def test_missing_usage_rejects_completed_turn(self) -> None: workspace=Path("/runtime/work"), system_context="test", locked_efforts=("low", "medium", "high", "xhigh", "max"), - locked_thread_cli_version="0.144.3", + locked_thread_cli_version="0.146.0", expected_codex_home=Path("/runtime/codex-home"), ).run("request", time.monotonic() + 5) def test_disagreeing_final_message_is_rejected(self) -> None: - for field in ("id", "phase", "text"): - with self.subTest(field=field): - transport = ScriptedTransport( - Path("/runtime/work"), - disagree_final_field=field, - turn_items_view="full", - ) - with self.assertRaisesRegex(ProviderError, "disagrees"): - _protocol(transport).run("request", time.monotonic() + 5) + for items_view in ("full", "summary"): + for field in ("id", "phase", "text"): + with self.subTest(items_view=items_view, field=field): + transport = ScriptedTransport( + Path("/runtime/work"), + disagree_final_field=field, + turn_items_view=items_view, + ) + with self.assertRaisesRegex(ProviderError, "disagrees"): + _protocol(transport).run("request", time.monotonic() + 5) def test_full_turn_rejects_duplicate_matching_completion_events(self) -> None: transport = ScriptedTransport( @@ -717,7 +1264,9 @@ def test_full_turn_rejects_duplicate_matching_completion_events(self) -> None: def test_not_loaded_turn_rejects_duplicate_completion_events(self) -> None: transport = ScriptedTransport( - Path("/runtime/work"), duplicate_item_completed=True + Path("/runtime/work"), + duplicate_item_completed=True, + turn_items_view="notLoaded", ) with self.assertRaisesRegex(ProviderError, "repeated an agent-message id"): _protocol(transport).run("request", time.monotonic() + 5) @@ -731,6 +1280,7 @@ def test_not_loaded_turn_rejects_multiple_final_messages(self) -> None: "text": "second final", "type": "agentMessage", }, + turn_items_view="notLoaded", ) with self.assertRaisesRegex(ProviderError, "multiple final-answer messages"): _protocol(transport).run("request", time.monotonic() + 5) @@ -740,7 +1290,7 @@ def test_last_agent_message_must_be_terminal(self) -> None: ("commentary", "ended with commentary"), (None, "final-answer message was not terminal"), ) - for items_view in ("notLoaded", "full"): + for items_view in ("notLoaded", "full", "summary"): for phase, expected in cases: with self.subTest(items_view=items_view, phase=phase): transport = ScriptedTransport( @@ -757,7 +1307,7 @@ def test_last_agent_message_must_be_terminal(self) -> None: _protocol(transport).run("request", time.monotonic() + 5) def test_terminal_agent_message_must_contain_non_whitespace_text(self) -> None: - for items_view in ("notLoaded", "full"): + for items_view in ("notLoaded", "full", "summary"): with self.subTest(items_view=items_view): transport = ScriptedTransport( Path("/runtime/work"), @@ -792,15 +1342,48 @@ def test_duplicate_agent_message_ids_are_rejected(self) -> None: with self.assertRaisesRegex(ProviderError, "repeated an agent-message id"): _protocol(transport).run("request", time.monotonic() + 5) - def test_full_and_legacy_full_turn_items_remain_compatible(self) -> None: - for items_view in ("full", None): + def test_full_and_summary_turn_items_are_validated(self) -> None: + for items_view in ("full", "summary"): with self.subTest(items_view=items_view): outcome = _protocol( ScriptedTransport(Path("/runtime/work"), turn_items_view=items_view) ).run("request", time.monotonic() + 5) self.assertEqual(outcome.final_output, "completed fixture") - self.assertEqual(outcome.raw_response["turn"]["items_view"], "full") + self.assertEqual(outcome.raw_response["turn"]["items_view"], items_view) + + def test_summary_uses_the_last_nonblank_completed_agent_message(self) -> None: + outcome = _protocol( + ScriptedTransport( + Path("/runtime/work"), + additional_item_completed={ + "id": "message-2", + "phase": None, + "text": " ", + "type": "agentMessage", + }, + turn_items_view="summary", + ) + ).run("request", time.monotonic() + 5) + + self.assertEqual(outcome.final_output, "completed fixture") + self.assertEqual(outcome.raw_response["turn"]["items_view"], "summary") + + def test_summary_cannot_discard_a_blank_final_answer(self) -> None: + transport = ScriptedTransport( + Path("/runtime/work"), + additional_item_completed={ + "id": "message-2", + "phase": "final_answer", + "text": " ", + "type": "agentMessage", + }, + final_phase=None, + turn_items_view="summary", + ) + + with self.assertRaisesRegex(ProviderError, "omitted.*final-answer"): + _protocol(transport).run("request", time.monotonic() + 5) def test_full_turn_rejects_non_object_items(self) -> None: transport = ScriptedTransport( @@ -813,34 +1396,44 @@ def test_full_turn_rejects_non_object_items(self) -> None: ): _protocol(transport).run("request", time.monotonic() + 5) - def test_summary_and_unknown_turn_item_views_are_rejected(self) -> None: - for items_view in ("summary", "future"): - with self.subTest(items_view=items_view): - with self.assertRaisesRegex(ProviderError, "unsupported item view"): - _protocol( - ScriptedTransport( - Path("/runtime/work"), turn_items_view=items_view - ) - ).run("request", time.monotonic() + 5) + def test_missing_and_unknown_turn_item_views_are_rejected(self) -> None: + for items_view, expected in ( + (None, "itemsView"), + ("future", "unsupported item view"), + ): + with ( + self.subTest(items_view=items_view), + self.assertRaisesRegex(ProviderError, expected), + ): + _protocol( + ScriptedTransport(Path("/runtime/work"), turn_items_view=items_view) + ).run("request", time.monotonic() + 5) def test_not_loaded_turn_items_must_be_empty(self) -> None: - transport = ScriptedTransport(Path("/runtime/work"), not_loaded_has_items=True) + transport = ScriptedTransport( + Path("/runtime/work"), + not_loaded_has_items=True, + turn_items_view="notLoaded", + ) with self.assertRaisesRegex( ProviderError, "not-loaded turn items must be empty" ): _protocol(transport).run("request", time.monotonic() + 5) def test_turn_requires_authoritative_completed_message(self) -> None: - for items_view in ("notLoaded", "full"): + for items_view in ("notLoaded", "full", "summary"): with self.subTest(items_view=items_view): transport = ScriptedTransport( Path("/runtime/work"), omit_item_completed=True, turn_items_view=items_view, ) - with self.assertRaisesRegex( - ProviderError, "completed final agent message" - ): + expected = ( + "summary message disagrees" + if items_view == "summary" + else "completed final agent message" + ) + with self.assertRaisesRegex(ProviderError, expected): _protocol(transport).run("request", time.monotonic() + 5) def test_skill_disable_refusal_is_rejected(self) -> None: @@ -855,7 +1448,7 @@ def test_skill_disable_refusal_is_rejected(self) -> None: workspace=Path("/runtime/work"), system_context="test", locked_efforts=("low", "medium", "high", "xhigh", "max"), - locked_thread_cli_version="0.144.3", + locked_thread_cli_version="0.146.0", expected_codex_home=Path("/runtime/codex-home"), ).run("request", time.monotonic() + 5) @@ -907,195 +1500,1627 @@ def on_dispatched() -> None: "request", time.monotonic() + 5 ) - self.assertEqual(outcome.final_output, "completed fixture") - self.assertEqual(dispatched_after, ["turn/start"]) + self.assertEqual(outcome.final_output, "completed fixture") + self.assertEqual(dispatched_after, ["turn/start"]) + + def test_repeated_pagination_cursor_and_total_overflow_fail_closed(self) -> None: + repeated = QueueTransport( + [ + _line( + {"id": 1, "result": {"data": [], "nextCursor": "again"}} + ).removesuffix(b"\n"), + _line( + {"id": 2, "result": {"data": [], "nextCursor": "again"}} + ).removesuffix(b"\n"), + ] + ) + with self.assertRaisesRegex(ProviderError, "repeated a pagination cursor"): + _protocol(repeated)._paged( + "model/list", {}, time.monotonic() + 1, maximum=10 + ) + + overflow = QueueTransport( + [ + _line( + {"id": 1, "result": {"data": [{}, {}], "nextCursor": None}} + ).removesuffix(b"\n") + ] + ) + with self.assertRaisesRegex(ProviderError, "total item limit"): + _protocol(overflow)._paged( + "model/list", {}, time.monotonic() + 1, maximum=1 + ) + + def test_untrusted_protocol_fields_are_not_echoed_in_errors(self) -> None: + secret = "SENTINEL_DO_NOT_ECHO" + frames = { + "duplicate key": ( + f'{{"id":1,"result":{{}},"{secret}":1,"{secret}":2}}'.encode() + ), + "unknown response key": _line( + {"id": 1, "result": {}, secret: True} + ).removesuffix(b"\n"), + "JSON-RPC message": _line( + {"error": {"code": -1, "message": secret}, "id": 1} + ).removesuffix(b"\n"), + "server request method": _line( + {"id": "server", "method": secret, "params": {}} + ).removesuffix(b"\n"), + "notification method": _line({"method": secret, "params": {}}).removesuffix( + b"\n" + ), + "notification method with invalid params": _line( + {"method": secret, "params": None} + ).removesuffix(b"\n"), + "error notification message": _line( + {"method": "error", "params": {"error": {"message": secret}}} + ).removesuffix(b"\n"), + } + notification_protocol = _protocol(QueueTransport([])) + for label, frame in frames.items(): + with self.subTest(label=label): + transport = QueueTransport([frame]) + session = _JsonRpcSession(transport) + with self.assertRaises(ProviderError) as caught: + session.call( + "initialize", + {}, + time.monotonic() + 1, + notification_protocol._handle_notification, + ) + self.assertNotIn(secret, str(caught.exception)) + + notification_protocol._thread_id = "thread-1" + notification_protocol._turn_id = "turn-1" + with self.assertRaises(ProviderError) as caught: + notification_protocol._handle_notification( + "item/completed", + { + "completedAtMs": 1, + "item": {"id": "item-1", "type": secret}, + "threadId": "thread-1", + "turnId": "turn-1", + }, + ) + self.assertNotIn(secret, str(caught.exception)) + + def test_prohibited_notifications_and_post_isolation_skill_changes_fail( + self, + ) -> None: + protocol = _protocol(QueueTransport([])) + with self.assertRaisesRegex(ProviderError, "prohibited"): + protocol._handle_notification("thread/settings/updated", {}) + with self.assertRaisesRegex(ProviderError, "changed after isolation"): + protocol._handle_notification("skills/changed", {}) + protocol._handle_notification( + "remoteControl/status/changed", + { + "environmentId": None, + "installationId": "installation-secret", + "serverName": "remote.example.invalid", + "status": "disabled", + }, + ) + with self.assertRaisesRegex(ProviderError, "not disabled"): + protocol._handle_notification( + "remoteControl/status/changed", + { + "environmentId": None, + "installationId": "installation-secret", + "serverName": "remote.example.invalid", + "status": "connected", + }, + ) + + protocol._handle_notification("thread/started", {"thread": {"id": "thread-1"}}) + with self.assertRaisesRegex(ProviderError, "more than one thread"): + protocol._handle_notification( + "thread/started", {"thread": {"id": "thread-1"}} + ) + + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + scoped_notifications = ( + "thread/status/changed", + "turn/diff/updated", + "model/verification", + ) + for method in scoped_notifications: + with self.subTest(method=method, scope="empty"): + with self.assertRaises(ProviderError): + protocol._handle_notification(method, {}) + with self.subTest(method=method, scope="wrong"): + with self.assertRaises(ProviderError): + protocol._handle_notification( + method, + {"threadId": "other-thread", "turnId": "other-turn"}, + ) + + def test_item_type_allowlist_matches_isolated_local_tools(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + allowed = ( + "agentMessage", + "collabAgentToolCall", + "commandExecution", + "contextCompaction", + "fileChange", + "imageView", + "reasoning", + "userMessage", + ) + self.assertEqual(_ALLOWED_ITEM_TYPES, frozenset(allowed)) + for item_type in allowed: + with self.subTest(item_type=item_type, disposition="allowed"): + item = {"id": "item-1", "type": item_type} + if item_type == "agentMessage": + item["text"] = "message" + elif item_type == "collabAgentToolCall": + item.update( + { + "agentsStates": {}, + "model": "", + "prompt": "delegated task", + "reasoningEffort": "medium", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + } + ) + protocol._handle_notification( + "item/started", + { + "item": item, + "startedAtMs": 1, + "threadId": "thread-1", + "turnId": "turn-1", + }, + ) + + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": "child-1", + }, + ) + with self.assertRaisesRegex( + ProviderError, "thread/status/changed changed thread scope" + ): + protocol._handle_notification( + "thread/status/changed", + {"status": {"type": "idle"}, "threadId": "unknown-child"}, + ) + + sentinel = "SENTINEL_MUST_NOT_BE_DISCLOSED" + prohibited = ( + "dynamicToolCall", + "enteredReviewMode", + "exitedReviewMode", + "hookPrompt", + "imageGeneration", + "mcpToolCall", + "plan", + "sleep", + "subAgentActivity", + "webSearch", + sentinel, + ) + for item_type in prohibited: + with self.subTest(item_type=item_type, disposition="prohibited"): + with self.assertRaisesRegex( + ProviderError, "prohibited or unknown Codex item type" + ) as raised: + protocol._handle_notification( + "item/started", + { + "item": {"id": "item-1", "type": item_type}, + "startedAtMs": 1, + "threadId": "thread-1", + "turnId": "turn-1", + }, + ) + self.assertNotIn(sentinel, str(raised.exception)) + + def test_spawn_agent_matches_the_pinned_wire_lifecycle(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + envelope = {"threadId": "thread-1", "turnId": "turn-1"} + initial = { + "agentsStates": {}, + "id": "item-1", + "model": "", + "prompt": "delegated task", + "reasoningEffort": "medium", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._handle_notification( + "item/started", {**envelope, "item": initial, "startedAtMs": 1} + ) + self.assertEqual(protocol._collab_thread_ids, set()) + self.assertEqual( + protocol._pending_spawn_items, + { + ("thread-1", "item-1"): ( + None, + hashlib.sha256(b"delegated task").digest(), + ) + }, + ) + with self.assertRaisesRegex(ProviderError, "already in progress"): + protocol._handle_notification( + "item/started", {**envelope, "item": initial, "startedAtMs": 1} + ) + + completed = { + **initial, + "agentsStates": {"child-1": {"status": "completed"}}, + "model": "gpt-5.6-luna", + "reasoningEffort": "low", + "receiverThreadIds": ["child-1"], + "status": "completed", + } + protocol._handle_notification( + "item/completed", {**envelope, "completedAtMs": 2, "item": completed} + ) + self.assertEqual(protocol._collab_thread_ids, {"child-1"}) + self.assertEqual(protocol._collab_parent_ids, {"child-1": "thread-1"}) + self.assertEqual(protocol._pending_spawn_items, {}) + + def test_spawn_agent_accepts_explicit_pinned_start_and_failure_metadata( + self, + ) -> None: + for terminal_status, receivers in ( + ("completed", ["child-1"]), + ("failed", []), + ): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + initial = { + "agentsStates": {}, + "id": "item-1", + "model": "gpt-5.6-luna", + "prompt": "delegated task", + "reasoningEffort": "low", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + + with self.subTest(terminal_status=terminal_status): + protocol._validate_item(initial, lifecycle="started") + protocol._validate_item( + { + **initial, + "agentsStates": { + receiver: {"status": "running"} for receiver in receivers + }, + "receiverThreadIds": receivers, + "status": terminal_status, + }, + lifecycle="completed", + ) + + def test_terminal_collaboration_history_retains_fixed_size_signatures(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._collab_thread_ids.add("child-1") + prompt = "delegated " + "☄" * 50_000 + initial = { + "agentsStates": {}, + "id": "item-1", + "prompt": prompt, + "receiverThreadIds": ["child-1"], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "sendInput", + "type": "collabAgentToolCall", + } + + protocol._validate_item(initial, lifecycle="started") + protocol._validate_item( + { + **initial, + "status": "completed", + }, + lifecycle="completed", + ) + signature = protocol._terminal_collab_history[("thread-1", "item-1")] + self.assertIsInstance(signature, bytes) + self.assertEqual(len(signature), hashlib.sha256().digest_size) + + def test_wait_completion_accepts_timeout_subset_and_unordered_receivers( + self, + ) -> None: + cases = ( + ([], {}), + (["child-1"], {"child-1": {"status": "completed"}}), + ( + ["child-2", "child-1"], + { + "child-2": {"status": "completed"}, + "child-1": {"status": "completed"}, + }, + ), + ) + for receivers, states in cases: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._collab_thread_ids.update(("child-1", "child-2")) + initial = { + "agentsStates": {}, + "id": "wait-1", + "receiverThreadIds": ["child-1", "child-2"], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "wait", + "type": "collabAgentToolCall", + } + + with self.subTest(receivers=receivers): + protocol._validate_item(initial, lifecycle="started") + protocol._validate_item( + { + **initial, + "agentsStates": states, + "receiverThreadIds": receivers, + "status": "completed", + }, + lifecycle="completed", + ) + + def test_send_input_completion_prompt_must_match_its_start(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._collab_thread_ids.add("child-1") + initial = { + "agentsStates": {}, + "id": "send-1", + "prompt": "original input", + "receiverThreadIds": ["child-1"], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "sendInput", + "type": "collabAgentToolCall", + } + protocol._validate_item(initial, lifecycle="started") + + with self.assertRaisesRegex(ProviderError, "prompt disagreed"): + protocol._validate_item( + {**initial, "prompt": "changed input", "status": "completed"}, + lifecycle="completed", + ) + + def test_successful_close_deactivates_receiver_and_descendants(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._collab_thread_ids.update(("child-1", "grandchild-1")) + protocol._collab_parent_ids.update( + {"child-1": "thread-1", "grandchild-1": "child-1"} + ) + protocol._active_collab_thread_ids.update(("child-1", "grandchild-1")) + initial = { + "agentsStates": {}, + "id": "close-1", + "receiverThreadIds": ["child-1"], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "closeAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(initial, lifecycle="started") + protocol._validate_item( + { + **initial, + "agentsStates": {"child-1": {"status": "running"}}, + "status": "completed", + }, + lifecycle="completed", + ) + + self.assertEqual(protocol._active_collab_thread_ids, set()) + + def test_item_lifecycle_requires_pinned_timestamp_fields(self) -> None: + item = { + "id": "message-1", + "phase": "final_answer", + "text": "result", + "type": "agentMessage", + } + for method, field in ( + ("item/started", "startedAtMs"), + ("item/completed", "completedAtMs"), + ): + for value, expected in ((None, "required"), (True, "supported range")): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + params = { + "item": item, + "threadId": "thread-1", + "turnId": "turn-1", + } + if value is not None: + params[field] = value + with ( + self.subTest(method=method, value=value), + self.assertRaisesRegex(ProviderError, expected), + ): + protocol._handle_notification(method, params) + + def test_spawn_completion_prompt_must_match_its_start(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + initial = { + "agentsStates": {}, + "id": "item-1", + "model": "", + "prompt": "original task", + "reasoningEffort": "medium", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + envelope = {"threadId": "thread-1", "turnId": "turn-1"} + protocol._handle_notification( + "item/started", {**envelope, "item": initial, "startedAtMs": 1} + ) + + with self.assertRaisesRegex(ProviderError, "prompt disagreed"): + protocol._handle_notification( + "item/completed", + { + **envelope, + "completedAtMs": 2, + "item": { + **initial, + "agentsStates": {"child-1": {"status": "running"}}, + "model": "gpt-5.6-luna", + "prompt": "changed task", + "reasoningEffort": "low", + "receiverThreadIds": ["child-1"], + "status": "completed", + }, + }, + ) + + self.assertEqual( + protocol._pending_spawn_items, + { + ("thread-1", "item-1"): ( + None, + hashlib.sha256(b"original task").digest(), + ) + }, + ) + + def test_stale_spawn_completion_does_not_reactivate_completed_child(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + initial = { + "agentsStates": {}, + "id": "item-1", + "model": "", + "prompt": "delegated task", + "reasoningEffort": "medium", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + envelope = {"threadId": "thread-1", "turnId": "turn-1"} + protocol._handle_notification( + "item/started", {**envelope, "item": initial, "startedAtMs": 1} + ) + protocol._collab_thread_ids.add("child-1") + protocol._unbound_collab_thread_ids.add("child-1") + protocol._seen_collab_turn_ids.add(("child-1", "child-turn")) + + protocol._handle_notification( + "item/completed", + { + **envelope, + "completedAtMs": 2, + "item": { + **initial, + "agentsStates": {"child-1": {"status": "running"}}, + "model": "gpt-5.6-luna", + "reasoningEffort": "low", + "receiverThreadIds": ["child-1"], + "status": "completed", + }, + }, + ) + + self.assertNotIn("child-1", protocol._active_collab_thread_ids) + + def test_spawn_start_rejects_an_explicit_receiver(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + item = { + "agentsStates": {}, + "id": "item-1", + "model": "", + "prompt": "delegated task", + "reasoningEffort": "medium", + "receiverThreadIds": ["child-1"], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + + with self.assertRaisesRegex(ProviderError, "must not have receivers"): + protocol._handle_notification( + "item/started", + { + "item": item, + "startedAtMs": 1, + "threadId": "thread-1", + "turnId": "turn-1", + }, + ) + self.assertEqual(protocol._pending_spawn_items, {}) + + def test_failed_spawn_releases_unstarted_receiver(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + initial = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(initial) + protocol._validate_item( + { + **initial, + "agentsStates": {"child-1": {"status": "errored"}}, + "receiverThreadIds": ["child-1"], + "status": "failed", + } + ) + + self.assertEqual(protocol._pending_spawn_items, {}) + self.assertEqual(protocol._collab_thread_ids, set()) + self.assertEqual(protocol._collab_parent_ids, {}) + + def test_spawn_snapshot_does_not_mutate_live_scope(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._collab_thread_ids.add("child-1") + protocol._pending_spawn_items[("thread-1", "item-1")] = ( + "child-1", + None, + ) + + with self.assertRaisesRegex(ProviderError, "live terminal history"): + protocol._validate_item( + { + "agentsStates": {"child-1": {"status": "errored"}}, + "id": "item-1", + "receiverThreadIds": ["child-1"], + "senderThreadId": "thread-1", + "status": "failed", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + }, + owner_thread_id="thread-1", + lifecycle="snapshot", + ) + + self.assertEqual( + protocol._pending_spawn_items, + {("thread-1", "item-1"): ("child-1", None)}, + ) + self.assertEqual(protocol._collab_thread_ids, {"child-1"}) + + def test_spawn_snapshot_requires_matching_live_terminal_history(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + initial = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + completed = { + **initial, + "agentsStates": {"child-1": {"status": "running"}}, + "model": "gpt-5.6-luna", + "reasoningEffort": "low", + "receiverThreadIds": ["child-1"], + "status": "completed", + } + + with self.assertRaisesRegex(ProviderError, "live terminal history"): + protocol._validate_item(completed, lifecycle="snapshot") + protocol._validate_item(initial) + protocol._validate_item(completed) + live_scope = ( + dict(protocol._terminal_collab_history), + set(protocol._collab_thread_ids), + dict(protocol._collab_parent_ids), + ) + protocol._validate_item(completed, lifecycle="snapshot") + with self.assertRaisesRegex(ProviderError, "live terminal history"): + protocol._validate_item( + { + **completed, + "agentsStates": {}, + "model": "", + "reasoningEffort": "medium", + "receiverThreadIds": [], + "status": "failed", + }, + lifecycle="snapshot", + ) + self.assertEqual( + live_scope, + ( + protocol._terminal_collab_history, + protocol._collab_thread_ids, + protocol._collab_parent_ids, + ), + ) + + def test_completed_spawn_requires_exactly_one_receiver(self) -> None: + initial = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + for receivers in ([], ["child-1", "child-2"]): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._validate_item(initial) + with self.subTest(receivers=receivers): + with self.assertRaisesRegex(ProviderError, "exactly one receiver"): + protocol._validate_item( + { + **initial, + "receiverThreadIds": receivers, + "status": "completed", + } + ) + self.assertEqual( + protocol._pending_spawn_items, + {("thread-1", "item-1"): (None, None)}, + ) + self.assertEqual(protocol._collab_thread_ids, set()) + + def test_spawn_rejects_multiple_receivers_before_successful_completion( + self, + ) -> None: + initial = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + for status in ("inProgress", "failed"): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + if status == "failed": + protocol._validate_item(initial) + with self.subTest(status=status): + with self.assertRaisesRegex(ProviderError, "at most one receiver"): + protocol._validate_item( + { + **initial, + "receiverThreadIds": ["child-1", "child-2"], + "status": status, + } + ) + expected_pending = ( + {("thread-1", "item-1"): (None, None)} if status == "failed" else {} + ) + self.assertEqual(protocol._pending_spawn_items, expected_pending) + self.assertEqual(protocol._collab_thread_ids, set()) + + def test_live_terminal_spawn_requires_a_pending_start(self) -> None: + base = { + "agentsStates": {}, + "id": "spawn-1", + "senderThreadId": "thread-1", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + for status, receivers in (("failed", []), ("completed", ["child-1"])): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + protocol._collab_thread_ids.add("child-1") + item = { + **base, + "model": "" if status == "failed" else "gpt-5.6-luna", + "prompt": "delegated task", + "reasoningEffort": "medium" if status == "failed" else "low", + "receiverThreadIds": receivers, + "status": status, + } + with self.subTest(status=status): + with self.assertRaisesRegex(ProviderError, "without a pending child"): + protocol._handle_notification( + "item/completed", + { + "completedAtMs": 2, + "item": item, + "threadId": "thread-1", + "turnId": "turn-1", + }, + ) + + with self.assertRaisesRegex(ProviderError, "live terminal history"): + protocol._validate_item(item, lifecycle="snapshot") + + def test_terminal_collaboration_item_id_cannot_restart(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._turn_id = "turn-1" + item = { + "agentsStates": {}, + "id": "spawn-1", + "model": "", + "prompt": "delegated task", + "reasoningEffort": "medium", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + envelope = {"threadId": "thread-1", "turnId": "turn-1"} + protocol._handle_notification( + "item/started", {**envelope, "item": item, "startedAtMs": 1} + ) + protocol._handle_notification( + "item/completed", + { + **envelope, + "completedAtMs": 2, + "item": {**item, "status": "failed"}, + }, + ) + + with self.assertRaisesRegex(ProviderError, "reused after termination"): + protocol._handle_notification( + "item/started", {**envelope, "item": item, "startedAtMs": 1} + ) + protocol._validate_item({**item, "status": "failed"}, lifecycle="snapshot") + + def test_spawn_agent_bounds_total_child_thread_scope(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + receivers = [f"child-{index}" for index in range(_MAX_COLLAB_THREADS)] + base = { + "agentsStates": {}, + "senderThreadId": "thread-1", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + for index, receiver in enumerate(receivers): + item = { + **base, + "id": f"item-{index}", + "receiverThreadIds": [receiver], + } + protocol._validate_item({**item, "status": "inProgress"}) + protocol._validate_item({**item, "status": "completed"}) + + with self.assertRaisesRegex(ProviderError, "thread count exceeds"): + protocol._validate_item( + { + **base, + "id": "overflow-item", + "receiverThreadIds": ["overflow-child"], + "status": "inProgress", + } + ) + + protocol._validate_item( + { + **base, + "id": "pending-overflow", + "receiverThreadIds": [], + "status": "inProgress", + } + ) + with self.assertRaisesRegex(ProviderError, "thread count exceeds"): + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": "overflow-child", + }, + ) + self.assertEqual(protocol._collab_thread_ids, set(receivers)) + self.assertEqual( + protocol._pending_spawn_items, + {("thread-1", "pending-overflow"): (None, None)}, + ) + + def test_pending_spawn_bounds_early_child_status_scope(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + initial = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(initial) + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": "child-1", + }, + ) + self.assertEqual(protocol._collab_thread_ids, {"child-1"}) + self.assertEqual( + protocol._pending_spawn_items, {("thread-1", "item-1"): (None, None)} + ) + self.assertEqual(protocol._unbound_collab_thread_ids, {"child-1"}) + + protocol._turn_id = "main-turn" + with self.assertRaisesRegex(ProviderError, "preceded its provenance"): + protocol._handle_notification( + "turn/started", + {"threadId": "child-1", "turn": {"id": "child-turn"}}, + ) + protocol._handle_notification("thread/started", {"thread": _child_thread()}) + protocol._handle_notification( + "turn/started", + {"threadId": "child-1", "turn": {"id": "child-turn"}}, + ) + protocol._handle_notification( + "item/started", + { + "item": {"id": "reasoning-1", "type": "reasoning"}, + "startedAtMs": 1, + "threadId": "child-1", + "turnId": "child-turn", + }, + ) + protocol._handle_notification( + "model/verification", + {"threadId": "child-1", "turnId": "child-turn"}, + ) + protocol._handle_notification( + "item/completed", + { + "completedAtMs": 2, + "item": { + "id": "message-1", + "phase": "final_answer", + "text": "child result", + "type": "agentMessage", + }, + "threadId": "child-1", + "turnId": "child-turn", + }, + ) + protocol._handle_notification( + "thread/tokenUsage/updated", + { + "threadId": "child-1", + "tokenUsage": { + "last": { + "cachedInputTokens": 0, + "inputTokens": 1, + "outputTokens": 1, + "reasoningOutputTokens": 0, + "totalTokens": 2, + } + }, + "turnId": "child-turn", + }, + ) + protocol._handle_notification( + "turn/completed", + { + "threadId": "child-1", + "turn": { + "id": "child-turn", + "items": [], + "itemsView": "notLoaded", + "status": "completed", + }, + }, + ) + self.assertIsNone(protocol._announced_turn_id) + self.assertEqual(protocol._completed_messages, []) + self.assertIsNone(protocol._turn_completed) + + with self.assertRaisesRegex( + ProviderError, "thread/status/changed changed thread scope" + ): + protocol._handle_notification( + "thread/status/changed", + {"status": {"type": "idle"}, "threadId": "unknown-child"}, + ) + with self.assertRaisesRegex(ProviderError, "preceded its provenance"): + protocol._handle_notification( + "turn/started", + {"threadId": "unknown-child", "turn": {"id": "unknown-turn"}}, + ) + with self.assertRaisesRegex(ProviderError, "did not match"): + protocol._validate_item( + { + **initial, + "receiverThreadIds": ["different-child"], + "status": "completed", + } + ) + + protocol._validate_item( + { + **initial, + "agentsStates": {"child-1": {"status": "completed"}}, + "receiverThreadIds": ["child-1"], + "status": "completed", + } + ) + self.assertEqual(protocol._pending_spawn_items, {}) + self.assertEqual(protocol._unbound_collab_thread_ids, set()) + + def test_receiverless_spawn_failure_releases_unambiguous_child_scope(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + item = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(item) + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": "child-1", + }, + ) + + protocol._validate_item({**item, "status": "failed"}) + + self.assertEqual(protocol._pending_spawn_items, {}) + self.assertEqual(protocol._unbound_collab_thread_ids, set()) + self.assertEqual(protocol._collab_thread_ids, set()) + self.assertEqual(protocol._failed_collab_thread_ids, {"child-1"}) + + protocol._validate_item({**item, "id": "item-2"}) + with self.assertRaisesRegex(ProviderError, "failed spawn child"): + protocol._validate_item( + { + **item, + "id": "item-2", + "receiverThreadIds": ["child-1"], + } + ) + with self.assertRaisesRegex(ProviderError, "failed spawn child"): + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": "child-1", + }, + ) + + def test_child_thread_provenance_is_validated_before_scope_claim(self) -> None: + initial = { + "agentsStates": {}, + "id": "spawn-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + mutations = { + "model provider": lambda thread: thread.update(modelProvider="other"), + "CLI version": lambda thread: thread.update(cliVersion="0.145.0"), + "working directory": lambda thread: thread.update(cwd="/other"), + "session": lambda thread: thread.update(sessionId="other-session"), + "parent": lambda thread: thread.update(parentThreadId="other-parent"), + "source parent": lambda thread: thread["source"]["subAgent"][ + "thread_spawn" + ].update(parent_thread_id="other-parent"), + } + for label, mutate in mutations.items(): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + protocol._validate_item(initial) + thread = _child_thread() + mutate(thread) + with self.subTest(label=label), self.assertRaises(ProviderError): + protocol._handle_notification("thread/started", {"thread": thread}) + self.assertEqual(protocol._collab_thread_ids, set()) + self.assertEqual(protocol._unbound_collab_thread_ids, set()) + self.assertEqual(protocol._validated_collab_thread_ids, set()) + + for field in ("agent_nickname", "agent_path", "agent_role"): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + protocol._validate_item(initial) + thread = _child_thread() + thread["source"]["subAgent"]["thread_spawn"][field] = {} + with ( + self.subTest(field=field), + self.assertRaisesRegex(ProviderError, "string or null"), + ): + protocol._handle_notification("thread/started", {"thread": thread}) + self.assertEqual(protocol._collab_thread_ids, set()) + self.assertEqual(protocol._unbound_collab_thread_ids, set()) + self.assertEqual(protocol._validated_collab_thread_ids, set()) + + invalid_metadata = { + "agentNickname": {}, + "agentRole": [], + "canAcceptDirectInput": "true", + "extra": [], + "gitInfo": {"branch": []}, + "isPinned": 0, + "name": {}, + "recencyAt": "now", + "unknownMetadata": None, + } + for field, value in invalid_metadata.items(): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + protocol._validate_item(initial) + thread = _child_thread() + thread[field] = value + with self.subTest(field=field), self.assertRaises(ProviderError): + protocol._handle_notification("thread/started", {"thread": thread}) + self.assertEqual(protocol._collab_thread_ids, set()) + self.assertEqual(protocol._validated_collab_thread_ids, set()) + + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + protocol._validate_item(initial) + thread = _child_thread() + thread.update( + { + "agentNickname": "swift-fox", + "agentRole": None, + "canAcceptDirectInput": True, + "extra": {}, + "gitInfo": {"branch": None, "originUrl": None, "sha": None}, + "isPinned": False, + "name": None, + "recencyAt": 1_700_000_001, + } + ) + protocol._handle_notification("thread/started", {"thread": thread}) + self.assertEqual(protocol._validated_collab_thread_ids, {"child-1"}) + + def test_child_thread_parent_must_own_a_pending_spawn(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + protocol._collab_thread_ids.add("child-a") + protocol._validated_collab_thread_ids.add("child-a") + protocol._validate_item( + { + "agentsStates": {}, + "id": "spawn-a", + "receiverThreadIds": [], + "senderThreadId": "child-a", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + ) + + with self.assertRaisesRegex(ProviderError, "parent-owned spawn history"): + protocol._handle_notification("thread/started", {"thread": _child_thread()}) + self.assertNotIn("child-1", protocol._validated_collab_thread_ids) + + root_spawn = { + "agentsStates": {}, + "id": "spawn-root", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(root_spawn) + protocol._handle_notification("thread/started", {"thread": _child_thread()}) + with self.assertRaisesRegex(ProviderError, "spawning sender"): + protocol._validate_item( + { + **root_spawn, + "agentsStates": {"child-1": {"status": "running"}}, + "id": "spawn-a", + "receiverThreadIds": ["child-1"], + "senderThreadId": "child-a", + "status": "completed", + } + ) + + def test_child_thread_depth_matches_its_parent_chain(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + spawn = { + "agentsStates": {}, + "id": "spawn-root", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(spawn) + with self.assertRaisesRegex(ProviderError, "provenance"): + protocol._handle_notification( + "thread/started", + {"thread": _child_thread(depth=_MAX_COLLAB_THREADS)}, + ) + self.assertEqual(protocol._collab_depths, {}) + + protocol._handle_notification("thread/started", {"thread": _child_thread()}) + protocol._validate_item( + { + **spawn, + "agentsStates": {"child-1": {"status": "running"}}, + "receiverThreadIds": ["child-1"], + "status": "completed", + } + ) + nested_spawn = { + **spawn, + "id": "spawn-nested", + "senderThreadId": "child-1", + } + protocol._validate_item(nested_spawn) + with self.assertRaisesRegex(ProviderError, "provenance"): + protocol._handle_notification( + "thread/started", + { + "thread": _child_thread( + "child-2", + parent_id="child-1", + ) + }, + ) + protocol._handle_notification( + "thread/started", + { + "thread": _child_thread( + "child-2", + depth=2, + parent_id="child-1", + ) + }, + ) + self.assertEqual(protocol._collab_depths, {"child-1": 1, "child-2": 2}) + + def test_child_thread_parent_matches_preannouncement_spawn_binding(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + root_spawn = { + "agentsStates": {}, + "id": "spawn-root", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(root_spawn) + protocol._validate_item( + { + **root_spawn, + "agentsStates": {"child-1": {"status": "running"}}, + "receiverThreadIds": ["child-1"], + "status": "completed", + } + ) + protocol._collab_thread_ids.add("child-a") + protocol._validated_collab_thread_ids.add("child-a") + protocol._validate_item( + { + **root_spawn, + "id": "spawn-child", + "senderThreadId": "child-a", + } + ) + thread = _child_thread() + thread["parentThreadId"] = "child-a" + thread["source"]["subAgent"]["thread_spawn"]["parent_thread_id"] = "child-a" + + with self.assertRaisesRegex(ProviderError, "provenance"): + protocol._handle_notification("thread/started", {"thread": thread}) + self.assertEqual(protocol._collab_parent_ids["child-1"], "thread-1") + self.assertNotIn("child-1", protocol._validated_collab_thread_ids) + + def test_child_thread_accepts_matching_completed_spawn_history(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._session_id = "session-1" + spawn = { + "agentsStates": {}, + "id": "spawn-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item(spawn) + protocol._validate_item( + { + **spawn, + "agentsStates": {"child-1": {"status": "running"}}, + "receiverThreadIds": ["child-1"], + "status": "completed", + } + ) + + protocol._handle_notification("thread/started", {"thread": _child_thread()}) + self.assertEqual(protocol._validated_collab_thread_ids, {"child-1"}) + + def test_parallel_pending_spawns_accept_cross_ordered_child_announcements( + self, + ) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + initial = { + "agentsStates": {}, + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item({**initial, "id": "spawn-1"}) + protocol._validate_item({**initial, "id": "spawn-2"}) + + for child in ("child-2", "child-1"): + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": child, + }, + ) + + for item_id, child in (("spawn-1", "child-1"), ("spawn-2", "child-2")): + protocol._validate_item( + { + **initial, + "agentsStates": {child: {"status": "running"}}, + "id": item_id, + "receiverThreadIds": [child], + "status": "completed", + } + ) + + self.assertEqual(protocol._pending_spawn_items, {}) + self.assertEqual(protocol._unbound_collab_thread_ids, set()) + self.assertEqual(protocol._collab_thread_ids, {"child-1", "child-2"}) + + def test_pending_spawn_ids_are_scoped_to_the_sender_thread(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._collab_thread_ids.add("child-a") + base = { + "agentsStates": {}, + "id": "same-id", + "receiverThreadIds": [], + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item({**base, "senderThreadId": "main-thread"}) + protocol._validate_item({**base, "senderThreadId": "child-a"}) + self.assertEqual( + set(protocol._pending_spawn_items), + {("main-thread", "same-id"), ("child-a", "same-id")}, + ) + + protocol._validate_item( + { + **base, + "agentsStates": {"child-b": {"status": "running"}}, + "receiverThreadIds": ["child-b"], + "senderThreadId": "child-a", + "status": "completed", + } + ) + self.assertEqual( + protocol._pending_spawn_items, + {("main-thread", "same-id"): (None, None)}, + ) + protocol._validate_item( + { + **base, + "agentsStates": {"child-c": {"status": "running"}}, + "receiverThreadIds": ["child-c"], + "senderThreadId": "main-thread", + "status": "completed", + } + ) + self.assertEqual(protocol._pending_spawn_items, {}) + + def test_child_collaboration_item_can_target_root_thread(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._collab_thread_ids.add("child-1") + + protocol._validate_item( + { + "agentsStates": {}, + "id": "message-1", + "prompt": "root follow-up", + "receiverThreadIds": ["main-thread"], + "senderThreadId": "child-1", + "status": "completed", + "tool": "sendInput", + "type": "collabAgentToolCall", + } + ) + + def test_one_child_cannot_complete_two_pending_spawns(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + base = { + "agentsStates": {}, + "receiverThreadIds": [], + "senderThreadId": "main-thread", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + protocol._validate_item({**base, "id": "spawn-1"}) + protocol._validate_item({**base, "id": "spawn-2"}) + protocol._handle_notification( + "thread/status/changed", + { + "status": {"activeFlags": [], "type": "active"}, + "threadId": "only-child", + }, + ) + completed = { + **base, + "agentsStates": {"only-child": {"status": "running"}}, + "receiverThreadIds": ["only-child"], + "status": "completed", + } + protocol._validate_item({**completed, "id": "spawn-1"}) + with self.assertRaisesRegex(ProviderError, "already claimed"): + protocol._validate_item({**completed, "id": "spawn-2"}) + self.assertEqual( + protocol._pending_spawn_items, + {("main-thread", "spawn-2"): (None, None)}, + ) + + def test_collaboration_optional_metadata_rejects_non_strings(self) -> None: + base = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + for field in ("model", "prompt", "reasoningEffort"): + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + with ( + self.subTest(field=field), + self.assertRaisesRegex( + ProviderError, f"collaboration {field} must be a string or null" + ), + ): + protocol._validate_item({**base, field: {"malformed": True}}) - def test_repeated_pagination_cursor_and_total_overflow_fail_closed(self) -> None: - repeated = QueueTransport( - [ - _line( - {"id": 1, "result": {"data": [], "nextCursor": "again"}} - ).removesuffix(b"\n"), - _line( - {"id": 2, "result": {"data": [], "nextCursor": "again"}} - ).removesuffix(b"\n"), - ] + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + with self.assertRaisesRegex(ProviderError, "non-empty string or null"): + protocol._validate_item({**base, "reasoningEffort": ""}) + + def test_collaboration_model_metadata_matches_pinned_configuration(self) -> None: + base = { + "agentsStates": {}, + "id": "item-1", + "receiverThreadIds": [], + "senderThreadId": "thread-1", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", + } + mismatches = ( + ("model", "other-model", "pinned model"), + ("reasoningEffort", "high", "pinned reasoning effort"), ) - with self.assertRaisesRegex(ProviderError, "repeated a pagination cursor"): - _protocol(repeated)._paged( - "model/list", {}, time.monotonic() + 1, maximum=10 - ) + for field, value, message in mismatches: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + with ( + self.subTest(field=field), + self.assertRaisesRegex(ProviderError, message), + ): + protocol._validate_item({**base, field: value}) + self.assertEqual(protocol._pending_spawn_items, {}) - overflow = QueueTransport( - [ - _line( - {"id": 1, "result": {"data": [{}, {}], "nextCursor": None}} - ).removesuffix(b"\n") - ] + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "thread-1" + protocol._validate_item( + { + **base, + "model": "gpt-5.6-luna", + "reasoningEffort": "low", + } + ) + self.assertEqual( + protocol._pending_spawn_items, {("thread-1", "item-1"): (None, None)} ) - with self.assertRaisesRegex(ProviderError, "total item limit"): - _protocol(overflow)._paged( - "model/list", {}, time.monotonic() + 1, maximum=1 - ) - def test_untrusted_protocol_fields_are_not_echoed_in_errors(self) -> None: - secret = "SENTINEL_DO_NOT_ECHO" - frames = { - "duplicate key": ( - f'{{"id":1,"result":{{}},"{secret}":1,"{secret}":2}}'.encode() - ), - "unknown response key": _line( - {"id": 1, "result": {}, secret: True} - ).removesuffix(b"\n"), - "JSON-RPC message": _line( - {"error": {"code": -1, "message": secret}, "id": 1} - ).removesuffix(b"\n"), - "server request method": _line( - {"id": "server", "method": secret, "params": {}} - ).removesuffix(b"\n"), - "notification method": _line({"method": secret, "params": {}}).removesuffix( - b"\n" - ), - "notification method with invalid params": _line( - {"method": secret, "params": None} - ).removesuffix(b"\n"), - "error notification message": _line( - {"method": "error", "params": {"error": {"message": secret}}} - ).removesuffix(b"\n"), + def test_collaboration_sender_and_lifecycle_match_event_envelope(self) -> None: + protocol = _protocol(QueueTransport([])) + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + protocol._collab_thread_ids.add("child-1") + protocol._collab_turn_ids["child-1"] = {"child-turn"} + item = { + "agentsStates": {}, + "id": "spawn-1", + "receiverThreadIds": [], + "senderThreadId": "main-thread", + "status": "inProgress", + "tool": "spawnAgent", + "type": "collabAgentToolCall", } - notification_protocol = _protocol(QueueTransport([])) - for label, frame in frames.items(): - with self.subTest(label=label): - transport = QueueTransport([frame]) - session = _JsonRpcSession(transport) - with self.assertRaises(ProviderError) as caught: - session.call( - "initialize", - {}, - time.monotonic() + 1, - notification_protocol._handle_notification, - ) - self.assertNotIn(secret, str(caught.exception)) - notification_protocol._thread_id = "thread-1" - notification_protocol._turn_id = "turn-1" - with self.assertRaises(ProviderError) as caught: - notification_protocol._handle_notification( + with self.assertRaisesRegex(ProviderError, "disagrees with its envelope"): + protocol._handle_notification( + "item/started", + { + "item": item, + "startedAtMs": 1, + "threadId": "child-1", + "turnId": "child-turn", + }, + ) + self.assertEqual(protocol._pending_spawn_items, {}) + + with self.assertRaisesRegex(ProviderError, "still in progress"): + protocol._handle_notification( "item/completed", { - "item": {"id": "item-1", "type": secret}, - "threadId": "thread-1", - "turnId": "turn-1", + "completedAtMs": 2, + "item": item, + "threadId": "main-thread", + "turnId": "main-turn", }, ) - self.assertNotIn(secret, str(caught.exception)) + self.assertEqual(protocol._pending_spawn_items, {}) - def test_prohibited_notifications_and_post_isolation_skill_changes_fail( - self, - ) -> None: + def test_child_payloads_are_validated_before_they_are_discarded(self) -> None: protocol = _protocol(QueueTransport([])) - with self.assertRaisesRegex(ProviderError, "prohibited"): - protocol._handle_notification("thread/settings/updated", {}) - with self.assertRaisesRegex(ProviderError, "changed after isolation"): - protocol._handle_notification("skills/changed", {}) - protocol._handle_notification( - "remoteControl/status/changed", - { - "environmentId": None, - "installationId": "installation-secret", - "serverName": "remote.example.invalid", - "status": "disabled", - }, - ) - with self.assertRaisesRegex(ProviderError, "not disabled"): + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + protocol._collab_thread_ids.add("child-1") + protocol._collab_turn_ids["child-1"] = {"child-turn"} + + with self.assertRaisesRegex(ProviderError, "agent message omitted keys"): protocol._handle_notification( - "remoteControl/status/changed", + "item/completed", { - "environmentId": None, - "installationId": "installation-secret", - "serverName": "remote.example.invalid", - "status": "connected", + "completedAtMs": 2, + "item": {"type": "agentMessage"}, + "threadId": "child-1", + "turnId": "child-turn", }, ) - - protocol._handle_notification("thread/started", {"thread": {"id": "thread-1"}}) - with self.assertRaisesRegex(ProviderError, "more than one thread"): + with self.assertRaisesRegex(ProviderError, "status must be a non-empty string"): protocol._handle_notification( - "thread/started", {"thread": {"id": "thread-1"}} + "turn/completed", + { + "threadId": "child-1", + "turn": { + "id": "child-turn", + "items": "not-an-array", + "status": 42, + }, + }, ) - protocol._thread_id = "thread-1" - protocol._turn_id = "turn-1" - scoped_notifications = ( - "thread/status/changed", - "turn/diff/updated", - "model/verification", - ) - for method in scoped_notifications: - with self.subTest(method=method, scope="empty"): - with self.assertRaises(ProviderError): - protocol._handle_notification(method, {}) - with self.subTest(method=method, scope="wrong"): - with self.assertRaises(ProviderError): - protocol._handle_notification( - method, - {"threadId": "other-thread", "turnId": "other-turn"}, - ) - - def test_item_type_allowlist_matches_isolated_local_tools(self) -> None: + def test_child_turn_rejects_duplicate_completion_and_late_items(self) -> None: protocol = _protocol(QueueTransport([])) - protocol._thread_id = "thread-1" - protocol._turn_id = "turn-1" - allowed = ( - "agentMessage", - "commandExecution", - "contextCompaction", - "fileChange", - "imageView", - "reasoning", - "userMessage", + protocol._thread_id = "main-thread" + protocol._turn_id = "main-turn" + protocol._collab_thread_ids.add("child-1") + protocol._validated_collab_thread_ids.add("child-1") + protocol._collab_turn_usage[("child-1", "child-turn")] = { + "cached_input_tokens": 0, + "input_tokens": 1, + "output_tokens": 0, + "reasoning_output_tokens": 0, + "total_tokens": 1, + } + protocol._handle_notification( + "turn/started", + {"threadId": "child-1", "turn": {"id": "child-turn"}}, ) - self.assertEqual(_ALLOWED_ITEM_TYPES, frozenset(allowed)) - for item_type in allowed: - with self.subTest(item_type=item_type, disposition="allowed"): - protocol._handle_notification( - "item/started", - { - "item": {"id": "item-1", "type": item_type}, - "threadId": "thread-1", - "turnId": "turn-1", - }, - ) + completed = { + "threadId": "child-1", + "turn": { + "id": "child-turn", + "items": [], + "itemsView": "notLoaded", + "status": "completed", + }, + } + protocol._handle_notification("turn/completed", completed) - sentinel = "SENTINEL_MUST_NOT_BE_DISCLOSED" - prohibited = ( - "collabAgentToolCall", - "dynamicToolCall", - "enteredReviewMode", - "exitedReviewMode", - "hookPrompt", - "imageGeneration", - "mcpToolCall", - "plan", - "sleep", - "subAgentActivity", - "webSearch", - sentinel, - ) - for item_type in prohibited: - with self.subTest(item_type=item_type, disposition="prohibited"): - with self.assertRaisesRegex( - ProviderError, "prohibited or unknown Codex item type" - ) as raised: - protocol._handle_notification( - "item/started", - { - "item": {"id": "item-1", "type": item_type}, - "threadId": "thread-1", - "turnId": "turn-1", - }, - ) - self.assertNotIn(sentinel, str(raised.exception)) + with self.assertRaisesRegex(ProviderError, "more than once"): + protocol._handle_notification( + "turn/started", + {"threadId": "child-1", "turn": {"id": "child-turn"}}, + ) + with self.assertRaisesRegex(ProviderError, "unknown child turn"): + protocol._handle_notification("turn/completed", completed) + with self.assertRaisesRegex(ProviderError, "unknown turn"): + protocol._handle_notification( + "item/completed", + { + "completedAtMs": 2, + "item": { + "id": "message-1", + "text": "late", + "type": "agentMessage", + }, + "threadId": "child-1", + "turnId": "child-turn", + }, + ) def test_thread_instruction_runtime_root_and_source_drift_fail_closed(self) -> None: class DriftingTransport(ScriptedTransport): @@ -2180,9 +4205,12 @@ def holder() -> None: self.assertTrue(entered.wait(1)) try: other = _CleanupPoisonStore(self.root) - with self.assertRaisesRegex(ProviderError, "serialization timed out"): + with self.assertRaisesRegex( + ProviderTimeoutError, "serialization timed out" + ) as caught: with other.lock(time.monotonic() + 0.1): self.fail("independent provider unexpectedly acquired the lock") + self.assertTrue(caught.exception.cleanup_confirmed) finally: release.set() thread.join(2) @@ -2351,7 +4379,7 @@ def test_injected_transport_cannot_produce_result_but_exercises_launch_contract( self.assertEqual( provider.protocol_provenance["schema_sha256"], - "f5e8d20f3a8f9bb5e5b23ab0c5aa6bde7b12e7e0713606c5d0132651a4959d37", + "e554a74bd59d38d16acb1744750b2999156ee3d65d0fe906b22ab52edf17fbbc", ) self.assertEqual(provider.execution_policy.concurrency, "serialized") self.assertFalse(provider.execution_policy.release_authoritative) @@ -2404,6 +4432,231 @@ def factory( provider.run_agent(self.request()) self.assertTrue(self.transport.closed) + def test_timeout_is_confirmed_only_after_cleanup_and_cleanup_failure_wins( + self, + ) -> None: + def exercise(*, fail_workspace_cleanup: bool) -> tuple[list[str], Exception]: + events: list[str] = [] + + def assert_no_active_confirmed_timeout() -> None: + active = sys.exception() + if isinstance(active, ProviderTimeoutError): + self.assertFalse(active.cleanup_confirmed) + + @contextlib.contextmanager + def record_exit(name: str, value: Any) -> Iterator[Any]: + try: + yield value + finally: + assert_no_active_confirmed_timeout() + events.append(name) + + class TimeoutTransport(ScriptedTransport): + def __init__( + self, + command: tuple[str, ...], + cwd: Path, + _environment: dict[str, str], + _unit_name: str, + **_callbacks: Any, + ) -> None: + super().__init__(cwd) + self.evidence.update( + cleanup_confirmed=False, + command_sha256=_command_sha256(command), + kind="systemd-run-user+codex-permission-profile", + launch_confirmed=True, + ) + + def receive(self, _deadline: float) -> bytes: + events.append("protocol_timeout") + raise ProviderTimeoutError("Codex protocol read timed out") + + def close(self) -> None: + assert_no_active_confirmed_timeout() + super().close() + self.evidence["cleanup_confirmed"] = True + events.append("transport_close") + + poison_store = mock.Mock() + poison_store.lock.side_effect = lambda _deadline: record_exit( + "provider_lock_exit", (1, 2) + ) + poison_store.recover.return_value = False + poison_store.arm.side_effect = lambda *_args: events.append("poison_arm") + + def disarm_poison( + _binding: _PoisonBinding, evidence: dict[str, Any] + ) -> None: + assert_no_active_confirmed_timeout() + self.assertIs(evidence["cleanup_confirmed"], True) + events.append("poison_disarm") + + poison_store.disarm.side_effect = disarm_poison + + auth_checks = 0 + + def assert_auth_path_matches_descriptor( + _path: Path, _descriptor: int + ) -> tuple[int, int]: + nonlocal auth_checks + auth_checks += 1 + if auth_checks == 3: + assert_no_active_confirmed_timeout() + events.append("post_auth_identity") + return (3, 4) + + def cleanup_workspace_runtime(runtime: Any) -> None: + assert_no_active_confirmed_timeout() + _cleanup_workspace_runtime(runtime) + events.append("workspace_cleanup") + if fail_workspace_cleanup: + raise ProviderError("injected workspace cleanup failure") + + def remove_invocation_root(path: Path) -> None: + assert_no_active_confirmed_timeout() + _remove_invocation_root(path) + events.append("invocation_root_cleanup") + + provider = CodexAppServerProvider( + self.config(), + transport_factory=self.factory, + auth_path=self.auth, + runtime_root=self.runtime, + validate_lock=False, + ) + self.addCleanup(provider.close) + provider._transport_is_injected = False + with ( + mock.patch( + "skivolve.codex_app_server._ProcessTransport", TimeoutTransport + ), + mock.patch( + "skivolve.codex_app_server._CleanupPoisonStore", + return_value=poison_store, + ), + mock.patch( + "skivolve.codex_app_server._auth_lock", + side_effect=lambda *_args: record_exit("auth_lock_exit", None), + ), + mock.patch( + "skivolve.codex_app_server._held_auth_descriptor", + side_effect=lambda *_args: record_exit("auth_descriptor_exit", 1), + ), + mock.patch( + "skivolve.codex_app_server._assert_auth_path_matches_descriptor", + assert_auth_path_matches_descriptor, + ), + mock.patch( + "skivolve.codex_app_server._cleanup_workspace_runtime", + cleanup_workspace_runtime, + ), + mock.patch( + "skivolve.codex_app_server._remove_invocation_root", + remove_invocation_root, + ), + ): + try: + provider.run_agent(self.request()) + except Exception as exc: + return events, exc + self.fail("timeout transport unexpectedly returned a provider result") + + expected_order = ( + "poison_arm", + "protocol_timeout", + "transport_close", + "poison_disarm", + "workspace_cleanup", + "post_auth_identity", + "invocation_root_cleanup", + "auth_descriptor_exit", + "auth_lock_exit", + "provider_lock_exit", + ) + for fail_workspace_cleanup in (False, True): + with self.subTest(fail_workspace_cleanup=fail_workspace_cleanup): + events, error = exercise(fail_workspace_cleanup=fail_workspace_cleanup) + positions = [events.index(event) for event in expected_order] + self.assertEqual(positions, sorted(positions)) + if fail_workspace_cleanup: + self.assertIs(type(error), ProviderError) + self.assertEqual(str(error), "injected workspace cleanup failure") + else: + self.assertIs(type(error), ProviderTimeoutError) + self.assertTrue(error.cleanup_confirmed) + + def test_workspace_runtime_cleanup_handles_unreadable_directories_safely( + self, + ) -> None: + paths = tuple( + self.workspace / name + for name in ( + ".skill-eval-tmp", + ".skill-eval-cache", + ".skill-eval-home", + ) + ) + runtime = _prepare_workspace_runtime(self.workspace) + unreadable = paths[0] / "d" + nested = unreadable / "nested" + nested.mkdir(parents=True) + (nested / "proof").write_text("temporary", encoding="ascii") + outside = self.workspace / "outside-proof" + outside.write_text("preserve", encoding="ascii") + (paths[0] / "outside-link").symlink_to(outside) + nested.chmod(0) + unreadable.chmod(0) + + try: + _cleanup_workspace_runtime(runtime) + finally: + for directory in (nested, unreadable): + if directory.exists(): + directory.chmod(0o700) + + self.assertTrue(all(not path.exists() for path in paths)) + self.assertEqual(outside.read_text(encoding="ascii"), "preserve") + + def test_workspace_runtime_cleanup_handles_deep_directory_trees(self) -> None: + runtime = _prepare_workspace_runtime(self.workspace) + current = self.workspace / ".skill-eval-tmp" + for _ in range(sys.getrecursionlimit() + 64): + current /= "d" + current.mkdir() + + _cleanup_workspace_runtime(runtime) + + for name in (".skill-eval-tmp", ".skill-eval-cache", ".skill-eval-home"): + self.assertFalse(self.workspace.joinpath(name).exists()) + + def test_workspace_runtime_cleanup_does_not_follow_replaced_root(self) -> None: + runtime = _prepare_workspace_runtime(self.workspace) + target = self.workspace / ".skill-eval-tmp" + displaced = self.workspace / "displaced-runtime" + outside = self.workspace / "outside-runtime" + outside_nested = outside / "nested" + outside_nested.mkdir(parents=True) + outside.chmod(0o755) + outside_nested.chmod(0o755) + target.rename(displaced) + target.symlink_to(outside, target_is_directory=True) + + try: + with self.assertRaisesRegex(ProviderError, "identity changed"): + _cleanup_workspace_runtime(runtime) + self.assertEqual(stat.S_IMODE(outside.stat().st_mode), 0o755) + self.assertEqual(stat.S_IMODE(outside_nested.stat().st_mode), 0o755) + finally: + if target.is_symlink(): + target.unlink() + if displaced.exists(): + displaced.rmdir() + for name in (".skill-eval-cache", ".skill-eval-home"): + path = self.workspace / name + if path.exists(): + path.rmdir() + def test_exact_systemd_app_server_launch_and_cleanup_without_protocol_turn( self, ) -> None: @@ -2785,9 +5038,12 @@ def holder() -> None: thread.start() self.assertTrue(entered.wait(1)) try: - with self.assertRaisesRegex(ProviderError, "serialization timed out"): + with self.assertRaisesRegex( + ProviderTimeoutError, "serialization timed out" + ) as caught: with _auth_lock(self.auth, time.monotonic() + 0.1): self.fail("second auth lock unexpectedly acquired") + self.assertTrue(caught.exception.cleanup_confirmed) finally: release.set() thread.join(2) @@ -2866,13 +5122,13 @@ def test_lock_matches_installed_binary_and_regenerated_protocol_without_model_tu executable = _require_real_codex(self) lock = _load_protocol_lock(LOCK_PATH) validate_codex_protocol_lock(executable, lock) - self.assertEqual(lock.cli_version, "codex-cli 0.144.3") - self.assertEqual(lock.thread_cli_version, "0.144.3") + self.assertEqual(lock.cli_version, "codex-cli 0.146.0") + self.assertEqual(lock.thread_cli_version, "0.146.0") self.assertEqual( lock.protocol_sha256, - "f5e8d20f3a8f9bb5e5b23ab0c5aa6bde7b12e7e0713606c5d0132651a4959d37", + "e554a74bd59d38d16acb1744750b2999156ee3d65d0fe906b22ab52edf17fbbc", ) - self.assertEqual(lock.protocol_canonical_bytes, 308970) + self.assertEqual(lock.protocol_canonical_bytes, 331046) self.assertEqual( hashlib.sha256(LOCK_PATH.read_bytes()).hexdigest(), lock.sha256 )