fix(codex): keep fork and delete state consistent

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],

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]: