diff --git a/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md b/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md index a62c52d..ebcd7d3 100644 --- a/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md +++ b/.sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md @@ -63,6 +63,18 @@ Probe result on 2026-06-04: 5. Phase 3: delete archived session JSONL only after separate destructive approval. 6. Phase 4: delete/truncate logs and checkpoint/vacuum only after Codex is stopped and destructive approval is explicit. +## Prevention Helper + +`python3 tools/codex_ephemeral_exec.py` builds disposable worker commands as `codex exec --ephemeral`. It supports `--check` and `--print-command` validation paths that do not run Codex. + +Example dry command: + +```bash +python3 tools/codex_ephemeral_exec.py --print-command -C /path/to/repo "summarize current git status" +``` + +The helper is prevention only. It does not archive threads, delete JSONL, truncate logs, checkpoint, vacuum, read transcript bodies, or mutate Core. + ## Approval Boundary Blocked without explicit operator approval: @@ -82,4 +94,5 @@ 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: use the ephemeral exec helper for disposable non-interactive worker runs. - 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/WORKBOARD.yaml b/WORKBOARD.yaml index 7fde9a6..dd2aa90 100644 --- a/WORKBOARD.yaml +++ b/WORKBOARD.yaml @@ -476,3 +476,8 @@ items: status: validated source: .sot/03-PROTOCOLS/CTO-CODEX-RETENTION-ARCHIVE-EXECUTOR-PACKET.md owner: "" + - id: CTO-WORK-096 + title: Codex Ephemeral Exec Helper + status: validated + source: .sot/03-PROTOCOLS/CTO-CODEX-RETENTION-POLICY-PACKET.md + owner: "" diff --git a/tools/codex_ephemeral_exec.py b/tools/codex_ephemeral_exec.py new file mode 100644 index 0000000..93d885f --- /dev/null +++ b/tools/codex_ephemeral_exec.py @@ -0,0 +1,163 @@ +#!/usr/bin/env python3 +"""Run disposable Codex exec tasks with session persistence disabled.""" + +from __future__ import annotations + +import argparse +import json +import shutil +import subprocess +from pathlib import Path + +SCHEMA_VERSION = "cto-codex-ephemeral-exec-helper.v1" +WORK_ITEM_ID = "CTO-WORK-096" + + +def build_command(args: argparse.Namespace) -> list[str]: + codex_bin = args.codex_bin or shutil.which("codex") or "codex" + command = [codex_bin, "exec", "--ephemeral"] + if args.cwd: + command.extend(["-C", str(Path(args.cwd).expanduser())]) + if args.model: + command.extend(["-m", args.model]) + if args.json_events: + command.append("--json") + if args.sandbox: + command.extend(["--sandbox", args.sandbox]) + if args.ask_for_approval: + command.extend(["--ask-for-approval", args.ask_for_approval]) + if args.output_last_message: + command.extend(["--output-last-message", str(Path(args.output_last_message).expanduser())]) + command.extend(args.prompt) + return command + + +def command_report(command: list[str], *, execute_requested: bool) -> dict[str, object]: + return { + "schema_version": SCHEMA_VERSION, + "work_item_id": WORK_ITEM_ID, + "command": command, + "execute_requested": execute_requested, + "ephemeral_required": True, + "ephemeral_present": "--ephemeral" in command, + "mutation_performed_by_helper": False, + "session_persistence_requested": 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, + "false_effects": { + "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_report(report: dict[str, object]) -> list[str]: + errors: list[str] = [] + command = report.get("command") + if report.get("schema_version") != SCHEMA_VERSION: + errors.append("schema_version_invalid") + if report.get("work_item_id") != WORK_ITEM_ID: + errors.append("work_item_id_invalid") + if not isinstance(command, list) or len(command) < 3: + errors.append("command_invalid") + return errors + if "exec" not in command[:3]: + errors.append("codex_exec_missing") + if "--ephemeral" not in command: + errors.append("ephemeral_missing") + for forbidden in [ + "--dangerously-bypass-approvals-and-sandbox", + "--dangerously-bypass-hook-trust", + ]: + if forbidden in command: + errors.append(f"forbidden_flag:{forbidden}") + for field in [ + "mutation_performed_by_helper", + "session_persistence_requested", + "raw_transcript_bodies_read", + "raw_thread_text_fields_read", + "state_db_mutation", + "session_jsonl_deleted", + "logs_deleted_or_truncated", + ]: + if report.get(field) is not False: + errors.append(f"{field}_invalid") + false_effects = report.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-bin") + parser.add_argument("-C", "--cwd") + parser.add_argument("-m", "--model") + parser.add_argument("--json", dest="json_events", action="store_true") + parser.add_argument("--sandbox", choices=["read-only", "workspace-write"]) + parser.add_argument("--ask-for-approval", choices=["untrusted", "on-request", "never"]) + parser.add_argument("-o", "--output-last-message") + parser.add_argument("--print-command", action="store_true") + parser.add_argument("--check", action="store_true") + parser.add_argument("prompt", nargs=argparse.REMAINDER) + args = parser.parse_args() + + command = build_command(args) + report = command_report(command, execute_requested=not (args.print_command or args.check)) + errors = validate_report(report) + + if args.check: + print( + json.dumps( + { + "ok": not errors, + "validator": "cto-codex-ephemeral-exec-helper", + "errors": errors, + "warnings": [], + }, + indent=2, + sort_keys=True, + ) + ) + return 0 if not errors else 1 + + if args.print_command: + report["ok"] = not errors + report["errors"] = errors + print(json.dumps(report, indent=2, sort_keys=True)) + return 0 if not errors else 1 + + if not args.prompt: + print( + json.dumps( + { + "ok": False, + "validator": "cto-codex-ephemeral-exec-helper", + "errors": ["prompt_missing"], + "warnings": [], + }, + indent=2, + sort_keys=True, + ) + ) + return 2 + if errors: + print(json.dumps({"ok": False, "errors": errors}, indent=2, sort_keys=True)) + return 1 + return subprocess.run(command, check=False).returncode + + +if __name__ == "__main__": + raise SystemExit(main())