cleanup-subagents

gistfile1.txt

#!/usr/bin/env python3
"""Safely find and remove closed Codex subagent session JSONL files."""

import argparse
import json
import os
import stat
import subprocess
import sys
from pathlib import Path


def human_size(byte_count):
    units = ("B", "KiB", "MiB", "GiB", "TiB")
    value = float(byte_count)
    unit = units[0]
    for unit in units:
        if value < 1024 or unit == units[-1]:
            break
        value /= 1024
    if value < 10 and unit != "B":
        return f"{value:.2f} {unit}"
    return f"{value:.1f} {unit}"


def open_session_files(root, required):
    try:
        result = subprocess.run(
            ["lsof", "-Fn"],
            capture_output=True,
            text=True,
            check=False,
        )
    except FileNotFoundError:
        if required:
            raise RuntimeError("Cannot safely delete because lsof is not installed")
        return set(), "lsof is not installed; open-file status is unavailable"

    prefix = root + os.sep
    paths = {
        line[1:].strip()
        for line in result.stdout.splitlines()
        if line.startswith("n")
        and line[1:].strip().startswith(prefix)
        and line[1:].strip().endswith(".jsonl")
    }

    if required and result.returncode != 0 and not paths:
        detail = result.stderr.strip() or f"lsof returned status {result.returncode}"
        raise RuntimeError(f"Cannot safely detect active session files: {detail}")

    return paths, result.stderr.strip()


def safe_regular_file(path, root):
    try:
        file_stat = os.lstat(path)
        if not stat.S_ISREG(file_stat.st_mode) or stat.S_ISLNK(file_stat.st_mode):
            return False
        real_path = os.path.realpath(path)
        return os.path.commonpath((root, real_path)) == root
    except (FileNotFoundError, PermissionError, ValueError):
        return False


def verified_subagent_session(path):
    try:
        with open(path, "rb") as session_file:
            first_line = session_file.readline()
        if not first_line:
            return False
        record = json.loads(first_line)
        source = record.get("payload", {}).get("source")
        return (
            record.get("type") == "session_meta"
            and isinstance(source, dict)
            and "subagent" in source
        )
    except (OSError, UnicodeDecodeError, json.JSONDecodeError, AttributeError):
        return False


def iter_jsonl_files(root):
    for directory, directory_names, filenames in os.walk(root, followlinks=False):
        directory_names[:] = [
            name
            for name in directory_names
            if not os.path.islink(os.path.join(directory, name))
        ]
        for filename in filenames:
            if filename.endswith(".jsonl"):
                yield os.path.join(directory, filename)


def allocated_bytes(file_stat):
    return getattr(file_stat, "st_blocks", 0) * 512 or file_stat.st_size


def scan_sessions(root, open_files):
    matches = []
    unreadable = 0

    for path in iter_jsonl_files(root):
        if not safe_regular_file(path, root):
            unreadable += 1
            continue
        if not verified_subagent_session(path):
            continue

        try:
            file_stat = os.stat(path)
        except (FileNotFoundError, PermissionError):
            unreadable += 1
            continue

        matches.append(
            {
                "path": path,
                "size": file_stat.st_size,
                "allocated": allocated_bytes(file_stat),
                "open": path in open_files,
            }
        )

    return matches, unreadable


def prune_empty_directories(root):
    removed = 0
    for directory, _, _ in os.walk(root, topdown=False, followlinks=False):
        if directory == root or os.path.islink(directory):
            continue
        try:
            os.rmdir(directory)
            removed += 1
        except (FileNotFoundError, PermissionError, OSError):
            pass
    return removed


def parse_args():
    default_root = os.path.join(Path.home(), ".codex", "sessions")
    parser = argparse.ArgumentParser(
        description=(
            "Find Codex JSONL files whose first record identifies a subagent session. "
            "Dry-run is the default; currently open files are always preserved."
        )
    )
    parser.add_argument(
        "--delete",
        action="store_true",
        help="permanently delete verified, closed subagent sessions",
    )
    parser.add_argument(
        "--yes",
        action="store_true",
        help="skip the interactive DELETE confirmation (requires --delete)",
    )
    parser.add_argument(
        "--list",
        action="store_true",
        help="list every matching file and whether it is open",
    )
    parser.add_argument(
        "--prune-empty-dirs",
        action="store_true",
        help="remove empty date directories after deletion",
    )
    parser.add_argument(
        "--root",
        default=default_root,
        metavar="DIR",
        help=f"session root (default: {default_root})",
    )
    return parser.parse_args()


def main():
    args = parse_args()
    if args.yes and not args.delete:
        print("--yes only has an effect together with --delete", file=sys.stderr)
        return 2

    requested_root = os.path.expanduser(args.root)
    if not os.path.isdir(requested_root):
        print(f"Session directory does not exist: {requested_root}", file=sys.stderr)
        return 1
    root = os.path.realpath(requested_root)

    try:
        open_files, lsof_warning = open_session_files(root, required=args.delete)
    except RuntimeError as error:
        print(error, file=sys.stderr)
        return 1

    if lsof_warning:
        print(f"Warning: {lsof_warning}", file=sys.stderr)

    matches, unreadable = scan_sessions(root, open_files)
    closed = [entry for entry in matches if not entry["open"]]
    active = [entry for entry in matches if entry["open"]]

    logical_total = sum(entry["size"] for entry in matches)
    allocated_total = sum(entry["allocated"] for entry in matches)
    deletable_total = sum(entry["allocated"] for entry in closed)
    active_total = sum(entry["allocated"] for entry in active)

    print(f"Codex session root: {root}")
    print(
        f"Verified subagent sessions: {len(matches)} "
        f"({human_size(logical_total)} logical, {human_size(allocated_total)} allocated)"
    )
    print(f"Closed/deletable: {len(closed)} ({human_size(deletable_total)} allocated)")
    print(f"Open/preserved: {len(active)} ({human_size(active_total)} allocated)")
    if unreadable:
        print(f"Unreadable/raced files: {unreadable}")

    if args.list:
        for entry in sorted(matches, key=lambda item: item["path"]):
            state_label = "OPEN-PRESERVED" if entry["open"] else "CLOSED"
            print(
                f"{state_label:<14} {human_size(entry['allocated']):>10}  "
                f"{entry['path']}"
            )

    if not args.delete:
        print("\nDry run only. Run with --delete to remove the closed files.")
        return 0

    if not closed:
        print("Nothing to delete.")
        return 0

    if not args.yes:
        print(
            f"\nThis permanently deletes {len(closed)} closed Codex subagent "
            "session files."
        )
        print(
            "They will not be moved to Trash and those subagent threads may no "
            "longer be resumable."
        )
        confirmation = input(f"Type DELETE {len(closed)} to continue: ").strip()
        if confirmation != f"DELETE {len(closed)}":
            print("Aborted; nothing was deleted.")
            return 2

    try:
        open_files, _ = open_session_files(root, required=True)
    except RuntimeError as error:
        print(error, file=sys.stderr)
        return 1

    deleted = 0
    deleted_bytes = 0
    skipped_open = 0
    skipped_changed = 0
    errors = []

    for entry in closed:
        path = entry["path"]
        if path in open_files:
            skipped_open += 1
            continue
        if not safe_regular_file(path, root) or not verified_subagent_session(path):
            skipped_changed += 1
            continue

        try:
            file_stat = os.stat(path)
            size = allocated_bytes(file_stat)
            os.unlink(path)
            deleted += 1
            deleted_bytes += size
        except OSError as error:
            errors.append((path, error))

    pruned = prune_empty_directories(root) if args.prune_empty_dirs else 0

    print(f"\nDeleted: {deleted} files ({human_size(deleted_bytes)} allocated)")
    if skipped_open:
        print(f"Newly open/preserved: {skipped_open}")
    if skipped_changed:
        print(f"Changed or no longer verified/preserved: {skipped_changed}")
    if args.prune_empty_dirs:
        print(f"Empty directories removed: {pruned}")
    print(f"Errors: {len(errors)}")
    for path, error in errors[:20]:
        print(f"{type(error).__name__}: {error}: {path}", file=sys.stderr)

    return 0 if not errors else 1


if __name__ == "__main__":
    sys.exit(main())