diff --git a/cc_remote/wrapper/codex_checkpoints.py b/cc_remote/wrapper/codex_checkpoints.py index f4ea022..818ac68 100644 --- a/cc_remote/wrapper/codex_checkpoints.py +++ b/cc_remote/wrapper/codex_checkpoints.py @@ -196,6 +196,81 @@ def _checkpoint_session_key(session_id: str) -> str: ).hexdigest()[:24] +def cleanup_codex_checkpoint_session( + state_dir: Path, + session_id: str, +) -> int: + """Force-remove one session's journals without needing its former cwd.""" + if not isinstance(session_id, str) or not session_id.strip(): + raise ValueError("session_id is required") + root = ( + Path(state_dir).expanduser().resolve(strict=False) + / "codex-checkpoints" + ) + try: + repositories = [ + entry + for entry in os.scandir(root) + if entry.is_dir(follow_symlinks=False) + ] + except FileNotFoundError: + return 0 + except OSError as exc: + raise CheckpointError( + "Unable to inspect Codex checkpoint journals" + ) from exc + + session_key = _checkpoint_session_key(session_id) + removed = 0 + for repository in repositories: + candidate = Path(repository.path) / session_key + try: + mode = candidate.lstat().st_mode + except FileNotFoundError: + continue + except OSError as exc: + raise CheckpointError( + "Unable to inspect Codex checkpoint journal" + ) from exc + if stat.S_ISLNK(mode) or not stat.S_ISDIR(mode): + raise CheckpointError( + "Checkpoint session path is not a directory" + ) + + lock_fd: Optional[int] = None + tombstone: Optional[Path] = None + try: + lock_flags = os.O_RDWR | os.O_CREAT + lock_flags |= getattr(os, "O_NOFOLLOW", 0) + lock_fd = os.open(candidate / "journal.lock", lock_flags, 0o600) + fcntl.flock(lock_fd, fcntl.LOCK_EX) + tombstone = candidate.with_name( + f".{candidate.name}.delete-{uuid.uuid4().hex}" + ) + try: + os.replace(candidate, tombstone) + except FileNotFoundError: + tombstone = None + continue + except OSError as exc: + raise CheckpointError( + "Unable to retire Codex checkpoint journal" + ) from exc + finally: + if lock_fd is not None: + fcntl.flock(lock_fd, fcntl.LOCK_UN) + os.close(lock_fd) + if tombstone is not None: + try: + shutil.rmtree(tombstone, ignore_errors=False) + except OSError as exc: + raise CheckpointError( + "Unable to remove Codex checkpoint journal" + ) from exc + removed += 1 + return removed + + def migrate_codex_checkpoint_profiles( state_dir: Path, transform: Callable[[str], str], diff --git a/cc_remote/wrapper/codex_forks.py b/cc_remote/wrapper/codex_forks.py index c53e39c..b8443b1 100644 --- a/cc_remote/wrapper/codex_forks.py +++ b/cc_remote/wrapper/codex_forks.py @@ -143,6 +143,13 @@ def _validate_aliases(entries: OrderedDict[str, dict[str, Any]]) -> None: ): raise ValueError( "fork alias name state differs from its canonical root") + if ( + entry.get("title") != canonical.get("title") + or entry.get("title_updated_at") + != canonical.get("title_updated_at") + ): + raise ValueError( + "fork alias title differs from its canonical root") compatible = { "alias": {"intent", "submitted", "uncertain"}, "complete": {"complete"}, @@ -230,6 +237,18 @@ def _validate_entry(request_id: Any, entry: Any) -> None: name_finalized = entry.get("name_finalized") if name_finalized is not None and not isinstance(name_finalized, bool): raise ValueError("invalid fork name finalization state") + title = entry.get("title") + if title is not None and ( + not isinstance(title, str) or not title or len(title) > 200 + ): + raise ValueError("invalid fork title") + title_updated_at = entry.get("title_updated_at") + if title_updated_at is not None and ( + title is None + or isinstance(title_updated_at, bool) + or not isinstance(title_updated_at, (int, float)) + ): + raise ValueError("invalid fork title timestamp") def begin( self, @@ -448,6 +467,58 @@ def get(self, request_id: str) -> Optional[dict[str, Any]]: entry = self.entries.get(request_id) return dict(entry) if entry is not None else None + def completed_results(self, limit: int) -> list[dict[str, Any]]: + """Return newest unique completed forks for catalog recovery.""" + if isinstance(limit, bool) or not isinstance(limit, int) or limit < 1: + raise ForkJournalError("invalid completed fork result limit") + with self._lock: + results: list[dict[str, Any]] = [] + seen_sources: set[str] = set() + for entry in reversed(tuple(self.entries.values())): + source = entry.get("thread_source") + if ( + entry.get("status") != "complete" + or not isinstance(source, str) + or source in seen_sources + ): + continue + seen_sources.add(source) + results.append(dict(entry)) + if len(results) >= limit: + break + return results + + def set_title(self, session_id: str, title: str) -> bool: + """Persist a renamed fork while its native catalog row is absent.""" + if ( + not isinstance(session_id, str) + or not _SAFE_ID.fullmatch(session_id) + ): + raise ForkJournalError("invalid forked session id") + if not isinstance(title, str) or not title or len(title) > 200: + raise ForkJournalError("invalid fork title") + with self._lock: + updated = OrderedDict(self.entries) + changed = False + updated_at = time.time() + for request_id, entry in tuple(updated.items()): + if ( + entry.get("status") != "complete" + or entry.get("session_id") != session_id + ): + continue + renamed = dict(entry) + renamed["title"] = title + renamed["title_updated_at"] = updated_at + updated[request_id] = renamed + changed = True + if not changed: + return False + self._validate_aliases(updated) + self._persist(updated) + self.entries = updated + return True + def _set_status( self, request_id: str, status: str, **fields: Any, ) -> dict[str, Any]: diff --git a/cc_remote/wrapper/codex_handle.py b/cc_remote/wrapper/codex_handle.py index fcdcc5f..6b5c058 100644 --- a/cc_remote/wrapper/codex_handle.py +++ b/cc_remote/wrapper/codex_handle.py @@ -84,6 +84,11 @@ _STATUS_USAGE_BUCKET_SCAN_MAX = 4096 _RUNTIME_EVENT_PENDING_MAX = 32 _RUNTIME_EVENT_SEEN_MAX = 128 +_THREAD_DELETE_NOTIFY_MAX = 512 +_THREAD_DELETE_NOTIFY_TIMEOUT = 1.0 +_THREAD_DELETE_LIST_PAGE_SIZE = 100 +_THREAD_DELETE_LIST_MAX_PAGES = 20 +_THREAD_DELETE_LIST_MAX_IDS = 4096 _PENDING_STEER_USER_IDENTITIES_MAX = 512 _ACTIVE_STREAM_TURN_IDS_MAX = 8 _NOTICE_MESSAGE_MAX = 2 * 1024 @@ -1320,6 +1325,7 @@ def __init__(self, cfg, cwd: Optional[str] = None, self.daemon_mode, codex_home=self.codex_home) ) self._using_daemon_proxy = False + self._control_only_connection = False # Once a Code session has joined the official shared app-server, a # transport interruption must not silently turn it into a private stdio # session. Keep this affinity across proxy reconnects so Machine can @@ -1342,6 +1348,10 @@ def __init__(self, cfg, cwd: Optional[str] = None, self._reader: Optional[asyncio.Task] = None self._stderr_task: Optional[asyncio.Task] = None self._thread_settings_updated = asyncio.Event() + self._thread_delete_target: Optional[str] = None + self._thread_deleted_ids: Optional[list[str]] = None + self.thread_delete_notifications_overflowed = False + self._thread_delete_done = asyncio.Event() # Human approval can take minutes. It must not block the sole stdout # reader, which still has to consume turn/interrupt and other RPC replies. # Keep detached request handlers generation-owned and cancel them on @@ -1692,16 +1702,29 @@ async def connect( fork: bool = False, preserve_controls: bool = False, preserve_permission_profile: bool = True, + control_only: bool = False, ) -> None: + if control_only and (resume_id is not None or fork or self.work_mode): + raise ValueError( + "control-only Codex connections cannot bind a thread" + ) if self.proc is not None: await self.disconnect() + # Arm this before the reader starts. A shared daemon can publish a + # sibling thread/started during initialize, and a control connection + # must never adopt that unrelated thread while it is still unbound. + self._control_only_connection = control_only self._shared_resume_binding_thread_id = None self._cwd = cwd or self._cwd or getattr(self.cfg, "cc_cwd", None) or os.getcwd() # version-probes subprocesses on first call; keep it off the event loop. codex_bin = await asyncio.to_thread(_resolve_codex_bin) private_core = None http_only_resume = False - if not self.work_mode and not self._daemon_proxy_established: + if ( + not self.work_mode + and not self._daemon_proxy_established + and not control_only + ): http_only_resume = await asyncio.to_thread( _oversized_desktop_openai_resume_requires_http, resume_id, @@ -1787,6 +1810,12 @@ async def connect( # stdio. Leave the handle disconnected and let Machine retry the # shared proxy instead of manufacturing a false external-CLI lock. attempts = [(proxy_argv, True)] + if control_only: + if proxy_argv is None: + raise RuntimeError( + "shared Codex app-server proxy is unavailable" + ) + attempts = [(proxy_argv, True)] initialized: Any = None for argv, daemon_proxy in attempts: self._shared_resume_binding_thread_id = ( @@ -1812,7 +1841,11 @@ async def connect( if not daemon_proxy: raise self.daemon_manager.invalidate() - if strict_shared or self._daemon_proxy_established: + if ( + strict_shared + or self._daemon_proxy_established + or control_only + ): log.warning( "Codex shared daemon proxy unavailable; reconnect required", error_type=type(exc).__name__, @@ -1825,6 +1858,9 @@ async def connect( else: # pragma: no cover - the attempt list is never empty raise RuntimeError("unable to start Codex app-server transport") try: + if control_only: + log.info("codex control connection established", cwd=self._cwd) + return if self.work_mode: # Inspect the effective native runtime rather than guessing at # user-configured skill and MCP names. Failure is fatal: silently @@ -3342,6 +3378,178 @@ def reconcile_no_active_turn( self._spontaneous_turn_id = None return True + async def delete_thread( + self, expected_thread_id: Optional[str] = None, + ) -> tuple[str, ...]: + """Delete the loaded thread through its authoritative app-server.""" + loaded_thread_id = self.thread_id + thread_id = expected_thread_id or loaded_thread_id + if not thread_id: + raise RuntimeError("connect() or an explicit thread id is required") + if loaded_thread_id is None and not ( + self._control_only_connection and self.using_daemon_proxy + ): + raise RuntimeError( + "unloaded Codex deletion requires a live control connection" + ) + if ( + loaded_thread_id is not None + and expected_thread_id is not None + and expected_thread_id != loaded_thread_id + ): + raise ValueError("loaded Codex thread does not match delete target") + if self.turn_active or self.turn_start_pending: + raise RuntimeError("Codex turn is active") + if self._thread_delete_target is not None: + raise RuntimeError("Codex thread deletion is already active") + self._thread_delete_target = thread_id + self._thread_deleted_ids = [] + self.thread_delete_notifications_overflowed = False + self._thread_delete_done.clear() + try: + await self._request( + "thread/delete", + {"threadId": thread_id}, + ) + if self._reader is not None and not self._reader.done(): + loop = asyncio.get_running_loop() + notification_deadline = ( + loop.time() + _THREAD_DELETE_NOTIFY_TIMEOUT + ) + try: + await asyncio.wait_for( + self._thread_delete_done.wait(), + timeout=_THREAD_DELETE_NOTIFY_TIMEOUT, + ) + except asyncio.TimeoutError: + log.warning( + "Codex thread deletion notification timed out", + thread_id=thread_id, + ) + else: + # The current app-server emits the root last, but the wire + # contract promises one notification per deleted thread, + # not that the root is the final notification. Keep the + # collector alive for the original bounded window so a + # root-first batch cannot strand local descendants. + remaining = notification_deadline - loop.time() + if remaining > 0: + await asyncio.sleep(remaining) + deleted_ids = list(self._thread_deleted_ids) + finally: + self._thread_delete_target = None + self._thread_deleted_ids = None + self._thread_delete_done.clear() + if thread_id not in deleted_ids: + deleted_ids.append(thread_id) + if self.thread_id == thread_id: + self.thread_id = None + self._shared_resume_binding_thread_id = None + return tuple(deleted_ids) + + async def read_thread_parent(self, thread_id: str) -> Optional[str]: + """Return one thread's authoritative native fork parent.""" + if ( + not isinstance(thread_id, str) + or not _STATUS_WIRE_ID.fullmatch(thread_id) + ): + raise ValueError("invalid Codex thread id") + response = await self._request( + "thread/read", + {"threadId": thread_id, "includeTurns": False}, + ) + thread = ( + response.get("thread") + if isinstance(response, dict) else None + ) + if not isinstance(thread, dict) or thread.get("id") != thread_id: + raise RuntimeError("Codex thread/read returned another thread") + parent = thread.get("forkedFromId") + if parent is None: + return None + if ( + not isinstance(parent, str) + or not _STATUS_WIRE_ID.fullmatch(parent) + ): + raise RuntimeError("Codex thread/read returned an invalid parent") + return parent + + async def list_thread_delete_candidates( + self, + ) -> tuple[tuple[str, bool], ...]: + """List every bounded native thread before a recursive delete.""" + candidates: dict[str, bool] = {} + for archived in (False, True): + cursor: Optional[str] = None + seen_cursors: set[str] = set() + for _page in range(_THREAD_DELETE_LIST_MAX_PAGES): + params: dict[str, Any] = { + "archived": archived, + "limit": _THREAD_DELETE_LIST_PAGE_SIZE, + "sortKey": "updated_at", + "sortDirection": "desc", + } + if cursor is not None: + params["cursor"] = cursor + response = await self._request("thread/list", params) + rows = ( + response.get("data") + if isinstance(response, dict) + else None + ) + if not isinstance(rows, list): + raise RuntimeError( + "Codex thread/list returned an invalid response" + ) + if len(rows) > _THREAD_DELETE_LIST_PAGE_SIZE: + raise RuntimeError( + "Codex thread/list exceeded its requested page size" + ) + for thread in rows: + thread_id = ( + thread.get("id") + if isinstance(thread, dict) + else None + ) + if ( + not isinstance(thread_id, str) + or not _STATUS_WIRE_ID.fullmatch(thread_id) + ): + raise RuntimeError( + "Codex thread/list returned an invalid thread" + ) + status = thread.get("status") + status_type = ( + status.get("type") + if isinstance(status, dict) + else status + ) + candidates[thread_id] = bool( + candidates.get(thread_id) + or status_type == "active" + ) + if len(candidates) > _THREAD_DELETE_LIST_MAX_IDS: + raise RuntimeError( + "Codex deletion catalog exceeds the safety limit" + ) + next_cursor = response.get("nextCursor") + if next_cursor in (None, ""): + break + if ( + not isinstance(next_cursor, str) + or next_cursor in seen_cursors + ): + raise RuntimeError( + "Codex thread/list returned an invalid cursor" + ) + seen_cursors.add(next_cursor) + cursor = next_cursor + else: + raise RuntimeError( + "Codex deletion catalog exceeds the page limit" + ) + return tuple(candidates.items()) + async def disconnect(self) -> None: self._http_provider_repair_stop.set() proc = self.proc @@ -3404,6 +3612,7 @@ def stop(sig: signal.Signals, *, force: bool = False) -> None: if process_group is not None: stop(signal.SIGKILL, force=True) self._using_daemon_proxy = False + self._control_only_connection = False self._shared_resume_binding_thread_id = None self._proxy_read_buffer.clear() self._proxy_close_sent = False @@ -5366,6 +5575,27 @@ def _install_turn_start_response_boundary( control_turn_id, ) + def _capture_thread_deleted_notification(self, message: dict) -> bool: + """Collect the target and its descendants until the target arrives.""" + if message.get("method") != "thread/deleted": + return False + thread_id = _notification_thread_id(message) + deleted_ids = self._thread_deleted_ids + if deleted_ids is None or self._thread_delete_target is None: + return False + if ( + isinstance(thread_id, str) + and _STATUS_WIRE_ID.fullmatch(thread_id) + ): + if thread_id not in deleted_ids: + if len(deleted_ids) < _THREAD_DELETE_NOTIFY_MAX: + deleted_ids.append(thread_id) + else: + self.thread_delete_notifications_overflowed = True + if thread_id == self._thread_delete_target: + self._thread_delete_done.set() + return True + async def _dispatch(self, m: dict, raw_size: Optional[int] = None) -> None: has_id = "id" in m has_method = "method" in m @@ -5384,6 +5614,15 @@ async def _dispatch(self, m: dict, raw_size: Optional[int] = None) -> None: return if has_id and has_method: # server -> client request method = m.get("method") + if self._control_only_connection: + # This proxy has no thread or interaction owner. A shared + # daemon can broadcast a sibling's approval request here; + # silence leaves it available to that thread's real client. + log.warning( + "Codex control connection left server request pending", + method=method, + ) + return request_id = _server_request_key(m.get("id")) missing_callback = isinstance(method, str) and (( method in _NEW_APPROVAL_METHODS | _LEGACY_APPROVAL_METHODS @@ -5440,6 +5679,13 @@ async def _dispatch(self, m: dict, raw_size: Optional[int] = None) -> None: return # notification method = m.get("method") + if self._capture_thread_deleted_notification(m): + return + if self._control_only_connection: + # Responses are handled above and thread/deleted is the only + # notification used by this connection. In particular, never let + # shared-daemon sibling lifecycle bind or mutate this handle. + return if method == "thread/started" and self._using_daemon_proxy: target_thread_id = _notification_thread_id(m) params = m.get("params") diff --git a/cc_remote/wrapper/machine.py b/cc_remote/wrapper/machine.py index 144a09e..ff7834a 100644 --- a/cc_remote/wrapper/machine.py +++ b/cc_remote/wrapper/machine.py @@ -54,7 +54,7 @@ import time import unicodedata from collections import OrderedDict -from contextlib import contextmanager +from contextlib import AsyncExitStack, contextmanager from dataclasses import dataclass from pathlib import Path from uuid import uuid4 @@ -267,11 +267,12 @@ ) from cc_remote.wrapper.codex_checkpoints import ( CheckpointConflict, CheckpointError, CodexCheckpointJournal, - NotGitWorkspaceError, migrate_codex_checkpoint_profiles, + NotGitWorkspaceError, cleanup_codex_checkpoint_session, + migrate_codex_checkpoint_profiles, ) from cc_remote.wrapper.codex_forks import ( CodexForkJournal, ForkJournalError, find_rollout_fork, - fork_thread_source, + fork_thread_source, rollout_fork_meta, ) from cc_remote.wrapper.claude_forks import ( ClaudeForkJournal, ClaudeForkJournalError, claude_fork_marker, @@ -333,6 +334,7 @@ def _normalize_claude_new_session_model(model: Optional[str]) -> Optional[str]: switch unless it prevents completion. """ _CODEX_DAEMON_UNMARKED_EPOCH = "unmarked" +_CODEX_DELETE_ANCESTRY_LIMIT = 256 def _codex_fast_on(value: Optional[str]) -> bool: @@ -1285,6 +1287,7 @@ class WrapperMachine: # full and its catalog strings are at their protocol limits. CODEX_SESSION_LIST_MAX_ROWS = 400 CODEX_SESSION_LIST_MAX_BYTES = 8 * 1024 * 1024 + CODEX_FORK_CATALOG_MAX_SCAN_FILES = 20_000 PREVIEW_ASSET_MEDIA_TYPES = { ".png": "image/png", ".jpg": "image/jpeg", @@ -6061,6 +6064,17 @@ async def _enqueue_deferred_query( accepted = False async with ctx.emit_lock: async with ctx.queued_query_lock: + if not self._is_resident_context(ctx): + error = Error( + code=ERR_NOT_RUNNING, + message="该会话已被删除，本次排队未提交。", + msg_id=cmd.msg_id, + request_id=getattr(cmd, "cmd_id", None), + to=getattr(cmd, "client_id", None), + sid=self._ctx_wire_sid(ctx), + ) + await self.transport.send(error) + return error duplicate = next(( queued for queued in ctx.queued_queries if queued.msg_id == cmd.msg_id @@ -7581,16 +7595,32 @@ def _codex_sidebar_watch_state(watch: Optional[dict]) -> Optional[State]: return None return "running" if watch.get("active_external_turns") else None - def _codex_own_processes(self) -> set[ProcessIdentity]: + @staticmethod + def _codex_handle_process_identity( + sdk: CodexHandle, + ) -> ProcessIdentity | None: + proc = getattr(sdk, "proc", None) + pid = getattr(proc, "pid", None) + if ( + not isinstance(pid, int) + or getattr(proc, "returncode", None) is not None + ): + return None + return process_identity(pid, parent_pid=os.getpid()) + + def _codex_own_processes( + self, + extra_handles: tuple[CodexHandle, ...] = (), + ) -> set[ProcessIdentity]: own: set[ProcessIdentity] = set() - for ctx in list(self.sessions.values()): - if ctx.engine != "codex": - continue - proc = getattr(ctx.sdk, "proc", None) - pid = getattr(proc, "pid", None) - if not isinstance(pid, int) or getattr(proc, "returncode", None) is not None: - continue - identity = process_identity(pid, parent_pid=os.getpid()) + handles = [ + ctx.sdk + for ctx in list(self.sessions.values()) + if ctx.engine == "codex" + ] + handles.extend(extra_handles) + for sdk in handles: + identity = self._codex_handle_process_identity(sdk) if identity is not None: own.add(identity) return own @@ -7601,8 +7631,13 @@ def _codex_watch_paths(self, only_sid: Optional[str] = None) -> dict[str, str]: if w.get("engine") == "codex" and (only_sid is None or sid == only_sid) } - async def _probe_codex_holders(self, paths: dict[str, str]): - initial_own = self._codex_own_processes() + async def _probe_codex_holders( + self, + paths: dict[str, str], + *, + extra_handles: tuple[CodexHandle, ...] = (), + ): + initial_own = self._codex_own_processes(extra_handles) grouped: dict[str, dict[str, str]] = {} for sid, path in paths.items(): watch = self._watch.get(sid) or {} @@ -7784,7 +7819,7 @@ def wire_bucket( # A reconnect can replace an app-server while /proc is being scanned. # Remove both the initial and current exact child identities before the # result is allowed to influence ownership. - current_own = self._codex_own_processes() + current_own = self._codex_own_processes(extra_handles) for holders in scan.holders.values(): holders.difference_update(initial_own) holders.difference_update(current_own) @@ -7926,13 +7961,24 @@ def _codex_holder_sets( writers = raw.difference(ignored) return writers.difference(passive), writers, private.intersection(writers) - async def _prime_codex_ownership(self, sid: str) -> bool: + async def _prime_codex_ownership( + self, + sid: str, + *, + extra_handles: tuple[CodexHandle, ...] = (), + ) -> bool: """Atomically consume growth and refresh one owner before History/Query.""" w = self._watch.get(sid) if not w or w.get("engine") != "codex": return False async with self._codex_watch_lock: - scan = await self._probe_codex_holders({sid: w["path"]}) + if extra_handles: + scan = await self._probe_codex_holders( + {sid: w["path"]}, + extra_handles=extra_handles, + ) + else: + scan = await self._probe_codex_holders({sid: w["path"]}) holders, writers, private_holders = self._codex_holder_sets( w, scan, sid) scan_complete = self._codex_scan_complete_for_sid(scan, sid) @@ -11154,6 +11200,8 @@ async def _handle_query(self, cmd): if getattr(cmd, "delivery", "immediate") != "immediate": return await self._enqueue_deferred_query(ctx, cmd) async with ctx.query_lock: + if not self._is_resident_context(ctx): + return await self._missing_session_error(cmd, "发送消息") return await self._handle_immediate_query(ctx, cmd) async def _handle_immediate_query(self, ctx: SessionContext, cmd): @@ -16947,12 +16995,183 @@ async def _refresh_codex_session_catalog(self) -> list[dict]: if row.get("codex_profile_id") in failed_profile_ids ] raw = self._bounded_codex_profile_catalog([*raw, *stale_rows]) + raw = await asyncio.to_thread( + self._overlay_completed_codex_forks, raw) self._codex_session_profile_errors = profile_errors self._codex_session_list_cache = ( time.monotonic(), raw, profile_errors, ) return raw + def _overlay_completed_codex_forks( + self, raw: list[dict], + ) -> list[dict]: + """Keep fork-only children visible until app-server indexes a turn.""" + existing = { + row.get("session_id") + for row in raw + if isinstance(row.get("session_id"), str) + } + candidates: list[ + tuple[dict, CodexProfile, str, str, str] + ] = [] + for entry in self._codex_forks.completed_results( + self.CODEX_SESSION_LIST_MAX_ROWS, + ): + child_wire_sid = entry.get("session_id") + parent_wire_sid = entry.get("parent_session_id") + if ( + not isinstance(child_wire_sid, str) + or child_wire_sid in existing + or not isinstance(parent_wire_sid, str) + ): + continue + try: + child_profile, child_native_sid = self._codex_target( + child_wire_sid) + parent_profile, parent_native_sid = self._codex_target( + parent_wire_sid) + except (RuntimeError, ValueError): + continue + if child_profile.id != parent_profile.id: + continue + candidates.append(( + entry, + child_profile, + child_native_sid, + parent_native_sid, + parent_wire_sid, + )) + + if not candidates: + return raw + paths = self._find_completed_codex_fork_rollouts(candidates) + parent_rows = { + row.get("session_id"): row + for row in raw + if isinstance(row.get("session_id"), str) + } + overlays: list[dict] = [] + for ( + entry, + profile, + child_native_sid, + parent_native_sid, + parent_wire_sid, + ) in candidates: + child_wire_sid = entry["session_id"] + located = paths.get((profile.id, child_native_sid)) + if located is None: + continue + path, archived = located + meta = rollout_fork_meta(path) + if ( + meta is None + or meta.get("session_id") != child_native_sid + or meta.get("forked_from_id") != parent_native_sid + or meta.get("thread_source") != entry.get("thread_source") + ): + continue + try: + modified = max( + float(entry.get("created_at", 0)), + float(entry.get("title_updated_at", 0)), + os.stat(path).st_mtime, + ) + except (OSError, TypeError, ValueError): + continue + parent = parent_rows.get(parent_wire_sid, {}) + inherited_title = ( + parent.get("summary") or parent.get("first_prompt") + ) + persisted_title = entry.get("title") + summary = ( + str(persisted_title).strip() if persisted_title else ( + f"{str(inherited_title).strip()} (fork)" + if inherited_title else "Forked session" + ) + )[:500] + overlays.append({ + "session_id": child_wire_sid, + "native_session_id": child_native_sid, + "codex_profile_id": profile.id, + "codex_profile_label": profile.label, + "forked_from_id": parent_wire_sid, + "summary": summary, + "first_prompt": parent.get("first_prompt"), + "cwd": entry.get("cwd") or meta.get("cwd"), + "last_modified": str(modified), + "git_branch": parent.get("git_branch"), + "tag": "archived" if archived else None, + "status": "notLoaded", + }) + if not overlays: + return raw + return self._bounded_codex_profile_catalog([*raw, *overlays]) + + def _find_completed_codex_fork_rollouts( + self, + candidates: list[tuple[dict, CodexProfile, str, str, str]], + ) -> dict[tuple[str, str], tuple[str, bool]]: + """Locate many journaled children with one bounded scan per profile.""" + pending: dict[str, dict[str, None]] = {} + profiles: dict[str, CodexProfile] = {} + for _entry, profile, child_sid, _parent_sid, _parent_wire in candidates: + pending.setdefault(profile.id, {})[child_sid] = None + profiles[profile.id] = profile + + found: dict[tuple[str, str], tuple[str, bool]] = {} + scanned = 0 + exhausted = False + for profile_id, wanted in pending.items(): + profile = profiles[profile_id] + remaining = set(wanted) + for root_name, archived in ( + ("sessions", False), + ("archived_sessions", True), + ): + root = os.path.realpath(profile.home / root_name) + if not os.path.isdir(root): + continue + for directory, directories, files in os.walk(root): + directories.sort(reverse=True) + files.sort(reverse=True) + for filename in files: + if not filename.endswith(".jsonl"): + continue + scanned += 1 + if scanned > self.CODEX_FORK_CATALOG_MAX_SCAN_FILES: + exhausted = True + break + child_sid = next(( + sid for sid in remaining + if filename.endswith(f"-{sid}.jsonl") + ), None) + if child_sid is None: + continue + path = os.path.realpath( + os.path.join(directory, filename)) + try: + inside_root = os.path.commonpath( + (root, path)) == root + except ValueError: + inside_root = False + if not inside_root: + continue + found[(profile_id, child_sid)] = (path, archived) + remaining.remove(child_sid) + if exhausted or not remaining: + break + if exhausted or not remaining: + break + if exhausted: + log.warning( + "Codex fork catalog rollout scan reached its bound", + max_files=self.CODEX_FORK_CATALOG_MAX_SCAN_FILES, + ) + break + return found + def _bounded_codex_profile_catalog( self, candidates: list[dict], ) -> list[dict]: @@ -17909,6 +18128,8 @@ async def _handle_rename_session(self, cmd) -> None: await self._codex_rpc_for_wire(sid, "thread/name/set", { "threadId": sid, "name": cmd.title, }) + await asyncio.to_thread( + self._codex_forks.set_title, sid, cmd.title) if work_record is not None: await asyncio.to_thread( self._work.for_engine(engine).update_title, @@ -18125,11 +18346,7 @@ async def _handle_delete_work_session(self, cmd): self._codex_rollout_for_wire, sid) except (OSError, ValueError): codex_alias_path = None - if ctx is not None and ( - ctx.state != "idle" - or ctx.queued_queries - or self._query_queue_task_active(ctx) - ): + if ctx is not None and self._session_delete_busy(ctx): error = Error( code=ERR_BUSY, message="Work 会话仍在运行或有排队消息，请先停止并取消排队后再删除", @@ -18138,17 +18355,40 @@ async def _handle_delete_work_session(self, cmd): ) await self.transport.send(error) return error + codex_deleted = False if ctx is not None: - await ctx.sdk.disconnect() - self.sessions.pop(ctx.key or sid, None) - self._purge_preview_image_snapshots(ctx.preview_snapshot_token) + if engine == "codex": + delete_result = await self._delete_loaded_codex_thread( + cmd, + sid, + native_sid, + ctx, + rollout_path=codex_alias_path, + ) + if isinstance(delete_result, Error): + return delete_result + codex_deleted = True + else: + await ctx.sdk.disconnect() + self.sessions.pop(ctx.key or sid, None) + self._purge_preview_image_snapshots( + ctx.preview_snapshot_token, + ) try: if engine == "codex": - await self._codex_rpc_for_wire( - sid, "thread/delete", {"threadId": sid}) + if not codex_deleted: + await self._codex_rpc_for_wire( + sid, + "thread/delete", + {"threadId": sid}, + ) + self._invalidate_codex_session_catalog() else: await asyncio.to_thread( - delete_session, sid, directory=record.cwd) + delete_session, + sid, + directory=record.cwd, + ) await asyncio.to_thread( store.delete, native_sid, @@ -18192,201 +18432,1017 @@ async def _handle_delete_work_session(self, cmd): await self._handle_list_sessions(cmd) log.info("Work session deleted", engine=engine, session_id=sid) - async def _handle_delete_session(self, cmd): - """Delete one native session without confusing Code and Work roots.""" - if getattr(cmd, "space", "code") == "work": - return await self._handle_delete_work_session(cmd) - sid = self._resolve_session_alias(cmd.session_id) or cmd.session_id - requested_engine = getattr(cmd, "engine", "claude") - is_codex = await self._is_codex_session(sid) - engine = "codex" if is_codex else "claude" - if engine != requested_engine: - error = Error( - code=ERR_AUTH, - message="会话不属于请求的引擎", - sid=sid, - to=getattr(cmd, "client_id", None), - ) - await self.transport.send(error) - return error - try: - native_sid, work_profile_id = self._work_session_identity(engine, sid) - except ValueError: - native_sid, work_profile_id = sid, None - work_record = await asyncio.to_thread( - self._work.for_engine(engine).get_by_session, - native_sid, - codex_profile_id=work_profile_id, + def _session_delete_busy(self, ctx: SessionContext) -> bool: + active = any( + task is not None and not task.done() + for task in (ctx.turn_task, ctx.codex_spontaneous_task) ) - if work_record is not None: - error = Error( - code=ERR_AUTH, - message="Work 会话必须从 Work 空间删除", - sid=sid, - to=getattr(cmd, "client_id", None), - ) - await self.transport.send(error) - return error - ctx = self._ctx_for(sid) - if ctx is not None and ( + return bool( ctx.state != "idle" + or active or ctx.queued_queries + or ctx.queued_query_starting_msg_id or self._query_queue_task_active(ctx) + or getattr(ctx.sdk, "turn_active", False) + or getattr(ctx.sdk, "turn_start_pending", False) + ) + + def _codex_context_native_thread_id( + self, + ctx: SessionContext, + profile: CodexProfile, + ) -> str | None: + """Resolve a resident context to its native app-server thread id.""" + for candidate in ( + ctx.session_id, + ctx.btw_real_id, + getattr(ctx.sdk, "thread_id", None), ): - error = Error( - code=ERR_BUSY, - message="会话仍在运行或有排队消息，请先停止并取消排队后再删除", - sid=sid, - to=getattr(cmd, "client_id", None), - ) - await self.transport.send(error) - return error - cwd = ctx.cwd if ctx is not None else None - codex_alias_path = None - if engine == "codex": - try: - codex_alias_path = await asyncio.to_thread( - self._codex_rollout_for_wire, sid) - except (OSError, ValueError): - codex_alias_path = None - checkpoint_cleanup_journal = None - if engine == "codex" and cwd is None: + if not isinstance(candidate, str): + continue try: - cwd = await asyncio.to_thread( - self._codex_cwd_for_wire, sid) + self._codex_wire_sid(profile, candidate) except ValueError: - cwd = None - if engine == "claude" and cwd is None: - info = await asyncio.to_thread(get_session_info, sid) - cwd = info.cwd if info is not None else None - # A metadata-only Claude transcript has no cwd but is still a real - # exact-SID file in the SDK catalog. delete_session(directory=None) - # safely searches all project roots for that UUID and deletes only - # the matching transcript. Preserve the old not-found rejection - # when no exact transcript exists. - if not cwd and transcript_path(sid) is None: - error = Error( - code=ERR_NOT_RUNNING, - message="Claude 会话不存在", - sid=sid, - to=getattr(cmd, "client_id", None), - ) - await self.transport.send(error) - return error - if engine == "codex" and cwd: - existing_journal = ctx.codex_checkpoint if ctx is not None else None - if existing_journal is not False: - try: - checkpoint_cleanup_journal = existing_journal or ( - await asyncio.to_thread( - CodexCheckpointJournal, - cwd, - Path(self.cfg.state_dir), - sid, - profile_revision=self._codex_profile_revision, - ) - ) - except (CheckpointError, NotGitWorkspaceError) as exc: - log.warning( - "Codex checkpoint journal could not be opened for delete cleanup", - session_id=sid, - error_type=type(exc).__name__, - ) - if ctx is not None: - try: - await ctx.sdk.disconnect() - except Exception: - log.exception( - "session disconnect before delete failed", - engine=engine, - session_id=sid, - ) - error = Error( - code=ERR_INTERNAL, - message="无法安全停止会话，未执行删除", - sid=sid, - to=getattr(cmd, "client_id", None), - ) - await self.transport.send(error) - return error - self.sessions.pop(ctx.key or sid, None) - self._purge_preview_image_snapshots(ctx.preview_snapshot_token) - try: - if engine == "codex": + continue + return candidate + return None + + async def _codex_thread_descends_from( + self, + sdk: CodexHandle, + thread_id: str, + ancestor_id: str, + parent_cache: dict[str, str | None], + ) -> bool: + """Follow exact loaded-thread metadata to one requested ancestor.""" + current = thread_id + seen: set[str] = set() + for _depth in range(_CODEX_DELETE_ANCESTRY_LIMIT): + if current == ancestor_id: + return True + if current in seen: + raise RuntimeError("Codex fork ancestry contains a cycle") + seen.add(current) + if current not in parent_cache: try: - await self._codex_rpc_for_wire( - sid, "thread/delete", {"threadId": sid}) - except CodexRpcOutcomeUnknown: - profile, native_sid = self._codex_target(sid) - home = self._codex_home(profile) - remaining = ( - await list_codex_sessions(200) - if home is None else - await list_codex_sessions(200, codex_home=home) + parent_cache[current] = ( + await sdk.read_thread_parent(current) ) - if any( - row.get("session_id") == native_sid - for row in remaining - ): - raise - self._invalidate_codex_session_catalog() - else: - await asyncio.to_thread(delete_session, sid, directory=cwd) - except Exception: - log.exception("Code session deletion failed", engine=engine, session_id=sid) - error = Error( - code=ERR_INTERNAL, - message="会话删除失败，请刷新后重试", - sid=sid, - to=getattr(cmd, "client_id", None), + except (CodexAppServerError, CodexRpcRejected) as exc: + if self._codex_thread_not_loaded(exc, current): + return False + raise + parent = parent_cache[current] + if parent is None: + return False + current = parent + raise RuntimeError("Codex fork ancestry exceeds the safety limit") + + async def _preflight_codex_delete_descendants( + self, + ctx: SessionContext, + native_sid: str, + residents: tuple[SessionContext, ...], + ) -> tuple[SessionContext | str | None, tuple[str, ...]]: + """Collect recursive-delete descendants and find protected work.""" + profile = self._codex_profile_for_ctx(ctx) + parent_cache: dict[str, str | None] = {} + resident_by_native: dict[str, SessionContext] = {} + descendant_sids: list[str] = [] + for candidate in residents: + candidate_native_sid = self._codex_context_native_thread_id( + candidate, + profile, ) - await self.transport.send(error) - return error - if engine == "claude": - await self._delete_claude_client_message_ids(sid) - else: - await self._delete_codex_client_message_ids(codex_alias_path) - await self._drop_preview_session(engine, sid) - if engine == "codex" and checkpoint_cleanup_journal is not None: - try: - await asyncio.to_thread( - checkpoint_cleanup_journal.cleanup, force=True - ) - except CheckpointError: - log.warning( - "Codex checkpoint cleanup after delete failed", session_id=sid + if candidate_native_sid is None: + raise RuntimeError( + "Codex resident has no native thread id" ) - if engine == "codex" and self._codex_controls is not None: - try: - await asyncio.to_thread(self._codex_controls.delete, sid) - except CodexControlStoreError: - log.warning( - "stale Codex controls cleanup failed", session_id=sid) - if engine == "codex" and self._session_plans is not None: - try: - await asyncio.to_thread(self._session_plans.delete, sid) - except SessionPlanStoreError: - log.warning( - "stale Codex plan cleanup failed", session_id=sid) - if self._session_presentation is not None: + resident_by_native[candidate_native_sid] = candidate + + candidate_activity = dict( + await ctx.sdk.list_thread_delete_candidates() + ) + for candidate_native_sid in resident_by_native: + candidate_activity.setdefault(candidate_native_sid, False) + for candidate_sid, watch in tuple(self._watch.items()): + if watch.get("engine") != "codex": + continue try: - await asyncio.to_thread( - self._session_presentation.delete, sid - ) - except SessionPresentationStoreError: - log.warning( - "stale Code presentation cleanup failed", - session_id=sid, + candidate_profile, candidate_native_sid = ( + self._codex_target(candidate_sid) ) - if self._session_pins is not None: - try: - await asyncio.to_thread( - self._session_pins.set_pinned, engine, sid, False) - except SessionPinStoreError: - log.warning("stale Code session pin cleanup failed", - engine=engine, session_id=sid) - self._watch.pop(sid, None) - if self.focused_sid in {sid, getattr(ctx, "key", None)}: + except ValueError: + continue + if candidate_profile.id != profile.id: + continue + # The watch supplies identity even when a just-created thread has + # not materialized in thread/list yet. Its activity must still be + # refreshed by _cold_codex_delete_blocked below, not trusted as a + # potentially stale snapshot here. + candidate_activity.setdefault(candidate_native_sid, False) + + for candidate_native_sid in sorted(candidate_activity): + if candidate_native_sid == native_sid: + if candidate_activity[candidate_native_sid]: + return candidate_native_sid, tuple(descendant_sids) + continue + is_descendant = await self._codex_thread_descends_from( + ctx.sdk, + candidate_native_sid, + native_sid, + parent_cache, + ) + if not is_descendant: + continue + candidate_sid = self._codex_wire_sid( + profile, + candidate_native_sid, + ) + descendant_sids.append(candidate_sid) + candidate = resident_by_native.get(candidate_native_sid) + if candidate is not None: + if ( + self._session_delete_busy(candidate) + or ( + not candidate.btw + and await self._codex_delete_external_owner( + candidate_sid + ) + ) + ): + return candidate, tuple(descendant_sids) + continue + if candidate_activity[candidate_native_sid]: + return candidate_sid, tuple(descendant_sids) + try: + rollout_path = self._codex_rollout_for_wire(candidate_sid) + except (OSError, ValueError): + rollout_path = None + if await self._cold_codex_delete_blocked( + candidate_sid, + rollout_path, + own_handle=ctx.sdk, + ): + return candidate_sid, tuple(descendant_sids) + return None, tuple(descendant_sids) + + async def _codex_delete_external_owner(self, sid: str) -> bool: + """Close the ownership interval before deleting one Code thread.""" + self._watch_session(sid) + watch = self._watch.get(sid) + if watch is None or watch.get("engine") != "codex": + log.warning( + "Codex delete ownership watch unavailable; failing closed", + session_id=sid, + ) + return True + external = await self._prime_codex_ownership(sid) + watch = self._watch.get(sid) + if watch is None or watch.get("engine") != "codex": + log.warning( + "Codex delete ownership watch disappeared; failing closed", + session_id=sid, + ) + return True + active_external_turns = watch.get("active_external_turns") + # Shared-daemon CLI turns intentionally remain writable from Remote, + # so ``_is_external`` does not report them. Deletion is different: a + # recursive thread/delete must never remove their active thread. + return bool( + external + or not watch.get("scan_complete", False) + or active_external_turns + ) + + async def _send_code_delete_error( + self, + cmd, + sid: str, + code: str, + message: str, + ) -> Error: + error = Error( + code=code, + message=message, + sid=sid, + to=getattr(cmd, "client_id", None), + ) + await self.transport.send(error) + return error + + async def _codex_session_exists(self, sid: str) -> bool: + """Check one exact native thread without a bounded catalog scan.""" + _profile, native_sid = self._codex_target(sid) + try: + response = await self._codex_rpc_for_wire( + sid, + "thread/read", + { + "threadId": sid, + "includeTurns": False, + }, + ) + except CodexRpcRejected as exc: + if self._codex_thread_not_loaded(exc, native_sid): + return False + raise + thread = response.get("thread") if isinstance(response, dict) else None + if not isinstance(thread, dict) or thread.get("id") != native_sid: + raise RuntimeError("Codex thread/read returned another thread") + return True + + @staticmethod + def _codex_thread_not_loaded( + exc: CodexAppServerError | CodexRpcRejected, + native_sid: str, + ) -> bool: + """Recognize only the app-server's exact missing-thread rejection.""" + return bool( + exc.code == -32600 + and exc.message.endswith(f"thread not loaded: {native_sid}") + ) + + async def _cold_codex_delete_context( + self, + cmd, + sid: str, + native_sid: str, + ) -> SessionContext | Error: + """Open a non-resident control connection for one cold deletion.""" + try: + profile, resolved_native_sid = self._codex_target(sid) + except ValueError: + return await self._send_code_delete_error( + cmd, + sid, + ERR_AUTH, + "Codex 账号或会话标识无效", + ) + if resolved_native_sid != native_sid: + return await self._send_code_delete_error( + cmd, + sid, + ERR_AUTH, + "Codex 会话标识不一致", + ) + + controls = await self._load_codex_session_controls(sid) + try: + native_cwd = await asyncio.to_thread( + self._codex_cwd_for_wire, + sid, + ) + except (OSError, ValueError): + native_cwd = None + target_cwd = next(( + os.path.realpath(candidate) + for candidate in ( + controls.cwd_override, + native_cwd, + self.cfg.cc_cwd, + str(profile.home), + ) + if isinstance(candidate, str) and os.path.isdir(candidate) + ), None) + if target_cwd is None: + return await self._send_code_delete_error( + cmd, + sid, + ERR_INVALID_CWD, + "没有可用目录连接 Codex，无法删除会话", + ) + + handle_kwargs = { + "cwd": target_cwd, + "daemon_mode": getattr( + self.cfg, + "codex_daemon_mode", + "auto", + ), + "daemon_manager": self._codex_daemon_for_profile(profile), + } + codex_home = self._codex_home(profile) + if codex_home is not None: + handle_kwargs["codex_home"] = codex_home + sdk = CodexHandle(self.cfg, **handle_kwargs) + try: + if sdk.daemon_mode == "auto": + try: + await sdk.connect(cwd=target_cwd, control_only=True) + except Exception as proxy_exc: + try: + await sdk.disconnect() + except Exception: + pass + log.warning( + "cold Codex shared delete connection failed; " + "using stdio", + session_id=sid, + error_type=type(proxy_exc).__name__, + ) + fallback_kwargs = dict(handle_kwargs) + fallback_kwargs["daemon_mode"] = "off" + fallback_kwargs.pop("daemon_manager", None) + sdk = CodexHandle(self.cfg, **fallback_kwargs) + await sdk.connect( + resume_id=native_sid, + cwd=target_cwd, + ) + else: + await sdk.connect(resume_id=native_sid, cwd=target_cwd) + except Exception as exc: + try: + await sdk.disconnect() + except Exception: + pass + log.warning( + "cold Codex delete control connection failed", + session_id=sid, + error_type=type(exc).__name__, + ) + return await self._send_code_delete_error( + cmd, + sid, + ERR_NOT_RUNNING, + "Codex 删除通道连接失败，请稍后重试", + ) + ctx = SessionContext( + session_id=native_sid, + sdk=sdk, + buffer=RingBuffer( + self.cfg.ring_max_events, + self.cfg.ring_max_bytes, + ), + cwd=target_cwd, + engine="codex", + codex_profile_id=profile.id, + space="code", + ) + # This transient context never owns a checkpoint object. Confirmed + # cold deletion cleans persistent repository buckets by session id. + ctx.codex_checkpoint = False + return ctx + + async def _cleanup_deleted_codex_checkpoint( + self, + ctx: SessionContext, + sid: str, + ) -> None: + """Remove one deleted Code thread's private checkpoint journal.""" + journal = ctx.codex_checkpoint + if journal is False: + return + if journal is None: + # A resumed context can be using cfg.cc_cwd after its persisted cwd + # disappeared. Deletion is session-wide, so do not infer the + # journal bucket from that fallback directory. + ctx.codex_checkpoint = False + await self._cleanup_cold_deleted_codex_checkpoint(sid) + return + ctx.codex_checkpoint = False + try: + await asyncio.to_thread(journal.cleanup, force=True) + except (CheckpointError, OSError): + log.warning( + "Codex checkpoint cleanup after delete failed", + session_id=sid, + ) + + async def _cleanup_cold_deleted_codex_checkpoint( + self, + sid: str, + ) -> None: + """Remove a confirmed cold thread's journals from every former cwd.""" + try: + await asyncio.to_thread( + cleanup_codex_checkpoint_session, + Path(self.cfg.state_dir), + sid, + ) + except (CheckpointError, OSError, ValueError) as exc: + log.warning( + "cold Codex checkpoint cleanup after delete failed", + session_id=sid, + error_type=type(exc).__name__, + ) + + async def _cold_codex_delete_blocked( + self, + sid: str, + rollout_path: str | None, + *, + own_handle: CodexHandle, + ) -> bool: + """Check cold shared ownership without registering a resident watch.""" + watch = self._watch.get(sid) + if watch is not None and watch.get("engine") == "codex": + await self._prime_codex_ownership( + sid, + extra_handles=(own_handle,), + ) + return bool( + not watch.get("scan_complete", False) + or watch.get("active_external_turns") + ) + if not rollout_path: + return False + try: + stat_result = await asyncio.to_thread(os.stat, rollout_path) + except OSError: + return False + active_turns, _partial = self._codex_tail_state( + rollout_path, + stat_result.st_size, + ) + ephemeral_watch = { + "active_external_turns": { + turn_id: time.time() + for turn_id in active_turns + }, + "seeded_external_turns": set(active_turns), + # A private Codex App turn can write through short-lived opens and + # expose no stable holder. A destructive one-shot probe must keep + # its active tail marker; a later retry can observe the terminal, + # while a normal watch can retire a proven crashed orphan by TTL. + "preserve_seeded_without_holder": True, + "takeover_holders": set(), + "takeover_interactive_holders": set(), + } + async with self._codex_watch_lock: + scan = await self._probe_codex_holders( + {sid: rollout_path}, + extra_handles=(own_handle,), + ) + self._codex_holder_sets( + ephemeral_watch, + scan, + sid, + ) + scan_complete = self._codex_scan_complete_for_sid(scan, sid) + return bool( + not scan_complete + or ephemeral_watch["active_external_turns"] + ) + + async def _reconcile_loaded_codex_delete( + self, + ctx: SessionContext, + sid: str, + profile: CodexProfile, + preflight_descendant_sids: tuple[str, ...], + deleted_native_ids: tuple[str, ...] | list[str], + ) -> tuple[list[str], list[SessionContext], set[str]]: + """Confirm deleted identities and remove residents while locked.""" + self._invalidate_codex_session_catalog() + candidate_sids = list(preflight_descendant_sids) + for deleted_native_sid in deleted_native_ids: + if not isinstance(deleted_native_sid, str): + continue + try: + deleted_sid = self._codex_wire_sid( + profile, + deleted_native_sid, + ) + except ValueError: + continue + if deleted_sid not in candidate_sids: + candidate_sids.append(deleted_sid) + if getattr( + ctx.sdk, + "thread_delete_notifications_overflowed", + False, + ): + log.warning( + "Codex delete notifications overflowed; reconciling " + "preflight descendants", + session_id=sid, + candidate_count=len(candidate_sids), + ) + resident_identities: list[ + tuple[SessionContext, frozenset[str]] + ] = [] + indexed_contexts: set[int] = set() + for candidate in tuple(self.sessions.values()): + if ( + candidate.engine != "codex" + or self._codex_profile_for_ctx(candidate).id != profile.id + ): + continue + identity = id(candidate) + if identity in indexed_contexts: + continue + indexed_contexts.add(identity) + identity_sids: set[str] = set() + route_sid = self._ctx_wire_sid(candidate) + if route_sid is not None: + identity_sids.add(route_sid) + candidate_native_sid = self._codex_context_native_thread_id( + candidate, + profile, + ) + if candidate_native_sid is not None: + identity_sids.add(self._codex_wire_sid( + profile, + candidate_native_sid, + )) + resident_identities.append(( + candidate, + frozenset(identity_sids), + )) + deleted_sids = [sid] + if ctx.space == "code": + for candidate_sid in candidate_sids: + if candidate_sid == sid: + continue + try: + still_exists = await self._codex_session_exists( + candidate_sid, + ) + except Exception as exc: + log.warning( + "Codex descendant deletion could not be confirmed", + session_id=candidate_sid, + error_type=type(exc).__name__, + ) + continue + if not still_exists: + deleted_sids.append(candidate_sid) + deleted_set = set(deleted_sids) + deleted_contexts: list[SessionContext] = [] + deleted_context_ids: set[int] = set() + deleted_context_sids: set[str] = set() + for candidate, candidate_sids in resident_identities: + identity = id(candidate) + if ( + candidate_sids.isdisjoint(deleted_set) + or identity in deleted_context_ids + ): + continue + deleted_contexts.append(candidate) + deleted_context_ids.add(identity) + deleted_context_sids.update(candidate_sids) + for deleted_ctx_sid in sorted(deleted_context_sids): + if deleted_ctx_sid not in deleted_set: + deleted_sids.append(deleted_ctx_sid) + deleted_set.add(deleted_ctx_sid) + for deleted_ctx in deleted_contexts: + if ( + deleted_ctx.key is not None + and self.sessions.get(deleted_ctx.key) is deleted_ctx + ): + self.sessions.pop(deleted_ctx.key, None) + return deleted_sids, deleted_contexts, deleted_context_sids + + async def _delete_loaded_codex_thread( + self, + cmd, + sid: str, + native_sid: str, + ctx: SessionContext, + *, + rollout_path: str | None, + transient: bool = False, + ) -> Error | tuple[str, ...]: + """Delete a loaded Codex thread before releasing its writer.""" + async with ctx.query_lock: + if self._session_delete_busy(ctx): + return await self._send_code_delete_error( + cmd, + sid, + ERR_BUSY, + "会话仍在运行或有排队消息，请先停止并取消排队后再删除", + ) + if ctx.space == "code": + if transient: + if ( + ctx.sdk.daemon_mode == "auto" + and not self._codex_shared_live(ctx) + ): + return await self._send_code_delete_error( + cmd, + sid, + ERR_NOT_RUNNING, + "Codex 共享删除通道不可用，请重试", + ) + else: + control_error = await self._runtime_control_preflight( + ctx, + action="删除会话", + request_id=getattr(cmd, "cmd_id", None), + client_id=getattr(cmd, "client_id", None), + ) + if control_error is not None: + return control_error + if transient: + external_owner = await self._cold_codex_delete_blocked( + sid, + rollout_path, + own_handle=ctx.sdk, + ) + else: + external_owner = ( + await self._codex_delete_external_owner(sid) + ) + if external_owner: + return await self._send_code_delete_error( + cmd, + sid, + ERR_BUSY, + "会话正由 Codex App 使用，无法删除", + ) + if self._session_delete_busy(ctx): + return await self._send_code_delete_error( + cmd, + sid, + ERR_BUSY, + "会话状态已变化，请等待当前回合结束后再删除", + ) + + profile = self._codex_profile_for_ctx(ctx) + resident_candidates = sorted( + ( + candidate + for candidate in tuple(self.sessions.values()) + if candidate is not ctx + and candidate.engine == "codex" + and self._codex_profile_for_ctx(candidate).id == profile.id + ), + key=lambda candidate: candidate.key or "", + ) + async with AsyncExitStack() as resident_locks: + await resident_locks.enter_async_context( + ctx.queued_query_lock + ) + for candidate in resident_candidates: + await resident_locks.enter_async_context( + candidate.query_lock + ) + await resident_locks.enter_async_context( + candidate.queued_query_lock + ) + resident_candidates = [ + candidate + for candidate in resident_candidates + if self._is_resident_context(candidate) + ] + if self._session_delete_busy(ctx): + return await self._send_code_delete_error( + cmd, + sid, + ERR_BUSY, + "会话状态已变化，请等待当前回合结束后再删除", + ) + try: + ( + busy_descendant, + preflight_descendant_sids, + ) = await self._preflight_codex_delete_descendants( + ctx, + native_sid, + tuple(resident_candidates), + ) + except Exception as exc: + log.warning( + "Codex descendant deletion preflight failed", + session_id=sid, + error_type=type(exc).__name__, + ) + return await self._send_code_delete_error( + cmd, + sid, + ERR_INTERNAL, + "无法确认派生会话状态，未执行删除", + ) + if busy_descendant is not None: + return await self._send_code_delete_error( + cmd, + sid, + ERR_BUSY, + "派生会话仍在运行、排队或由其他 Codex 客户端使用", + ) + + try: + deleted_native_ids = await ctx.sdk.delete_thread( + native_sid + ) + except CodexAppServerError as exc: + log.warning( + "loaded Codex session deletion rejected", + session_id=sid, + error_code=exc.code, + ) + return await self._send_code_delete_error( + cmd, + sid, + ERR_INTERNAL, + "会话删除失败，请刷新后重试", + ) + except Exception as exc: + # A transport loss after the request write cannot prove + # whether thread/delete committed. Require two exact + # native-storage signals before deciding that local + # metadata may be removed. + log.warning( + "loaded Codex session deletion outcome unknown", + session_id=sid, + error_type=type(exc).__name__, + ) + try: + still_exists = await self._codex_session_exists(sid) + except Exception as reconcile_error: + log.warning( + "Codex deletion reconciliation failed", + session_id=sid, + error_type=type(reconcile_error).__name__, + ) + return await self._send_code_delete_error( + cmd, + sid, + ERR_INTERNAL, + "会话删除结果暂时无法确认，请刷新后重试", + ) + rollout_gone = bool( + rollout_path + and not await asyncio.to_thread( + os.path.exists, + rollout_path, + ) + ) + if still_exists or not rollout_gone: + return await self._send_code_delete_error( + cmd, + sid, + ERR_INTERNAL, + "会话删除失败，请刷新后重试", + ) + deleted_native_ids = (native_sid,) + if not isinstance(deleted_native_ids, (tuple, list)): + deleted_native_ids = (native_sid,) + ( + deleted_sids, + deleted_contexts, + deleted_context_sids, + ) = await self._reconcile_loaded_codex_delete( + ctx, + sid, + profile, + preflight_descendant_sids, + deleted_native_ids, + ) + + for deleted_ctx in deleted_contexts: + deleted_sid = self._ctx_wire_sid(deleted_ctx) + await self._discard_query_queue(deleted_ctx) + tasks = { + task + for task in ( + deleted_ctx.turn_task, + deleted_ctx.codex_spontaneous_task, + ) + if task is not None + and task is not asyncio.current_task() + and not task.done() + } + for task in tasks: + task.cancel() + if tasks: + await asyncio.gather(*tasks, return_exceptions=True) + try: + await deleted_ctx.sdk.disconnect() + except Exception as exc: + # Native deletion already committed. A failed proxy cleanup + # must not preserve a context which cannot be resumed. + log.warning( + "deleted Codex session disconnect failed", + session_id=self._ctx_wire_sid(deleted_ctx), + error_type=type(exc).__name__, + ) + finally: + await self._cleanup_codex_steer_attachments(deleted_ctx) + if deleted_ctx.space == "code": + await self._cleanup_deleted_codex_checkpoint( + deleted_ctx, + deleted_sid, + ) + self._purge_preview_image_snapshots( + deleted_ctx.preview_snapshot_token, + ) + for deleted_sid in deleted_sids: + if ( + deleted_sid != sid + and deleted_sid not in deleted_context_sids + ): + await self._cleanup_cold_deleted_codex_checkpoint( + deleted_sid, + ) + return tuple(deleted_sids) + + async def _handle_delete_session(self, cmd): + """Delete one native session without confusing Code and Work roots.""" + if getattr(cmd, "space", "code") == "work": + return await self._handle_delete_work_session(cmd) + sid = self._resolve_session_alias(cmd.session_id) or cmd.session_id + requested_engine = getattr(cmd, "engine", "claude") + is_codex = await self._is_codex_session(sid) + engine = "codex" if is_codex else "claude" + if engine != requested_engine: + error = Error( + code=ERR_AUTH, + message="会话不属于请求的引擎", + sid=sid, + to=getattr(cmd, "client_id", None), + ) + await self.transport.send(error) + return error + try: + native_sid, work_profile_id = self._work_session_identity(engine, sid) + except ValueError: + native_sid, work_profile_id = sid, None + work_record = await asyncio.to_thread( + self._work.for_engine(engine).get_by_session, + native_sid, + codex_profile_id=work_profile_id, + ) + if work_record is not None: + error = Error( + code=ERR_AUTH, + message="Work 会话必须从 Work 空间删除", + sid=sid, + to=getattr(cmd, "client_id", None), + ) + await self.transport.send(error) + return error + ctx = self._ctx_for(sid) + transient_codex_ctx = False + if engine == "codex" and ctx is None: + ctx_or_error = await self._cold_codex_delete_context( + cmd, + sid, + native_sid, + ) + if isinstance(ctx_or_error, Error): + return ctx_or_error + ctx = ctx_or_error + transient_codex_ctx = True + if ctx is not None and self._session_delete_busy(ctx): + return await self._send_code_delete_error( + cmd, + sid, + ERR_BUSY, + "会话仍在运行或有排队消息，请先停止并取消排队后再删除", + ) + cwd = ctx.cwd if ctx is not None else None + codex_alias_path = None + if engine == "codex": + try: + codex_alias_path = await asyncio.to_thread( + self._codex_rollout_for_wire, sid) + except (OSError, ValueError): + codex_alias_path = None + if engine == "codex" and cwd is None: + try: + cwd = await asyncio.to_thread( + self._codex_cwd_for_wire, sid) + except ValueError: + cwd = None + if engine == "claude" and cwd is None: + info = await asyncio.to_thread(get_session_info, sid) + cwd = info.cwd if info is not None else None + # A metadata-only Claude transcript has no cwd but is still a real + # exact-SID file in the SDK catalog. delete_session(directory=None) + # safely searches all project roots for that UUID and deletes only + # the matching transcript. Preserve the old not-found rejection + # when no exact transcript exists. + if not cwd and transcript_path(sid) is None: + error = Error( + code=ERR_NOT_RUNNING, + message="Claude 会话不存在", + sid=sid, + to=getattr(cmd, "client_id", None), + ) + await self.transport.send(error) + return error + if engine == "codex": + assert ctx is not None + try: + delete_result = await self._delete_loaded_codex_thread( + cmd, + sid, + native_sid, + ctx, + rollout_path=codex_alias_path, + transient=transient_codex_ctx, + ) + finally: + if transient_codex_ctx: + try: + await ctx.sdk.disconnect() + except Exception as exc: + log.warning( + "cold Codex delete control cleanup failed", + session_id=sid, + error_type=type(exc).__name__, + ) + if isinstance(delete_result, Error): + return delete_result + deleted_sids = delete_result + if transient_codex_ctx: + await self._cleanup_cold_deleted_codex_checkpoint(sid) + else: + if ctx is not None: + try: + await ctx.sdk.disconnect() + except Exception: + log.exception( + "session disconnect before delete failed", + engine=engine, + session_id=sid, + ) + return await self._send_code_delete_error( + cmd, + sid, + ERR_INTERNAL, + "无法安全停止会话，未执行删除", + ) + self.sessions.pop(ctx.key or sid, None) + self._purge_preview_image_snapshots( + ctx.preview_snapshot_token, + ) + try: + await asyncio.to_thread(delete_session, sid, directory=cwd) + except Exception: + log.exception( + "Code session deletion failed", + engine=engine, + session_id=sid, + ) + return await self._send_code_delete_error( + cmd, + sid, + ERR_INTERNAL, + "会话删除失败，请刷新后重试", + ) + deleted_sids = (sid,) + if engine == "claude": + await self._delete_claude_client_message_ids(sid) + else: + await self._delete_codex_client_message_ids(codex_alias_path) + for deleted_sid in deleted_sids: + await self._drop_preview_session(engine, deleted_sid) + for deleted_sid in deleted_sids: + if engine == "codex" and self._codex_controls is not None: + try: + await asyncio.to_thread( + self._codex_controls.delete, + deleted_sid, + ) + except CodexControlStoreError: + log.warning( + "stale Codex controls cleanup failed", + session_id=deleted_sid, + ) + if engine == "codex" and self._session_plans is not None: + try: + await asyncio.to_thread( + self._session_plans.delete, + deleted_sid, + ) + except SessionPlanStoreError: + log.warning( + "stale Codex plan cleanup failed", + session_id=deleted_sid, + ) + if self._session_presentation is not None: + try: + await asyncio.to_thread( + self._session_presentation.delete, + deleted_sid, + ) + except SessionPresentationStoreError: + log.warning( + "stale Code presentation cleanup failed", + session_id=deleted_sid, + ) + if self._session_pins is not None: + try: + await asyncio.to_thread( + self._session_pins.set_pinned, + engine, + deleted_sid, + False, + ) + except SessionPinStoreError: + log.warning( + "stale Code session pin cleanup failed", + engine=engine, + session_id=deleted_sid, + ) + self._watch.pop(deleted_sid, None) + self._codex_sidebar_watches.pop(deleted_sid, None) + if self.focused_sid in set(deleted_sids) | { + getattr(ctx, "key", None), + }: self.focused_sid = None await self._handle_list_sessions(cmd) log.info("Code session deleted", engine=engine, session_id=sid) @@ -19956,6 +21012,11 @@ async def _finish_same_cwd_fork( request_id=cmd.request_id, to=cmd.client_id, ) + # A successful fork mutates the native catalog. Drop any parent-only + # snapshot before publishing the child: otherwise the refresh below can + # immediately erase the newly focused child from the browser and make a + # later fork target whichever session the sidebar selects instead. + self._invalidate_codex_session_catalog() await self.transport.send(event) try: await self._list_codex_sessions(cmd) @@ -20819,6 +21880,9 @@ async def _finish_worktree_fork( request_id=cmd.request_id, to=cmd.client_id, ) + # Worktree forks enter the same native catalog as same-cwd forks. Keep + # both publication paths on the same cache-generation boundary. + self._invalidate_codex_session_catalog() await self.transport.send(event) try: await self._list_codex_sessions(cmd) diff --git a/tests/test_claude_permission_state.py b/tests/test_claude_permission_state.py index 9829f85..f2f345d 100644 --- a/tests/test_claude_permission_state.py +++ b/tests/test_claude_permission_state.py @@ -388,6 +388,12 @@ async def go(): SdkHandle, "preflight", staticmethod(lambda _path: None)) monkeypatch.setenv("HOME", str(tmp_path)) monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) + monkeypatch.delenv("ANTHROPIC_MODEL", raising=False) + monkeypatch.setattr( + machine_module.WrapperMachine, + "_claude_managed_settings_paths", + staticmethod(lambda: []), + ) machine, transport = _mk_machine() machine._load_history = lambda *_args: asyncio.sleep(0) @@ -428,6 +434,12 @@ async def go(): SdkHandle, "preflight", staticmethod(lambda _path: None)) monkeypatch.setenv("HOME", str(tmp_path)) monkeypatch.delenv("CLAUDE_CONFIG_DIR", raising=False) + monkeypatch.delenv("ANTHROPIC_MODEL", raising=False) + monkeypatch.setattr( + machine_module.WrapperMachine, + "_claude_managed_settings_paths", + staticmethod(lambda: []), + ) machine, transport = _mk_machine() machine._load_history = lambda *_args: asyncio.sleep(0) diff --git a/tests/test_codex_checkpoints.py b/tests/test_codex_checkpoints.py index fd11fa4..1b664fd 100644 --- a/tests/test_codex_checkpoints.py +++ b/tests/test_codex_checkpoints.py @@ -15,6 +15,7 @@ CompletedCheckpoint, CodexCheckpointJournal, NotGitWorkspaceError, + cleanup_codex_checkpoint_session, ) @@ -564,3 +565,28 @@ def test_git_boundary_abort_and_cleanup_are_explicit(tmp_path): corrupt.manifest_path.write_text("{not-json", encoding="utf-8") corrupt.cleanup(force=True) assert not corrupt.session_dir.exists() + + +def test_cleanup_session_removes_journals_from_every_repository(tmp_path): + first_parent = tmp_path / "first" + second_parent = tmp_path / "second" + first_parent.mkdir() + second_parent.mkdir() + first = _repo(first_parent, {"first.txt": "first\n"}) + second = _repo(second_parent, {"second.txt": "second\n"}) + state = tmp_path / "state" + first_journal = CodexCheckpointJournal( + str(first), + state, + "cold-child", + ) + second_journal = CodexCheckpointJournal( + str(second), + state, + "cold-child", + ) + + assert cleanup_codex_checkpoint_session(state, "cold-child") == 2 + assert not first_journal.session_dir.exists() + assert not second_journal.session_dir.exists() + assert cleanup_codex_checkpoint_session(state, "cold-child") == 0 diff --git a/tests/test_codex_controls.py b/tests/test_codex_controls.py index 29a753c..d010d6a 100644 --- a/tests/test_codex_controls.py +++ b/tests/test_codex_controls.py @@ -90,6 +90,396 @@ async def send(request): asyncio.run(run()) +def test_codex_delete_thread_uses_loaded_app_server_connection(): + async def run(): + handle = CodexHandle(_Cfg()) + handle.thread_id = "thread-1" + requests = [] + + async def request(method, params): + requests.append((method, params)) + return {} + + handle._request = request + + deleted = await handle.delete_thread("thread-1") + + assert requests == [( + "thread/delete", + {"threadId": "thread-1"}, + )] + assert deleted == ("thread-1",) + assert handle.thread_id is None + + asyncio.run(run()) + + +def test_codex_read_thread_parent_uses_exact_loaded_thread_metadata(): + async def run(): + handle = CodexHandle(_Cfg()) + requests = [] + + async def request(method, params): + requests.append((method, params)) + thread_id = params["threadId"] + return {"thread": { + "id": thread_id, + "forkedFromId": ( + "thread-1" if thread_id == "thread-2" else None + ), + }} + + handle._request = request + + assert await handle.read_thread_parent("thread-2") == "thread-1" + assert await handle.read_thread_parent("thread-1") is None + assert requests == [ + ( + "thread/read", + {"threadId": "thread-2", "includeTurns": False}, + ), + ( + "thread/read", + {"threadId": "thread-1", "includeTurns": False}, + ), + ] + + asyncio.run(run()) + + +def test_codex_delete_catalog_pages_active_and_archived_threads(): + async def run(): + handle = CodexHandle(_Cfg()) + requests = [] + + async def request(method, params): + requests.append((method, params)) + key = (params["archived"], params.get("cursor")) + pages = { + (False, None): { + "data": [{ + "id": "thread-root", + "status": {"type": "idle"}, + }], + "nextCursor": "active-next", + }, + (False, "active-next"): { + "data": [{ + "id": "thread-child", + "status": {"type": "active"}, + }], + "nextCursor": None, + }, + (True, None): { + "data": [{ + "id": "thread-archived", + "status": {"type": "notLoaded"}, + }], + "nextCursor": None, + }, + } + assert method == "thread/list" + return pages[key] + + handle._request = request + + assert await handle.list_thread_delete_candidates() == ( + ("thread-root", False), + ("thread-child", True), + ("thread-archived", False), + ) + assert [params for _method, params in requests] == [ + { + "archived": False, + "limit": 100, + "sortKey": "updated_at", + "sortDirection": "desc", + }, + { + "archived": False, + "limit": 100, + "sortKey": "updated_at", + "sortDirection": "desc", + "cursor": "active-next", + }, + { + "archived": True, + "limit": 100, + "sortKey": "updated_at", + "sortDirection": "desc", + }, + ] + + asyncio.run(run()) + + +def test_codex_unloaded_delete_requires_control_connection(): + async def run(): + handle = CodexHandle(_Cfg()) + + async def forbidden_request(*_args): + raise AssertionError("invalid delete must not reach app-server") + + handle._request = forbidden_request + + with pytest.raises(RuntimeError, match="live control connection"): + await handle.delete_thread("thread-1") + + asyncio.run(run()) + + +@pytest.mark.parametrize( + ("work_mode", "connect_kwargs"), + [ + (False, {"resume_id": "thread-1"}), + (False, {"fork": True}), + (True, {}), + ], +) +def test_codex_control_connection_rejects_thread_binding( + work_mode, + connect_kwargs, +): + async def run(): + handle = CodexHandle(_Cfg(), work_mode=work_mode) + + with pytest.raises(ValueError, match="cannot bind a thread"): + await handle.connect(control_only=True, **connect_kwargs) + + asyncio.run(run()) + + +def test_codex_delete_thread_rejects_wrong_or_active_thread(): + async def run(): + handle = CodexHandle(_Cfg()) + handle.thread_id = "thread-1" + + async def forbidden_request(*_args): + raise AssertionError("invalid delete must not reach app-server") + + handle._request = forbidden_request + + with pytest.raises(ValueError, match="does not match"): + await handle.delete_thread("thread-2") + + handle.turn_active = True + with pytest.raises(RuntimeError, match="turn is active"): + await handle.delete_thread("thread-1") + + asyncio.run(run()) + + +def test_codex_control_only_connect_skips_thread_binding(monkeypatch): + class Manager: + mode = "auto" + strict_shared_affinity = True + + async def proxy_args(self, _bin, _env): + return ["/usr/bin/codex", "app-server", "proxy"] + + def invalidate(self): + pass + + async def run(): + monkeypatch.setattr( + codex_handle_module, + "_resolve_codex_bin", + lambda: "/usr/bin/codex", + ) + monkeypatch.setattr( + codex_handle_module, + "_newer_private_core_for_oversized_resume", + lambda _bin, _sid: None, + ) + monkeypatch.setattr( + codex_handle_module, + "_oversized_desktop_openai_resume_requires_http", + lambda _sid: False, + ) + handle = CodexHandle( + _Cfg(), + daemon_mode="auto", + daemon_manager=Manager(), + ) + requests = [] + server_requests = [] + + async def handle_server_request(message): + server_requests.append(message) + + handle._handle_server_request = handle_server_request + + async def open_process(_argv, _bin, *, daemon_proxy): + assert daemon_proxy is True + handle.proc = SimpleNamespace(returncode=None) + handle._using_daemon_proxy = True + handle._dead = False + + async def request(method, params=None): + requests.append((method, params)) + if method == "initialize": + await handle._dispatch({ + "method": "thread/started", + "params": {"thread": { + "id": "initializing-sibling-thread", + "source": "cli", + }}, + }) + await handle._dispatch({ + "id": "initializing-approval", + "method": "item/commandExecution/requestApproval", + "params": {"threadId": "initializing-sibling-thread"}, + }) + return {"userAgent": "codex_cli_rs/0.147.0 (test)"} + assert method == "thread/delete" + return {} + + handle._open_process = open_process + handle._request = request + handle._notify = lambda *_args, **_kwargs: asyncio.sleep(0) + + await handle.connect(cwd="/tmp", control_only=True) + + assert requests == [( + "initialize", + codex_handle_module._initialize_params(), + )] + await handle._dispatch({ + "method": "thread/started", + "params": {"thread": { + "id": "sibling-thread", + "source": "cli", + }}, + }) + await handle._dispatch({ + "id": "approval", + "method": "item/commandExecution/requestApproval", + "params": {"threadId": "sibling-thread"}, + }) + await asyncio.sleep(0) + assert handle.thread_id is None + assert server_requests == [] + assert handle._pending_server_request_ids == set() + assert handle.using_daemon_proxy is True + assert await handle.delete_thread("thread-1") == ("thread-1",) + assert requests[-1] == ( + "thread/delete", + {"threadId": "thread-1"}, + ) + handle.proc = None + + asyncio.run(run()) + + +def test_codex_delete_thread_collects_descendant_notifications(): + async def run(): + handle = CodexHandle(_Cfg()) + handle.thread_id = "thread-1" + + async def request(method, params): + assert method == "thread/delete" + assert params == {"threadId": "thread-1"} + await handle._dispatch({ + "method": "thread/deleted", + "params": {"threadId": "child-1"}, + }) + await handle._dispatch({ + "method": "thread/deleted", + "params": {"threadId": "thread-1"}, + }) + return {} + + handle._request = request + + deleted = await handle.delete_thread("thread-1") + + assert deleted == ("child-1", "thread-1") + assert handle._capture_thread_deleted_notification({ + "method": "thread/deleted", + "params": {"threadId": "unrelated"}, + }) is False + + asyncio.run(run()) + + +def test_codex_delete_thread_marks_notification_overflow(monkeypatch): + async def run(): + monkeypatch.setattr( + codex_handle_module, + "_THREAD_DELETE_NOTIFY_MAX", + 2, + ) + handle = CodexHandle(_Cfg()) + handle.thread_id = "thread-1" + + async def request(_method, _params): + for thread_id in ("child-1", "child-2", "child-3", "thread-1"): + await handle._dispatch({ + "method": "thread/deleted", + "params": {"threadId": thread_id}, + }) + return {} + + handle._request = request + + deleted = await handle.delete_thread("thread-1") + + assert deleted == ("child-1", "child-2", "thread-1") + assert handle.thread_delete_notifications_overflowed is True + + asyncio.run(run()) + + +def test_codex_delete_thread_collects_notifications_after_root( + monkeypatch, +): + async def run(): + monkeypatch.setattr( + codex_handle_module, + "_THREAD_DELETE_NOTIFY_TIMEOUT", + 0.05, + ) + handle = CodexHandle(_Cfg()) + handle.thread_id = "thread-1" + reader_release = asyncio.Event() + handle._reader = asyncio.create_task(reader_release.wait()) + notification_task = None + + async def notify_after_response(): + await asyncio.sleep(0) + await handle._dispatch({ + "method": "thread/deleted", + "params": {"threadId": "thread-1"}, + }) + await asyncio.sleep(0.01) + await handle._dispatch({ + "method": "thread/deleted", + "params": {"threadId": "child-1"}, + }) + + async def request(method, params): + nonlocal notification_task + assert method == "thread/delete" + assert params == {"threadId": "thread-1"} + notification_task = asyncio.create_task( + notify_after_response() + ) + return {} + + handle._request = request + try: + deleted = await handle.delete_thread("thread-1") + assert notification_task is not None + await notification_task + finally: + reader_release.set() + await handle._reader + + assert deleted == ("thread-1", "child-1") + + asyncio.run(run()) + + def test_codex_initialize_declares_experimental_api_for_collaboration_mode(): assert codex_handle_module._initialize_params() == { "clientInfo": {"name": "cc-remote", "version": __version__}, diff --git a/tests/test_codex_forks.py b/tests/test_codex_forks.py index f681354..b228678 100644 --- a/tests/test_codex_forks.py +++ b/tests/test_codex_forks.py @@ -159,6 +159,34 @@ def test_fork_journal_aliases_same_unresolved_identity_to_one_canonical(tmp_path assert later["thread_source"] == "cc-remote-fork:request-later" +def test_fork_journal_lists_unique_completed_results_newest_first(tmp_path): + journal = CodexForkJournal(tmp_path) + journal.begin("request-old", "parent", "turn-old", "/repo") + journal.claim_submission("request-old") + journal.begin("request-alias", "parent", "turn-old", "/repo") + journal.complete("request-alias", "child-old") + journal.begin("request-new", "parent", "turn-new", "/repo") + journal.complete("request-new", "child-new") + assert journal.set_title("child-old", "Renamed fork") is True + assert journal.set_title("not-a-fork", "Ignored") is False + + results = journal.completed_results(10) + + assert [entry["session_id"] for entry in results] == [ + "child-new", + "child-old", + ] + assert len({entry["thread_source"] for entry in results}) == 2 + old_result = next( + entry for entry in results if entry["session_id"] == "child-old") + assert old_result["title"] == "Renamed fork" + assert old_result["title_updated_at"] >= old_result["created_at"] + reloaded = CodexForkJournal(tmp_path) + assert reloaded.completed_results(10)[1]["title"] == "Renamed fork" + with pytest.raises(ForkJournalError, match="invalid completed"): + journal.completed_results(0) + + def test_fork_journal_rejects_orphaned_alias_on_reload(tmp_path): journal = CodexForkJournal(tmp_path) journal.begin("request-old", "parent", "turn-1", "/repo") diff --git a/tests/test_codex_session_delete.py b/tests/test_codex_session_delete.py new file mode 100644 index 0000000..2e31bfb --- /dev/null +++ b/tests/test_codex_session_delete.py @@ -0,0 +1,1800 @@ +"""Zero-token regressions for authoritative Codex thread deletion.""" +from __future__ import annotations + +import asyncio +import json +import shutil +import subprocess +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from cc_remote.config import WrapperConfig +from cc_remote.protocol import ( + DeleteSession, + DeleteWorkSession, + ERR_BUSY, + ERR_INTERNAL, + ERR_NOT_RUNNING, + Error, + Query, +) +from cc_remote.wrapper import machine as machine_module +from cc_remote.wrapper.codex_handle import CodexAppServerError +from cc_remote.wrapper.codex_external import HolderScan +from cc_remote.wrapper.codex_rpc import CodexRpcRejected +from cc_remote.wrapper.machine import WrapperMachine +from cc_remote.wrapper.process_scan import ProcessIdentity +from tests.test_multisession import _mk_ctx, _mk_machine + + +class _Transport: + def __init__(self) -> None: + self.sent: list[object] = [] + self.on_connected = None + + async def send(self, message: object) -> None: + self.sent.append(message) + + +class _DeleteRecorder: + def __init__(self, kind: str, calls: list[tuple]) -> None: + self.kind = kind + self.calls = calls + + def delete(self, sid: str) -> None: + self.calls.append((self.kind, sid)) + + +class _PinRecorder: + def __init__(self, calls: list[tuple]) -> None: + self.calls = calls + + def set_pinned(self, engine: str, sid: str, pinned: bool) -> None: + self.calls.append(("pin", engine, sid, pinned)) + + +class _DeleteHandle: + def __init__( + self, + failure: Exception | None = None, + deleted_ids: tuple[str, ...] | None = None, + ): + self.thread_id = "codex-thread" + self.turn_active = False + self.turn_start_pending = False + self.shared_daemon_affinity = True + self.using_daemon_proxy = True + self.failure = failure + self.deleted_ids = deleted_ids + self.calls: list[tuple[str, str | None]] = [] + self.parents: dict[str, str | None] = {} + self.parent_calls: list[str] = [] + self.delete_candidates: tuple[tuple[str, bool], ...] = () + self.thread_delete_notifications_overflowed = False + + async def list_thread_delete_candidates( + self, + ) -> tuple[tuple[str, bool], ...]: + return self.delete_candidates + + async def read_thread_parent(self, thread_id: str) -> str | None: + self.parent_calls.append(thread_id) + return self.parents.get(thread_id) + + async def delete_thread( + self, + expected_thread_id: str, + ) -> tuple[str, ...]: + self.calls.append(("delete", expected_thread_id)) + if self.failure is not None: + raise self.failure + self.thread_id = None + return self.deleted_ids or (expected_thread_id,) + + async def disconnect(self) -> None: + self.calls.append(("disconnect", None)) + + +class _ColdDeleteHandle(_DeleteHandle): + created: list[_ColdDeleteHandle] = [] + + def __init__( + self, + _cfg, + cwd=None, + daemon_mode=None, + daemon_manager=None, + codex_home=None, + ) -> None: + super().__init__() + self.thread_id = None + self.cwd = cwd + self.daemon_mode = daemon_mode + self.daemon_manager = daemon_manager + self.codex_home = codex_home + self.using_daemon_proxy = False + self.shared_daemon_affinity = False + self.proc = SimpleNamespace(pid=4242, returncode=None) + self.created.append(self) + + async def connect(self, **kwargs) -> None: + self.calls.append(("connect", kwargs)) + assert kwargs.get("resume_id") is None + assert kwargs["control_only"] is True + self.using_daemon_proxy = True + self.shared_daemon_affinity = True + + +class _FallbackColdDeleteHandle(_ColdDeleteHandle): + async def connect(self, **kwargs) -> None: + self.calls.append(("connect", kwargs)) + if self.daemon_mode == "auto": + assert kwargs == { + "cwd": self.cwd, + "control_only": True, + } + raise RuntimeError("shared proxy unavailable") + assert self.daemon_mode == "off" + assert kwargs == { + "resume_id": "codex-thread", + "cwd": self.cwd, + } + self.thread_id = "codex-thread" + + +class _CheckpointRecorder: + created: list[_CheckpointRecorder] = [] + + def __init__( + self, + cwd, + state_dir, + sid, + *, + profile_revision, + ) -> None: + self.cwd = cwd + self.state_dir = state_dir + self.sid = sid + self.profile_revision = profile_revision + self.cleanup_calls: list[bool] = [] + self.created.append(self) + + def cleanup(self, *, force: bool = False) -> None: + self.cleanup_calls.append(force) + + +class _FailingCheckpoint(_CheckpointRecorder): + def cleanup(self, *, force: bool = False) -> None: + self.cleanup_calls.append(force) + raise OSError("checkpoint storage unavailable") + + +def _delete_command() -> DeleteSession: + return DeleteSession( + session_id="codex-thread", + engine="codex", + space="code", + cmd_id="delete-1", + client_id="client-1", + ) + + +def _resident(machine, handle: _DeleteHandle): + ctx = _mk_ctx("codex-thread", "codex-thread") + ctx.engine = "codex" + ctx.sdk = handle + ctx.codex_checkpoint = False + machine.sessions = {ctx.key: ctx} + machine.focused_sid = ctx.key + return ctx + + +def _prepare(machine, monkeypatch, *, external: bool = False): + async def is_codex(_sid): + return True + + async def ensure_generation(*_args, **_kwargs): + return True + + async def prime_ownership(_sid, **_kwargs): + return external + + async def list_sessions(_cmd): + return None + + async def forbidden_rpc(*_args, **_kwargs): + raise AssertionError( + "loaded delete must not start a private app-server" + ) + + monkeypatch.setattr(machine, "_is_codex_session", is_codex) + monkeypatch.setattr( + machine, + "_ensure_codex_daemon_generation", + ensure_generation, + ) + monkeypatch.setattr(machine, "_prime_codex_ownership", prime_ownership) + monkeypatch.setattr(machine, "_handle_list_sessions", list_sessions) + monkeypatch.setattr(machine, "_codex_rpc_for_wire", forbidden_rpc) + monkeypatch.setattr(machine, "_codex_rollout_for_wire", lambda _sid: None) + for ctx in machine.sessions.values(): + if ctx.engine != "codex" or ctx.space != "code": + continue + sid = machine._ctx_wire_sid(ctx) + machine._watch.setdefault(sid, { + "engine": "codex", + "scan_complete": True, + "active_external_turns": {}, + }) + + +def test_resident_shared_thread_deletes_before_proxy_disconnect(monkeypatch): + async def run(): + machine, transport = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + machine._codex_session_list_cache = ( + 0.0, + [{"session_id": "codex-thread"}], + (), + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert handle.calls == [ + ("delete", "codex-thread"), + ("disconnect", None), + ] + assert ctx.key not in machine.sessions + assert machine.focused_sid is None + assert machine._codex_session_list_cache is None + assert not [item for item in transport.sent if isinstance(item, Error)] + + asyncio.run(run()) + + +def test_rejected_loaded_delete_preserves_resident_session(monkeypatch): + async def run(): + machine, _ = _mk_machine() + failure = CodexAppServerError({ + "code": -32600, + "message": "thread already has an active writer", + }) + handle = _DeleteHandle(failure) + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_INTERNAL + assert handle.calls == [("delete", "codex-thread")] + assert machine.sessions[ctx.key] is ctx + assert machine.focused_sid == ctx.key + + asyncio.run(run()) + + +def test_unknown_loaded_delete_preserves_thread_when_exact_read_finds_it( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle(ConnectionError("proxy closed")) + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + + async def session_exists(_sid): + return True + + monkeypatch.setattr( + machine, + "_codex_session_exists", + session_exists, + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_INTERNAL + assert handle.calls == [("delete", "codex-thread")] + assert machine.sessions[ctx.key] is ctx + + asyncio.run(run()) + + +def test_unknown_loaded_delete_commits_after_exact_absence( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + rollout = tmp_path / "rollout.jsonl" + rollout.write_text("session\n", encoding="utf-8") + + async def delete_then_disconnect(expected_thread_id): + handle.calls.append(("delete", expected_thread_id)) + rollout.unlink() + raise ConnectionError("proxy closed") + + async def session_exists(_sid): + return False + + handle.delete_thread = delete_then_disconnect + monkeypatch.setattr(machine, "_codex_session_exists", session_exists) + monkeypatch.setattr( + machine, + "_codex_rollout_for_wire", + lambda _sid: str(rollout), + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert handle.calls == [ + ("delete", "codex-thread"), + ("disconnect", None), + ] + assert ctx.key not in machine.sessions + + asyncio.run(run()) + + +def test_unknown_loaded_delete_preserves_unconfirmed_absence( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle(ConnectionError("proxy closed")) + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + rollout = tmp_path / "rollout.jsonl" + rollout.write_text("session\n", encoding="utf-8") + + async def session_exists(_sid): + return False + + monkeypatch.setattr(machine, "_codex_session_exists", session_exists) + monkeypatch.setattr( + machine, + "_codex_rollout_for_wire", + lambda _sid: str(rollout), + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_INTERNAL + assert machine.sessions[ctx.key] is ctx + assert rollout.exists() + + asyncio.run(run()) + + +def test_exact_delete_reconciliation_does_not_use_bounded_catalog( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + calls = [] + + async def read_thread(sid, method, params=None, **_kwargs): + calls.append((sid, method, params)) + return {"thread": {"id": "codex-thread"}} + + monkeypatch.setattr(machine, "_codex_rpc_for_wire", read_thread) + + assert await machine._codex_session_exists("codex-thread") is True + assert calls == [( + "codex-thread", + "thread/read", + { + "threadId": "codex-thread", + "includeTurns": False, + }, + )] + + async def missing_thread(*_args, **_kwargs): + raise CodexRpcRejected( + "codex app-server error -32600: " + "thread not loaded: codex-thread", + code=-32600, + ) + + monkeypatch.setattr(machine, "_codex_rpc_for_wire", missing_thread) + assert await machine._codex_session_exists("codex-thread") is False + + async def other_rejection(*_args, **_kwargs): + raise CodexRpcRejected( + "codex app-server error -32600: invalid request", + code=-32600, + ) + + monkeypatch.setattr(machine, "_codex_rpc_for_wire", other_rejection) + with pytest.raises(CodexRpcRejected, match="invalid request"): + await machine._codex_session_exists("codex-thread") + + asyncio.run(run()) + + +def test_confirmed_descendant_delete_evicts_both_resident_contexts( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("child-thread", "codex-thread"), + ) + root = _resident(machine, root_handle) + child_handle = _DeleteHandle() + child_handle.thread_id = "child-thread" + child = _mk_ctx("child-thread", "child-thread") + child.engine = "codex" + child.sdk = child_handle + child_checkpoint = _CheckpointRecorder( + child.cwd, + tmp_path / "state", + "child-thread", + profile_revision=machine._codex_profile_revision, + ) + child.codex_checkpoint = child_checkpoint + machine.sessions[child.key] = child + _prepare(machine, monkeypatch) + cleanup_calls = [] + machine._codex_controls = _DeleteRecorder( + "controls", + cleanup_calls, + ) + machine._session_plans = _DeleteRecorder("plan", cleanup_calls) + machine._session_presentation = _DeleteRecorder( + "presentation", + cleanup_calls, + ) + machine._session_pins = _PinRecorder(cleanup_calls) + machine._watch = { + "codex-thread": { + "engine": "codex", + "scan_complete": True, + }, + "child-thread": { + "engine": "codex", + "scan_complete": True, + }, + } + + async def session_exists(sid): + assert sid == "child-thread" + return False + + monkeypatch.setattr(machine, "_codex_session_exists", session_exists) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert machine.sessions == {} + assert root_handle.calls == [ + ("delete", "codex-thread"), + ("disconnect", None), + ] + assert child_handle.calls == [("disconnect", None)] + assert child_checkpoint.cleanup_calls == [True] + assert root.key == "codex-thread" + assert cleanup_calls == [ + ("controls", "codex-thread"), + ("plan", "codex-thread"), + ("presentation", "codex-thread"), + ("pin", "codex", "codex-thread", False), + ("controls", "child-thread"), + ("plan", "child-thread"), + ("presentation", "child-thread"), + ("pin", "codex", "child-thread", False), + ] + assert machine._watch == {} + + asyncio.run(run()) + + +def test_resumed_child_delete_cleans_checkpoint_after_cwd_removal( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + machine.cfg.state_dir = tmp_path / "state" + root_handle = _DeleteHandle( + deleted_ids=("child-thread", "codex-thread"), + ) + _resident(machine, root_handle) + child_handle = _DeleteHandle() + child_handle.thread_id = "child-thread" + child = _mk_ctx("child-thread", "child-thread") + child.engine = "codex" + child.sdk = child_handle + machine.sessions[child.key] = child + + original_cwd = tmp_path / "removed-repository" + original_cwd.mkdir() + subprocess.run( + ["git", "-C", str(original_cwd), "init"], + check=True, + capture_output=True, + ) + stale_journal = machine_module.CodexCheckpointJournal( + str(original_cwd), + machine.cfg.state_dir, + "child-thread", + profile_revision=machine._codex_profile_revision, + ) + assert stale_journal.session_dir.exists() + shutil.rmtree(original_cwd) + fallback_cwd = tmp_path / "fallback" + fallback_cwd.mkdir() + child.cwd = str(fallback_cwd) + child.codex_checkpoint = None + root_handle.parents["child-thread"] = "codex-thread" + _prepare(machine, monkeypatch) + + async def session_exists(sid): + assert sid == "child-thread" + return False + + monkeypatch.setattr(machine, "_codex_session_exists", session_exists) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert child.codex_checkpoint is False + assert not stale_journal.session_dir.exists() + + asyncio.run(run()) + + +def test_descendant_checkpoint_oserror_does_not_abort_delete_cleanup( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("child-thread", "codex-thread"), + ) + _resident(machine, root_handle) + child_handle = _DeleteHandle() + child_handle.thread_id = "child-thread" + child = _mk_ctx("child-thread", "child-thread") + child.engine = "codex" + child.sdk = child_handle + child_checkpoint = _FailingCheckpoint( + child.cwd, + tmp_path / "state", + "child-thread", + profile_revision=machine._codex_profile_revision, + ) + child.codex_checkpoint = child_checkpoint + machine.sessions[child.key] = child + root_handle.parents["child-thread"] = "codex-thread" + _prepare(machine, monkeypatch) + cleanup_calls = [] + machine._codex_controls = _DeleteRecorder( + "controls", + cleanup_calls, + ) + machine._session_pins = _PinRecorder(cleanup_calls) + list_calls = [] + + async def session_exists(sid): + assert sid == "child-thread" + return False + + async def list_sessions(cmd): + list_calls.append(cmd.cmd_id) + + monkeypatch.setattr(machine, "_codex_session_exists", session_exists) + monkeypatch.setattr(machine, "_handle_list_sessions", list_sessions) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert machine.sessions == {} + assert child_checkpoint.cleanup_calls == [True] + assert cleanup_calls == [ + ("controls", "codex-thread"), + ("pin", "codex", "codex-thread", False), + ("controls", "child-thread"), + ("pin", "codex", "child-thread", False), + ] + assert machine._watch == {} + assert list_calls == ["delete-1"] + + asyncio.run(run()) + + +def test_confirmed_descendant_delete_evicts_resident_btw_by_native_id( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("child-thread", "codex-thread"), + ) + _resident(machine, root_handle) + child_handle = _DeleteHandle() + child_handle.thread_id = "child-thread" + child = _mk_ctx("btw-child") + child.engine = "codex" + child.btw = True + child.sdk = child_handle + child.codex_checkpoint = False + machine.sessions[child.key] = child + _prepare(machine, monkeypatch) + cleanup_calls = [] + machine._codex_controls = _DeleteRecorder( + "controls", + cleanup_calls, + ) + machine._session_plans = _DeleteRecorder("plan", cleanup_calls) + machine._session_presentation = _DeleteRecorder( + "presentation", + cleanup_calls, + ) + machine._session_pins = _PinRecorder(cleanup_calls) + machine._watch[child.key] = {"engine": "codex"} + machine._codex_sidebar_watches[child.key] = None + owner_calls = [] + + async def external_owner(sid): + owner_calls.append(sid) + return sid == "child-thread" + + monkeypatch.setattr( + machine, + "_codex_delete_external_owner", + external_owner, + ) + + async def session_exists(sid): + assert sid == "child-thread" + return False + + monkeypatch.setattr(machine, "_codex_session_exists", session_exists) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert machine.sessions == {} + assert root_handle.calls == [ + ("delete", "codex-thread"), + ("disconnect", None), + ] + assert child_handle.calls == [("disconnect", None)] + assert owner_calls == ["codex-thread"] + assert cleanup_calls == [ + ("controls", "codex-thread"), + ("plan", "codex-thread"), + ("presentation", "codex-thread"), + ("pin", "codex", "codex-thread", False), + ("controls", "child-thread"), + ("plan", "child-thread"), + ("presentation", "child-thread"), + ("pin", "codex", "child-thread", False), + ("controls", "btw-child"), + ("plan", "btw-child"), + ("presentation", "btw-child"), + ("pin", "codex", "btw-child", False), + ] + assert "btw-child" not in machine._watch + assert "btw-child" not in machine._codex_sidebar_watches + + asyncio.run(run()) + + +def test_confirmed_cold_descendant_delete_cleans_all_metadata( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("codex-thread",), + ) + root_handle.thread_delete_notifications_overflowed = True + _resident(machine, root_handle) + root_handle.delete_candidates = (("cold-child", False),) + root_handle.parents["cold-child"] = "codex-thread" + _prepare(machine, monkeypatch) + cleanup_calls = [] + machine._codex_controls = _DeleteRecorder( + "controls", + cleanup_calls, + ) + machine._session_plans = _DeleteRecorder("plan", cleanup_calls) + machine._session_presentation = _DeleteRecorder( + "presentation", + cleanup_calls, + ) + machine._session_pins = _PinRecorder(cleanup_calls) + machine._watch = { + "codex-thread": {"engine": "codex"}, + "cold-child": { + "engine": "codex", + "scan_complete": True, + "active_external_turns": {}, + }, + } + machine._watch["codex-thread"]["scan_complete"] = True + machine._codex_sidebar_watches["cold-child"] = None + checkpoint_calls = [] + + async def session_exists(sid): + assert sid == "cold-child" + return False + + async def cleanup_checkpoint(sid): + checkpoint_calls.append(sid) + + monkeypatch.setattr( + machine, + "_codex_session_exists", + session_exists, + ) + monkeypatch.setattr( + machine, + "_cleanup_cold_deleted_codex_checkpoint", + cleanup_checkpoint, + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert checkpoint_calls == ["cold-child"] + assert cleanup_calls == [ + ("controls", "codex-thread"), + ("plan", "codex-thread"), + ("presentation", "codex-thread"), + ("pin", "codex", "codex-thread", False), + ("controls", "cold-child"), + ("plan", "cold-child"), + ("presentation", "cold-child"), + ("pin", "codex", "cold-child", False), + ] + assert machine._watch == {} + assert machine._codex_sidebar_watches == {} + + asyncio.run(run()) + + +def test_parent_delete_rejects_queued_resident_descendant(monkeypatch): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("child-thread", "codex-thread"), + ) + _resident(machine, root_handle) + child_handle = _DeleteHandle() + child_handle.thread_id = "child-thread" + child = _mk_ctx("child-thread", "child-thread") + child.engine = "codex" + child.sdk = child_handle + child.codex_checkpoint = False + queued = Query( + sid="child-thread", + prompt="keep this work", + msg_id="queued-child", + delivery="queue", + cmd_id="queue-child", + client_id="client-1", + ) + child.queued_queries.append(queued) + machine.sessions[child.key] = child + root_handle.parents["child-thread"] = "codex-thread" + _prepare(machine, monkeypatch) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert root_handle.parent_calls == ["child-thread"] + assert root_handle.calls == [] + assert machine.sessions[child.key] is child + assert child.queued_queries == [queued] + + asyncio.run(run()) + + +@pytest.mark.parametrize("delivery", ["immediate", "queue", "replace"]) +@pytest.mark.parametrize("target_sid", ["codex-thread", "child-thread"]) +def test_delete_rejects_query_waiting_for_reconciliation( + monkeypatch, + delivery, + target_sid, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("child-thread", "codex-thread"), + ) + root = _resident(machine, root_handle) + child_handle = _DeleteHandle() + child_handle.thread_id = "child-thread" + child = _mk_ctx("child-thread", "child-thread") + child.engine = "codex" + child.sdk = child_handle + child.codex_checkpoint = False + machine.sessions[child.key] = child + root_handle.delete_candidates = (("child-thread", False),) + root_handle.parents["child-thread"] = "codex-thread" + _prepare(machine, monkeypatch) + reconciliation_started = asyncio.Event() + release_reconciliation = asyncio.Event() + + async def session_exists(sid): + assert sid == "child-thread" + reconciliation_started.set() + await release_reconciliation.wait() + return False + + monkeypatch.setattr( + machine, + "_codex_session_exists", + session_exists, + ) + delete_task = asyncio.create_task( + machine._handle_delete_session(_delete_command()) + ) + await reconciliation_started.wait() + query = Query( + sid=target_sid, + prompt="do not lose this prompt", + msg_id=f"racing-{delivery}", + delivery=delivery, + cmd_id=f"query-{delivery}", + client_id="client-1", + ) + query_task = asyncio.create_task(machine._handle_query(query)) + await asyncio.sleep(0) + + assert not query_task.done() + release_reconciliation.set() + assert await delete_task is None + query_result = await query_task + + assert isinstance(query_result, Error) + assert query_result.code == ERR_NOT_RUNNING + assert root.queued_queries == [] + assert child.queued_queries == [] + assert "codex-thread" not in machine.sessions + assert "child-thread" not in machine.sessions + + asyncio.run(run()) + + +@pytest.mark.parametrize("owner_kind", ["private-app", "shared-cli"]) +def test_parent_delete_rejects_externally_owned_descendant( + monkeypatch, + owner_kind, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("child-thread", "codex-thread"), + ) + _resident(machine, root_handle) + child_handle = _DeleteHandle() + child_handle.thread_id = "child-thread" + child = _mk_ctx("child-thread", "child-thread") + child.engine = "codex" + child.sdk = child_handle + child.codex_checkpoint = False + machine.sessions[child.key] = child + root_handle.parents["child-thread"] = "codex-thread" + _prepare(machine, monkeypatch) + ownership_calls = [] + + async def external_owner(sid): + ownership_calls.append(sid) + return owner_kind == "private-app" and sid == "child-thread" + + monkeypatch.setattr( + machine, + "_prime_codex_ownership", + external_owner, + ) + if owner_kind == "shared-cli": + machine._watch["child-thread"] = { + "engine": "codex", + "active_external_turns": {"cli-turn": 1.0}, + } + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert ownership_calls == ["codex-thread", "child-thread"] + assert root_handle.parent_calls == ["child-thread"] + assert root_handle.calls == [] + assert machine.sessions[child.key] is child + + asyncio.run(run()) + + +def test_parent_delete_rejects_active_cold_watched_descendant( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle() + _resident(machine, root_handle) + root_handle.parents["cold-child"] = "codex-thread" + _prepare(machine, monkeypatch) + machine._watch["cold-child"] = { + "engine": "codex", + "active_external_turns": {"app-turn": 1.0}, + } + ownership_calls = [] + + async def prime_ownership(sid, *, extra_handles=()): + ownership_calls.append((sid, extra_handles)) + return False + + monkeypatch.setattr( + machine, + "_prime_codex_ownership", + prime_ownership, + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert root_handle.parent_calls == ["cold-child"] + assert root_handle.calls == [] + assert ownership_calls == [ + ("codex-thread", ()), + ("cold-child", (root_handle,)), + ] + + asyncio.run(run()) + + +def test_parent_delete_rejects_active_cold_catalog_descendant( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle() + _resident(machine, root_handle) + root_handle.delete_candidates = (("cold-child", True),) + root_handle.parents["cold-child"] = "codex-thread" + _prepare(machine, monkeypatch) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert root_handle.parent_calls == ["cold-child"] + assert root_handle.calls == [] + + asyncio.run(run()) + + +def test_parent_delete_ignores_stale_watch_only_thread( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle() + _resident(machine, root_handle) + _prepare(machine, monkeypatch) + machine._watch["stale-thread"] = {"engine": "codex"} + + async def read_parent(thread_id): + root_handle.parent_calls.append(thread_id) + raise CodexAppServerError({ + "code": -32600, + "message": f"thread not loaded: {thread_id}", + }) + + root_handle.read_thread_parent = read_parent + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert root_handle.parent_calls == ["stale-thread"] + assert root_handle.calls == [ + ("delete", "codex-thread"), + ("disconnect", None), + ] + + asyncio.run(run()) + + +def test_unconfirmed_delete_notification_preserves_sibling_context( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + root_handle = _DeleteHandle( + deleted_ids=("unrelated-thread", "codex-thread"), + ) + _resident(machine, root_handle) + sibling_handle = _DeleteHandle() + sibling_handle.thread_id = "unrelated-thread" + sibling = _mk_ctx("unrelated-thread", "unrelated-thread") + sibling.engine = "codex" + sibling.sdk = sibling_handle + sibling.codex_checkpoint = False + machine.sessions[sibling.key] = sibling + _prepare(machine, monkeypatch) + + async def session_exists(sid): + assert sid == "unrelated-thread" + return True + + monkeypatch.setattr(machine, "_codex_session_exists", session_exists) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert machine.sessions == {sibling.key: sibling} + assert sibling_handle.calls == [] + + asyncio.run(run()) + + +def test_multi_profile_delete_keeps_wire_id_out_of_native_rpc( + monkeypatch, + tmp_path: Path, +): + async def run(): + cfg = WrapperConfig() + cfg.state_dir = tmp_path / "state" + cfg.claude_work_root = tmp_path / "work" / "claude" + cfg.codex_work_root = tmp_path / "work" / "codex" + cfg.codex_profiles_json = json.dumps({ + "primary": { + "label": "Primary", + "home": str(tmp_path / "primary"), + "default": True, + }, + "stack": { + "label": "Stack", + "home": str(tmp_path / "stack"), + }, + }) + machine = WrapperMachine(cfg, _Transport()) + handle = _DeleteHandle() + ctx = _mk_ctx("stack@codex-thread", "codex-thread") + ctx.engine = "codex" + ctx.codex_profile_id = "stack" + ctx.sdk = handle + ctx.codex_checkpoint = False + machine.sessions[ctx.key] = ctx + _prepare(machine, monkeypatch) + command = _delete_command().model_copy(update={ + "session_id": "stack@codex-thread", + }) + + result = await machine._handle_delete_session(command) + + assert result is None + assert handle.calls[0] == ("delete", "codex-thread") + assert machine.sessions == {} + + asyncio.run(run()) + + +def test_resident_codex_work_delete_uses_loaded_connection( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + store = machine._work.for_engine("codex") + profile_id = machine._codex_profiles.default.id + record = store.create_session(codex_profile_id=profile_id) + store.bind_session( + record.work_id, + "codex-thread", + codex_profile_id=profile_id, + ) + handle = _DeleteHandle() + ctx = _mk_ctx("codex-thread", "codex-thread") + ctx.engine = "codex" + ctx.space = "work" + ctx.work_id = record.work_id + ctx.codex_profile_id = profile_id + ctx.sdk = handle + machine.sessions[ctx.key] = ctx + _prepare(machine, monkeypatch) + command = DeleteWorkSession( + session_id="codex-thread", + engine="codex", + cmd_id="delete-work-1", + client_id="client-1", + ) + + result = await machine._handle_delete_work_session(command) + + assert result is None + assert handle.calls == [ + ("delete", "codex-thread"), + ("disconnect", None), + ] + assert machine.sessions == {} + assert store.get_by_session( + "codex-thread", + codex_profile_id=profile_id, + ) is None + + asyncio.run(run()) + + +def test_rejected_codex_work_delete_preserves_registry_and_context( + monkeypatch, +): + async def run(): + machine, _ = _mk_machine() + store = machine._work.for_engine("codex") + profile_id = machine._codex_profiles.default.id + record = store.create_session(codex_profile_id=profile_id) + store.bind_session( + record.work_id, + "codex-thread", + codex_profile_id=profile_id, + ) + failure = CodexAppServerError({ + "code": -32600, + "message": "thread already has an active writer", + }) + handle = _DeleteHandle(failure) + ctx = _mk_ctx("codex-thread", "codex-thread") + ctx.engine = "codex" + ctx.space = "work" + ctx.work_id = record.work_id + ctx.codex_profile_id = profile_id + ctx.sdk = handle + machine.sessions[ctx.key] = ctx + _prepare(machine, monkeypatch) + command = DeleteWorkSession( + session_id="codex-thread", + engine="codex", + cmd_id="delete-work-1", + client_id="client-1", + ) + + result = await machine._handle_delete_work_session(command) + + assert isinstance(result, Error) + assert handle.calls == [("delete", "codex-thread")] + assert machine.sessions == {ctx.key: ctx} + assert store.get_by_session( + "codex-thread", + codex_profile_id=profile_id, + ) is not None + + asyncio.run(run()) + + +def test_cold_codex_delete_bypasses_full_resident_pool( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + machine.cfg.cc_cwd = str(tmp_path) + machine.cfg.max_concurrent_sessions = 1 + machine.WATCH_MAX = 1 + active = _mk_ctx("active-thread", "active-thread") + active.state = "running" + machine.sessions = {active.key: active} + machine.focused_sid = active.key + original_watch = {"watched-thread": {"engine": "claude"}} + machine._watch = dict(original_watch) + machine._codex_sidebar_watches["watched-thread"] = None + _prepare(machine, monkeypatch) + _ColdDeleteHandle.created = [] + monkeypatch.setattr( + machine_module, + "CodexHandle", + _ColdDeleteHandle, + ) + monkeypatch.setattr( + machine, + "_codex_cwd_for_wire", + lambda _sid: str(tmp_path), + ) + + async def forbidden_spawn(*_args, **_kwargs): + raise AssertionError("cold deletion must not enter the pool") + + monkeypatch.setattr(machine, "_spawn", forbidden_spawn) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert machine.sessions == {active.key: active} + assert machine.focused_sid == active.key + assert machine._watch == original_watch + assert list(machine._codex_sidebar_watches) == ["watched-thread"] + assert len(_ColdDeleteHandle.created) == 1 + handle = _ColdDeleteHandle.created[0] + assert handle.calls == [ + ( + "connect", + { + "cwd": str(tmp_path), + "control_only": True, + }, + ), + ("delete", "codex-thread"), + ("disconnect", None), + ] + + asyncio.run(run()) + + +def test_cold_codex_delete_rejects_active_catalog_root_without_rollout( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + machine.cfg.cc_cwd = str(tmp_path) + _prepare(machine, monkeypatch) + _ColdDeleteHandle.created = [] + monkeypatch.setattr( + machine_module, + "CodexHandle", + _ColdDeleteHandle, + ) + monkeypatch.setattr( + machine, + "_codex_cwd_for_wire", + lambda _sid: str(tmp_path), + ) + + original_connect = _ColdDeleteHandle.connect + + async def connect_with_active_root(handle, **kwargs): + await original_connect(handle, **kwargs) + handle.delete_candidates = (("codex-thread", True),) + + monkeypatch.setattr( + _ColdDeleteHandle, + "connect", + connect_with_active_root, + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert len(_ColdDeleteHandle.created) == 1 + assert _ColdDeleteHandle.created[0].calls == [ + ( + "connect", + { + "cwd": str(tmp_path), + "control_only": True, + }, + ), + ("disconnect", None), + ] + + asyncio.run(run()) + + +def test_cold_codex_delete_falls_back_to_loaded_stdio( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + machine.cfg.cc_cwd = str(tmp_path) + _prepare(machine, monkeypatch) + _ColdDeleteHandle.created = [] + monkeypatch.setattr( + machine_module, + "CodexHandle", + _FallbackColdDeleteHandle, + ) + monkeypatch.setattr( + machine, + "_codex_cwd_for_wire", + lambda _sid: str(tmp_path), + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert len(_ColdDeleteHandle.created) == 2 + shared, stdio = _ColdDeleteHandle.created + assert shared.daemon_mode == "auto" + assert shared.calls == [ + ( + "connect", + { + "cwd": str(tmp_path), + "control_only": True, + }, + ), + ("disconnect", None), + ] + assert stdio.daemon_mode == "off" + assert stdio.calls == [ + ( + "connect", + { + "resume_id": "codex-thread", + "cwd": str(tmp_path), + }, + ), + ("delete", "codex-thread"), + ("disconnect", None), + ] + + asyncio.run(run()) + + +def test_cold_codex_delete_cleans_checkpoint_without_fallback_cwd( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + machine.cfg.cc_cwd = str(tmp_path) + _prepare(machine, monkeypatch) + _ColdDeleteHandle.created = [] + cleanup_calls = [] + monkeypatch.setattr( + machine_module, + "CodexHandle", + _ColdDeleteHandle, + ) + + def cleanup_checkpoint(state_dir, sid): + cleanup_calls.append((state_dir, sid)) + return 1 + + def forbidden_checkpoint(*_args, **_kwargs): + raise AssertionError( + "cold root cleanup must not depend on fallback cwd" + ) + + monkeypatch.setattr( + machine_module, + "CodexCheckpointJournal", + forbidden_checkpoint, + ) + monkeypatch.setattr( + machine_module, + "cleanup_codex_checkpoint_session", + cleanup_checkpoint, + ) + monkeypatch.setattr( + machine, + "_codex_cwd_for_wire", + lambda _sid: str(tmp_path / "removed-native-cwd"), + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert result is None + assert cleanup_calls == [( + machine.cfg.state_dir, + "codex-thread", + )] + + asyncio.run(run()) + + +@pytest.mark.parametrize("watched", [False, True]) +def test_cold_multi_profile_delete_excludes_its_control_proxy( + monkeypatch, + tmp_path: Path, + watched: bool, +): + async def run(): + cfg = WrapperConfig() + cfg.state_dir = tmp_path / "state" + cfg.cc_cwd = str(tmp_path) + cfg.claude_work_root = tmp_path / "work" / "claude" + cfg.codex_work_root = tmp_path / "work" / "codex" + cfg.codex_profiles_json = json.dumps({ + "primary": { + "label": "Primary", + "home": str(tmp_path / "primary"), + "default": True, + }, + "stack": { + "label": "Stack", + "home": str(tmp_path / "stack"), + }, + }) + machine = WrapperMachine(cfg, _Transport()) + _prepare(machine, monkeypatch) + rollout = tmp_path / "rollout.jsonl" + rollout.write_bytes(b"") + _ColdDeleteHandle.created = [] + monkeypatch.setattr( + machine_module, + "CodexHandle", + _ColdDeleteHandle, + ) + monkeypatch.setattr( + machine, + "_codex_cwd_for_wire", + lambda _sid: str(tmp_path), + ) + monkeypatch.setattr( + machine, + "_codex_rollout_for_wire", + lambda _sid: str(rollout), + ) + if watched: + monkeypatch.setattr( + machine, + "_prime_codex_ownership", + WrapperMachine._prime_codex_ownership.__get__(machine), + ) + machine._watch_session("primary@codex-thread") + proxy = ProcessIdentity(4242, 42) + + def identity(pid, *, parent_pid=None): + assert parent_pid is not None + return proxy if pid == proxy.pid else None + + own_sets = [] + + def holders(paths, own, **_kwargs): + own_set = set(own) + own_sets.append(own_set) + client_proxies = {} if proxy in own_set else {proxy: 1} + return HolderScan( + holders={sid: set() for sid in paths}, + complete=True, + passive_holders={sid: set() for sid in paths}, + client_proxies=client_proxies, + private_holders={sid: set() for sid in paths}, + ) + + tracker_calls = [] + + class UnreadableTracker: + @staticmethod + def bindings(paths, proxies): + tracker_calls.append((dict(paths), dict(proxies))) + return {}, False + + monkeypatch.setattr(machine_module, "process_identity", identity) + monkeypatch.setattr( + machine_module, + "writable_rollout_holders", + holders, + ) + machine._codex_tui_log_trackers = { + profile.id: UnreadableTracker() + for profile in machine._codex_profiles + } + + command = _delete_command().model_copy(update={ + "session_id": "primary@codex-thread", + }) + result = await machine._handle_delete_session(command) + + assert result is None + assert own_sets == [{proxy}] + assert tracker_calls == [ + ({"codex-thread": str(rollout)}, {}), + ] + assert _ColdDeleteHandle.created[0].calls == [ + ( + "connect", + { + "cwd": str(tmp_path), + "control_only": True, + }, + ), + ("delete", "codex-thread"), + ("disconnect", None), + ] + + asyncio.run(run()) + + +def test_cold_codex_delete_preserves_holderless_app_activity( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + machine.cfg.cc_cwd = str(tmp_path) + _prepare(machine, monkeypatch) + rollout = tmp_path / "rollout.jsonl" + rollout.write_bytes((json.dumps({ + "type": "event_msg", + "payload": { + "type": "task_started", + "turn_id": "private-app-turn", + }, + }) + "\n").encode()) + _ColdDeleteHandle.created = [] + monkeypatch.setattr( + machine_module, + "CodexHandle", + _ColdDeleteHandle, + ) + monkeypatch.setattr( + machine, + "_codex_cwd_for_wire", + lambda _sid: str(tmp_path), + ) + monkeypatch.setattr( + machine, + "_codex_rollout_for_wire", + lambda _sid: str(rollout), + ) + + async def no_holders(paths, *, extra_handles=()): + assert paths == {"codex-thread": str(rollout)} + assert extra_handles == (_ColdDeleteHandle.created[0],) + return HolderScan( + holders={"codex-thread": set()}, + complete=True, + passive_holders={"codex-thread": set()}, + private_holders={"codex-thread": set()}, + ) + + monkeypatch.setattr(machine, "_probe_codex_holders", no_holders) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert _ColdDeleteHandle.created[0].calls == [ + ( + "connect", + { + "cwd": str(tmp_path), + "control_only": True, + }, + ), + ("disconnect", None), + ] + + asyncio.run(run()) + + +def test_cold_codex_delete_rejects_active_watched_holder( + monkeypatch, + tmp_path: Path, +): + async def run(): + machine, _ = _mk_machine() + machine.cfg.cc_cwd = str(tmp_path) + _prepare(machine, monkeypatch) + machine._watch["codex-thread"] = { + "engine": "codex", + "scan_complete": False, + "active_external_turns": {}, + "holders": set(), + } + _ColdDeleteHandle.created = [] + monkeypatch.setattr( + machine_module, + "CodexHandle", + _ColdDeleteHandle, + ) + monkeypatch.setattr( + machine, + "_codex_cwd_for_wire", + lambda _sid: str(tmp_path), + ) + + async def active_holder(sid, *, extra_handles=()): + assert sid == "codex-thread" + assert extra_handles == (_ColdDeleteHandle.created[0],) + watch = machine._watch[sid] + watch["scan_complete"] = True + watch["active_external_turns"] = {"turn-1": 1.0} + watch["holders"] = {object()} + return False + + monkeypatch.setattr( + machine, + "_prime_codex_ownership", + active_holder, + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert len(_ColdDeleteHandle.created) == 1 + assert _ColdDeleteHandle.created[0].calls == [ + ( + "connect", + { + "cwd": str(tmp_path), + "control_only": True, + }, + ), + ("disconnect", None), + ] + + asyncio.run(run()) + + +def test_delete_rejects_stale_idle_state_with_active_turn_task(monkeypatch): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + release = asyncio.Event() + ctx.turn_task = asyncio.create_task(release.wait()) + + try: + result = await machine._handle_delete_session(_delete_command()) + finally: + release.set() + await ctx.turn_task + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert handle.calls == [] + assert machine.sessions[ctx.key] is ctx + + asyncio.run(run()) + + +@pytest.mark.parametrize( + "attribute", + ["turn_active", "turn_start_pending"], +) +def test_delete_rejects_sdk_turn_boundaries(monkeypatch, attribute): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + setattr(handle, attribute, True) + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert handle.calls == [] + assert machine.sessions[ctx.key] is ctx + + asyncio.run(run()) + + +def test_delete_rechecks_busy_state_after_control_preflight(monkeypatch): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + + async def state_changes(*_args, **_kwargs): + handle.turn_start_pending = True + return None + + monkeypatch.setattr( + machine, + "_runtime_control_preflight", + state_changes, + ) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert handle.calls == [] + assert machine.sessions[ctx.key] is ctx + + asyncio.run(run()) + + +def test_delete_rejects_private_codex_app_owner(monkeypatch): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch, external=True) + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert handle.calls == [] + assert machine.sessions[ctx.key] is ctx + + asyncio.run(run()) + + +@pytest.mark.parametrize("reason", ["missing-rollout", "watch-cap"]) +def test_delete_fails_closed_without_owner_watch( + monkeypatch, + tmp_path: Path, + reason: str, +): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + machine._watch.pop("codex-thread", None) + + if reason == "watch-cap": + rollout = tmp_path / "rollout.jsonl" + rollout.write_bytes(b"") + monkeypatch.setattr( + machine, + "_codex_rollout_for_wire", + lambda _sid: str(rollout), + ) + machine.WATCH_MAX = 1 + machine._watch["protected-thread"] = { + "engine": "codex", + "external": True, + "scan_complete": True, + } + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert handle.calls == [] + assert machine.sessions[ctx.key] is ctx + assert "codex-thread" not in machine._watch + + asyncio.run(run()) + + +def test_delete_fails_closed_when_owner_scan_is_incomplete(monkeypatch): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + machine._watch["codex-thread"]["scan_complete"] = False + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert handle.calls == [] + assert machine.sessions[ctx.key] is ctx + + asyncio.run(run()) + + +def test_delete_rejects_active_shared_cli_turn(monkeypatch): + async def run(): + machine, _ = _mk_machine() + handle = _DeleteHandle() + ctx = _resident(machine, handle) + _prepare(machine, monkeypatch) + machine._watch["codex-thread"] = { + "engine": "codex", + "active_external_turns": {"cli-turn": 1.0}, + } + + result = await machine._handle_delete_session(_delete_command()) + + assert isinstance(result, Error) + assert result.code == ERR_BUSY + assert handle.calls == [] + assert machine.sessions[ctx.key] is ctx + + asyncio.run(run()) diff --git a/tests/test_codex_worktree_fork.py b/tests/test_codex_worktree_fork.py index ecfea22..1b0d917 100644 --- a/tests/test_codex_worktree_fork.py +++ b/tests/test_codex_worktree_fork.py @@ -1,6 +1,8 @@ from __future__ import annotations import asyncio +import json +import time from types import SimpleNamespace import pytest @@ -13,6 +15,7 @@ is_downstream, serialize, ) +from cc_remote.codex_profiles import CodexProfile from cc_remote.wrapper import machine as machine_module from cc_remote.wrapper.codex_worktrees import WorktreeSpec from cc_remote.wrapper.codex_forks import CodexForkJournal @@ -77,6 +80,309 @@ def test_worktree_fork_protocol_roundtrips_as_control_messages(): assert ordinary.last_turn_id == "turn-2" +def test_same_cwd_fork_invalidates_catalog_before_publication(monkeypatch): + async def run(): + machine, transport = _mk_machine() + machine._codex_forks.begin( + "request-1", + "parent", + "turn-2", + "/repo/component", + ) + machine._codex_session_list_cache = ( + 0.0, + [{"session_id": "parent"}], + (), + ) + initial_epoch = machine._codex_session_list_epoch + sent = transport.send + + async def assert_fresh_catalog(message): + if message.type == "session_forked": + assert machine._codex_session_list_cache is None + assert machine._codex_session_list_epoch == initial_epoch + 1 + await sent(message) + + async def list_sessions(_cmd): + assert machine._codex_session_list_cache is None + + monkeypatch.setattr(transport, "send", assert_fresh_catalog) + monkeypatch.setattr(machine, "_list_codex_sessions", list_sessions) + + result = await machine._finish_same_cwd_fork( + _command(last_turn_id="turn-2"), + "parent", + "/repo/component", + "forked-thread", + ) + + assert result.session_id == "forked-thread" + assert [message.type for message in transport.sent] == [ + "session_forked", + ] + + asyncio.run(run()) + + +def test_worktree_fork_invalidates_catalog_before_publication(monkeypatch): + async def run(): + machine, transport = _mk_machine() + spec = _spec() + machine._codex_forks.begin( + "request-1", + "parent", + "cc-remote-worktree-head", + spec.cwd, + target="worktree", + ) + machine._codex_session_list_cache = ( + 0.0, + [{"session_id": "parent"}], + (), + ) + initial_epoch = machine._codex_session_list_epoch + sent = transport.send + + async def assert_fresh_catalog(message): + if message.type == "session_forked": + assert machine._codex_session_list_cache is None + assert machine._codex_session_list_epoch == initial_epoch + 1 + await sent(message) + + async def list_sessions(_cmd): + assert machine._codex_session_list_cache is None + + monkeypatch.setattr(transport, "send", assert_fresh_catalog) + monkeypatch.setattr(machine, "_list_codex_sessions", list_sessions) + + result = await machine._finish_worktree_fork( + _command(name=""), + "parent", + spec, + "forked-thread", + "cc-remote-fork:request-1", + ) + + assert result.session_id == "forked-thread" + assert [message.type for message in transport.sent] == [ + "session_forked", + ] + + asyncio.run(run()) + + +def test_same_cwd_fork_refresh_includes_child_after_fresh_parent_cache( + monkeypatch, tmp_path, +): + async def run(): + machine, _ = _mk_machine() + machine._codex_forks.begin( + "request-1", + "parent", + "turn-2", + "/repo/component", + ) + machine._codex_session_list_cache = ( + time.monotonic(), + [{"session_id": "parent"}], + (), + ) + published = [] + rollout = tmp_path / ( + "rollout-2026-08-13T09-06-10-forked-thread.jsonl") + rollout.write_text(json.dumps({ + "type": "session_meta", + "payload": { + "id": "forked-thread", + "cwd": "/repo/component", + "thread_source": "cc-remote-fork:request-1", + "forked_from_id": "parent", + }, + }) + "\n") + + async def read_catalog(): + return ([ + { + "session_id": "parent", + "native_session_id": "parent", + "codex_profile_id": "primary", + "codex_profile_label": "Primary", + "summary": "Parent task", + "cwd": "/repo/component", + "last_modified": "1", + }, + ], ()) + + async def send_list(_cmd, raw, **_kwargs): + published.append(raw) + + def find_rollouts(candidates): + assert [item[2] for item in candidates] == ["forked-thread"] + return {("primary", "forked-thread"): (str(rollout), False)} + + monkeypatch.setattr( + machine, + "_read_codex_profile_catalog", + read_catalog, + ) + monkeypatch.setattr( + machine, + "_find_completed_codex_fork_rollouts", + find_rollouts, + ) + monkeypatch.setattr(machine, "_send_codex_session_list", send_list) + + await machine._finish_same_cwd_fork( + _command(last_turn_id="turn-2"), + "parent", + "/repo/component", + "forked-thread", + ) + + assert len(published) == 1 + by_id = {row["session_id"]: row for row in published[0]} + assert set(by_id) == {"parent", "forked-thread"} + assert by_id["forked-thread"]["summary"] == "Parent task (fork)" + assert by_id["forked-thread"]["forked_from_id"] == "parent" + assert by_id["forked-thread"]["status"] == "notLoaded" + + asyncio.run(run()) + + +def test_completed_fork_without_rollout_is_not_restored(monkeypatch): + async def run(): + machine, _ = _mk_machine() + machine._codex_forks.begin( + "request-1", "parent", "turn-2", "/repo/component") + machine._codex_forks.complete("request-1", "deleted-child") + + async def read_catalog(): + return ([{ + "session_id": "parent", + "native_session_id": "parent", + "codex_profile_id": "primary", + }], ()) + + monkeypatch.setattr( + machine, "_read_codex_profile_catalog", read_catalog) + monkeypatch.setattr( + machine, + "_find_completed_codex_fork_rollouts", + lambda _candidates: {}, + ) + + rows = await machine._refresh_codex_session_catalog() + + assert [row["session_id"] for row in rows] == ["parent"] + + asyncio.run(run()) + + +def test_renamed_fork_keeps_title_while_native_catalog_omits_it( + monkeypatch, tmp_path, +): + async def run(): + machine, _ = _mk_machine() + machine._codex_forks.begin( + "request-1", "parent", "turn-2", "/repo/component") + machine._codex_forks.complete("request-1", "forked-thread") + rollout = tmp_path / ( + "rollout-2026-08-13T09-06-10-forked-thread.jsonl") + rollout.write_text(json.dumps({ + "type": "session_meta", + "payload": { + "id": "forked-thread", + "cwd": "/repo/component", + "thread_source": "cc-remote-fork:request-1", + "forked_from_id": "parent", + }, + }) + "\n") + published = [] + rpc_calls = [] + + async def is_codex(_sid): + return True + + async def rpc(sid, method, params, **_kwargs): + rpc_calls.append((sid, method, params)) + return {} + + async def read_catalog(): + return ([{ + "session_id": "parent", + "native_session_id": "parent", + "codex_profile_id": "primary", + "codex_profile_label": "Primary", + "summary": "Parent task", + }], ()) + + async def send_list(_cmd, raw, **_kwargs): + published.append(raw) + + monkeypatch.setattr(machine, "_is_codex_session", is_codex) + monkeypatch.setattr(machine, "_codex_rpc_for_wire", rpc) + monkeypatch.setattr( + machine, "_read_codex_profile_catalog", read_catalog) + monkeypatch.setattr( + machine, + "_find_completed_codex_fork_rollouts", + lambda _candidates: { + ("primary", "forked-thread"): (str(rollout), False), + }, + ) + monkeypatch.setattr(machine, "_send_codex_session_list", send_list) + + await machine._handle_rename_session(SimpleNamespace( + session_id="forked-thread", + title="Renamed fork", + engine="codex", + space="code", + )) + + assert rpc_calls == [( + "forked-thread", + "thread/name/set", + {"threadId": "forked-thread", "name": "Renamed fork"}, + )] + by_id = {row["session_id"]: row for row in published[-1]} + assert by_id["forked-thread"]["summary"] == "Renamed fork" + assert machine._codex_forks.completed_results(1)[0][ + "title"] == "Renamed fork" + + asyncio.run(run()) + + +def test_completed_fork_rollout_scan_is_batched_and_archive_aware(tmp_path): + machine, _ = _mk_machine() + profile = CodexProfile( + id="primary", + label="Primary", + home=tmp_path, + is_default=True, + ) + active = tmp_path / "sessions" / "2026" / "08" / "13" + archived = tmp_path / "archived_sessions" + active.mkdir(parents=True) + archived.mkdir() + active_child = active / ( + "rollout-2026-08-13T09-06-10-child-active.jsonl") + archived_child = archived / ( + "rollout-2026-08-12T09-06-10-child-archived.jsonl") + active_child.write_text("{}\n") + archived_child.write_text("{}\n") + candidates = [ + ({}, profile, "child-active", "parent", "parent"), + ({}, profile, "child-archived", "parent", "parent"), + ] + + found = machine._find_completed_codex_fork_rollouts(candidates) + + assert found[("primary", "child-active")] == ( + str(active_child.resolve()), False) + assert found[("primary", "child-archived")] == ( + str(archived_child.resolve()), True) + + def test_codex_same_cwd_fork_uses_selected_turn_and_is_durable(monkeypatch): async def run(): machine, _ = _mk_machine() diff --git a/web/tests/history-browser.spec.ts b/web/tests/history-browser.spec.ts index 9d51cbf..81cacc1 100644 --- a/web/tests/history-browser.spec.ts +++ b/web/tests/history-browser.spec.ts @@ -1294,7 +1294,17 @@ test("older history becoming available during a wheel gesture is restored once", expect(pending.id).toBe(before.id); expect(Math.abs(pending.offset - before.offset)).toBeLessThan(2); await expect(page.getByTestId("load-count")).toHaveText("0"); - await page.getByTestId("reveal-older-history").click(); + await viewport.evaluate((node) => { + node.dispatchEvent(new WheelEvent("wheel", { + bubbles: true, + deltaY: -80, + })); + const reveal = document.querySelector( + '[data-testid="reveal-older-history"]', + ); + if (!reveal) throw new Error("history reveal control is missing"); + reveal.click(); + }); await expect(page.getByTestId("load-count")).toHaveText("1"); await expect(page.locator('[data-turn-id="n8"]')).toBeAttached(); await page.waitForTimeout(250); @@ -3365,7 +3375,7 @@ test("desktop text selection keeps its original virtual turn while edge-dragging viewportBox.y + viewportBox.height - 2, { steps: 20 }, ); - for (let step = 0; step < 24; step += 1) { + for (let step = 0; step < 12; step += 1) { await page.mouse.wheel(0, 220); await page.mouse.move( viewportBox.x + viewportBox.width - 48 + (step % 2), @@ -3718,7 +3728,14 @@ test("iOS pointercancel releases process interactions and output following", asy node.scrollHeight - node.scrollTop - node.clientHeight, )).toBeLessThan(2); - await header.click(); + await header.evaluate((node) => { + const target = node as HTMLElement; + target.dispatchEvent(new MouseEvent("click", { + bubbles: true, + cancelable: true, + detail: 1, + })); + }); await expect(header).toHaveAttribute("aria-expanded", "true"); const reasoning = timeline.locator("details.process-reasoning"); await dispatchCancelledTouchTap(