diff --git a/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-ARCHIVE-EXECUTOR-PACKET.md b/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-ARCHIVE-EXECUTOR-PACKET.md index 0ecb0db..34d98fe 100644 --- a/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-ARCHIVE-EXECUTOR-PACKET.md +++ b/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-ARCHIVE-EXECUTOR-PACKET.md @@ -36,6 +36,12 @@ Focused check: python3 tools/archive_codex_inactive_threads.py --check ``` +Native Codex retention probe: + +```bash +python3 tools/probe_codex_native_retention.py --check +``` + Approved archive-only execution: ```bash @@ -56,6 +62,8 @@ python3 tools/archive_codex_inactive_threads.py --execute --approval-token "I ap - SQLite checkpoint or vacuum is blocked; - Core source mutation is blocked. +Installed Codex `0.134.0` advertises `--ephemeral` prevention but no native cleanup/archive/retention command. Cached latest is `0.137.0`; update and re-probe before approved archive execution if latest native behavior should be considered. + ## Backup Before any approved archive update, the executor backs up: @@ -86,4 +94,5 @@ Use this executor only after JP gives the exact archive-only approval token. Kee ## New Issues - must-fix: obtain exact approval token before running `--execute`. +- follow-up: decide whether to update Codex and re-run the native retention probe before archive-only execution. - follow-up: after archive-only execution, re-run retention planner and decide whether deletion is still worth separate approval. diff --git a/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md b/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md index ad7b13b..a62c52d 100644 --- a/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md +++ b/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md @@ -42,6 +42,18 @@ The planner classifies: - top log pressure targets; - approval boundaries. +## Native Codex Probe + +`python3 tools/probe_codex_native_retention.py` checks installed Codex CLI help, feature flags, and local version cache. It does not read transcript bodies, thread text fields, titles, previews, secrets, raw messages, or mutate Codex state. + +Probe result on 2026-06-04: + +- installed Codex version: `0.134.0`; +- cached latest Codex version: `0.137.0`; +- native cleanup/archive/retention command advertised by installed CLI: false; +- prevention flag advertised: `codex exec --ephemeral`; +- decision point: update Codex and re-run the probe before custom archive mutation if latest native behavior must be considered. + ## Policy 1. Prevention default: use `codex exec --ephemeral` for disposable non-interactive worker runs. @@ -70,4 +82,4 @@ Next safe action is to ask for archive-only approval. Delete and vacuum stay sep - must-fix: obtain explicit archive-only approval before any `threads.archived` update. - must-fix: obtain separate destructive approval before session deletion, log deletion, checkpoint, or vacuum. -- follow-up: check native Codex retention support before custom mutation. +- follow-up: native Codex retention support is checked for installed `0.134.0`; update/re-probe `0.137.0` before custom mutation if latest native behavior should be considered. diff --git a/tools/probe_codex_native_retention.py b/tools/probe_codex_native_retention.py new file mode 100644 index 0000000..f8d1750 --- /dev/null +++ b/tools/probe_codex_native_retention.py @@ -0,0 +1,200 @@ +#!/usr/bin/env python3 +"""Probe installed Codex CLI retention support without mutating Codex state.""" + +from __future__ import annotations + +import argparse +import json +import os +import re +import shutil +import subprocess +from pathlib import Path + +SCHEMA_VERSION = "cto-codex-native-retention-probe.v1" +RETENTION_TERMS = ("retention", "cleanup", "prune", "archive") + + +def run_help(codex_bin: str, args: list[str]) -> dict[str, object]: + command = [codex_bin, *args] + try: + completed = subprocess.run( + command, + check=False, + capture_output=True, + text=True, + timeout=10, + ) + except Exception as exc: + return { + "command": command, + "returncode": None, + "stdout": "", + "stderr": str(exc), + "ok": False, + } + return { + "command": command, + "returncode": completed.returncode, + "stdout": completed.stdout, + "stderr": completed.stderr, + "ok": completed.returncode == 0, + } + + +def parse_version(output: str) -> str | None: + match = re.search(r"(\d+\.\d+\.\d+)", output) + return match.group(1) if match else None + + +def read_version_cache(codex_home: Path) -> dict[str, object]: + path = codex_home / "version.json" + if not path.exists(): + return {"path": str(path), "present": False} + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except json.JSONDecodeError as exc: + return {"path": str(path), "present": True, "parse_error": str(exc)} + return { + "path": str(path), + "present": True, + "latest_version": payload.get("latest_version"), + "last_checked_at": payload.get("last_checked_at"), + "dismissed_version": payload.get("dismissed_version"), + } + + +def term_hits(text: str) -> list[str]: + lowered = text.lower() + return [term for term in RETENTION_TERMS if term in lowered] + + +def build_probe(codex_home: Path, codex_bin: str | None) -> dict[str, object]: + resolved_bin = codex_bin or shutil.which("codex") + help_results: list[dict[str, object]] = [] + installed_version = None + if resolved_bin: + version_result = run_help(resolved_bin, ["--version"]) + installed_version = parse_version(str(version_result.get("stdout", ""))) + help_results = [ + run_help(resolved_bin, ["--help"]), + run_help(resolved_bin, ["exec", "--help"]), + run_help(resolved_bin, ["exec", "resume", "--help"]), + run_help(resolved_bin, ["resume", "--help"]), + run_help(resolved_bin, ["features", "list"]), + ] + combined_help = "\n".join(str(result.get("stdout", "")) for result in help_results) + version_cache = read_version_cache(codex_home) + latest_version = version_cache.get("latest_version") + update_available = bool(installed_version and latest_version and installed_version != latest_version) + retention_hits = term_hits(combined_help) + return { + "schema_version": SCHEMA_VERSION, + "codex_home": str(codex_home), + "codex_bin": resolved_bin, + "metadata_only": True, + "mutation_performed": False, + "raw_transcript_bodies_read": False, + "raw_thread_text_fields_read": False, + "state_db_mutation": False, + "session_jsonl_deleted": False, + "logs_deleted_or_truncated": False, + "installed_version": installed_version, + "version_cache": version_cache, + "update_available": update_available, + "native_retention_cleanup_command_advertised": bool(retention_hits), + "native_retention_terms_seen": retention_hits, + "ephemeral_prevention_advertised": "--ephemeral" in combined_help, + "help_surfaces_checked": [ + "codex --help", + "codex exec --help", + "codex exec resume --help", + "codex resume --help", + "codex features list", + ], + "decision": ( + "Installed Codex advertises ephemeral prevention but no native retention cleanup command." + if not retention_hits + else "Installed Codex advertises possible retention cleanup terms; inspect before custom mutation." + ), + "recommended_next": ( + "Decide whether to update Codex and re-run this probe before archive-only execution." + if update_available + else "Use guarded archive executor only with exact approval token." + ), + "false_effects": { + "codex_update": False, + "archive_threads": False, + "delete_session_jsonl": False, + "delete_logs": False, + "sqlite_checkpoint_or_vacuum": False, + "raw_transcript_body_read": False, + "raw_thread_text_field_read": False, + "core_source_mutation": False, + }, + } + + +def validate_probe(probe: dict[str, object]) -> list[str]: + errors: list[str] = [] + if probe.get("schema_version") != SCHEMA_VERSION: + errors.append("schema_version_invalid") + for field in [ + "metadata_only", + "mutation_performed", + "raw_transcript_bodies_read", + "raw_thread_text_fields_read", + "state_db_mutation", + "session_jsonl_deleted", + "logs_deleted_or_truncated", + ]: + expected = field == "metadata_only" + if probe.get(field) is not expected: + errors.append(f"{field}_invalid") + if not probe.get("codex_bin"): + errors.append("codex_bin_missing") + if not probe.get("installed_version"): + errors.append("installed_version_missing") + if probe.get("ephemeral_prevention_advertised") is not True: + errors.append("ephemeral_prevention_not_advertised") + false_effects = probe.get("false_effects") + if not isinstance(false_effects, dict): + errors.append("false_effects_missing") + else: + for key, value in false_effects.items(): + if value is not False: + errors.append(f"false_effect_not_false:{key}") + return errors + + +def main() -> int: + parser = argparse.ArgumentParser() + parser.add_argument("--codex-home", default=os.environ.get("CODEX_HOME", str(Path.home() / ".codex"))) + parser.add_argument("--codex-bin") + parser.add_argument("--check", action="store_true") + args = parser.parse_args() + + probe = build_probe(Path(args.codex_home).expanduser(), args.codex_bin) + errors = validate_probe(probe) + if args.check: + print( + json.dumps( + { + "ok": not errors, + "validator": "cto-codex-native-retention-probe", + "errors": errors, + "warnings": [], + }, + indent=2, + sort_keys=True, + ) + ) + return 0 if not errors else 1 + probe["ok"] = not errors + probe["errors"] = errors + print(json.dumps(probe, indent=2, sort_keys=True)) + return 0 if not errors else 1 + + +if __name__ == "__main__": + raise SystemExit(main())