refactor(plugins): move codex hooks to uv scripts

Signed-off-by: phernandez <paul@basicmachines.co>

Author: phernandez

plugins/codex/README.md

@@ -29,8 +29,10 @@ verification, decision capture, and resumable checkpoints.
 | `.codex-plugin/plugin.json` | Codex plugin manifest |
 | `.mcp.json` | Basic Memory MCP server configuration |
 | `hooks/hooks.json` | SessionStart and PreCompact hook registration |
-| `hooks/session-start.sh` | Injects a compact memory brief at thread start |
-| `hooks/pre-compact.sh` | Writes an automatic Codex checkpoint before compaction |
+| `hooks/session-start.sh` | Launches the SessionStart uv script |
+| `hooks/session-start.py` | Injects a compact memory brief at thread start |
+| `hooks/pre-compact.sh` | Launches the PreCompact uv script |
+| `hooks/pre-compact.py` | Writes an automatic Codex checkpoint before compaction |
 | `skills/` | Codex-native Basic Memory workflows |
 | `schemas/` | Seed schemas for Codex sessions, decisions, and tasks |
 

plugins/codex/hooks/pre-compact.py

@@ -0,0 +1,232 @@
+#!/usr/bin/env -S uv run --script
+"""Checkpoint Codex work into Basic Memory before compaction."""
+
+import json
+import os
+import re
+import shlex
+import shutil
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+
+UUID_RE = re.compile(
+    r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
+    re.IGNORECASE,
+)
+
+
+def basic_memory_command() -> list[str] | None:
+    configured = os.environ.get("BM_BIN")
+    if configured:
+        return shlex.split(configured)
+    if shutil.which("basic-memory"):
+        return ["basic-memory"]
+    if shutil.which("bm"):
+        return ["bm"]
+    if shutil.which("uvx"):
+        return ["uvx", "basic-memory"]
+    if shutil.which("uv"):
+        return ["uv", "tool", "run", "basic-memory"]
+    return None
+
+
+def parse_payload() -> dict:
+    try:
+        payload = json.loads(sys.stdin.read() or "{}")
+    except Exception:
+        return {}
+    return payload if isinstance(payload, dict) else {}
+
+
+def load_config(directory: Path) -> dict:
+    path = directory / ".codex" / "basic-memory.json"
+    try:
+        data = json.loads(path.read_text())
+    except Exception:
+        return {}
+    if not isinstance(data, dict):
+        return {}
+    return data.get("basicMemory", data)
+
+
+def text_of(content):
+    if isinstance(content, str):
+        return content
+    if isinstance(content, list):
+        parts = []
+        for block in content:
+            if isinstance(block, dict) and block.get("type") == "text":
+                text = block.get("text")
+                if isinstance(text, str):
+                    parts.append(text)
+        return "\n".join(parts)
+    return ""
+
+
+def transcript_turns(path: str):
+    collected = []
+    if not path:
+        return collected
+    try:
+        with open(path) as handle:
+            for line in handle:
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    obj = json.loads(line)
+                except Exception:
+                    continue
+                if obj.get("isMeta") or obj.get("toolUseResult") is not None:
+                    continue
+                msg = obj.get("message") if isinstance(obj.get("message"), dict) else obj
+                role = msg.get("role") or obj.get("type")
+                if role not in ("user", "assistant"):
+                    continue
+                text = text_of(msg.get("content")).strip()
+                if text:
+                    collected.append((role, text))
+    except Exception:
+        return []
+    return collected
+
+
+def git_status(directory: Path) -> list[str]:
+    try:
+        out = subprocess.run(
+            ["git", "status", "--short"],
+            cwd=directory,
+            capture_output=True,
+            text=True,
+            timeout=5,
+        )
+    except Exception:
+        return []
+    if out.returncode != 0:
+        return []
+    return [line for line in out.stdout.splitlines() if line.strip()][:20]
+
+
+def clip(value: str, limit: int) -> str:
+    compact = " ".join(value.split())
+    return compact if len(compact) <= limit else compact[: limit - 1].rstrip() + "..."
+
+
+def main() -> int:
+    bm_cmd = basic_memory_command()
+    if not bm_cmd:
+        return 0
+
+    payload = parse_payload()
+    cwd = Path(payload.get("cwd") or os.getcwd())
+    transcript_path = payload.get("transcript_path") or ""
+    session_id = payload.get("session_id") or ""
+    turn_id = payload.get("turn_id") or ""
+    trigger = payload.get("trigger") or ""
+    model = payload.get("model") or ""
+
+    cfg = load_config(cwd)
+    primary_project = str(cfg.get("primaryProject") or "").strip()
+    capture_folder = str(cfg.get("captureFolder") or "codex-sessions").strip()
+
+    if not primary_project:
+        return 0
+
+    conversation = transcript_turns(transcript_path)
+    if not conversation or not any(role == "user" for role, _ in conversation):
+        return 0
+
+    user_messages = [text for role, text in conversation if role == "user"]
+    assistant_messages = [text for role, text in conversation if role == "assistant"]
+    opening = user_messages[0] if user_messages else ""
+    recent_user = user_messages[-3:]
+    recent_assistant = assistant_messages[-2:]
+    status_lines = git_status(cwd)
+
+    now = datetime.now(timezone.utc)
+    iso = now.isoformat(timespec="seconds")
+    title = f"Codex session {now.strftime('%Y-%m-%d %H:%M:%S')} - {clip(opening, 40)}"
+
+    frontmatter = [
+        "---",
+        "type: codex_session",
+        "status: open",
+        f"started: {iso}",
+        f"ended: {iso}",
+        f"project: {primary_project}",
+        f"cwd: {cwd}",
+    ]
+    if session_id:
+        frontmatter.append(f"codex_session_id: {session_id}")
+    if turn_id:
+        frontmatter.append(f"codex_turn_id: {turn_id}")
+    if trigger:
+        frontmatter.append(f"trigger: {trigger}")
+    if model:
+        frontmatter.append(f"model: {model}")
+    frontmatter += ["capture: extractive", "---"]
+
+    body = [
+        "",
+        f"# {title}",
+        "",
+        "_Automatic Codex pre-compaction checkpoint. It records the working cursor, "
+        "not a polished summary._",
+        "",
+        "## Summary",
+        f"Working in `{cwd}`.",
+        f"- Opening request: {clip(opening, 300)}" if opening else "",
+        "",
+        "## Recent User Cursor",
+    ]
+    body += [f"- {clip(message, 240)}" for message in recent_user]
+    if recent_assistant:
+        body += ["", "## Recent Assistant Notes"]
+        body += [f"- {clip(message, 240)}" for message in recent_assistant]
+    if status_lines:
+        body += ["", "## Working Tree"]
+        body += [f"- `{line}`" for line in status_lines]
+    body += [
+        "",
+        "## Observations",
+        f"- [context] Codex worked in `{cwd}`",
+        f"- [context] Session opened with: {clip(opening, 200)}" if opening else "",
+        "- [next_step] Re-read this checkpoint, inspect the current worktree, and "
+        "continue from the latest user request",
+    ]
+
+    content = "\n".join(frontmatter + body)
+    project_flag = "--project-id" if UUID_RE.match(primary_project) else "--project"
+
+    try:
+        subprocess.run(
+            [
+                *bm_cmd,
+                "tool",
+                "write-note",
+                "--title",
+                title,
+                "--folder",
+                capture_folder,
+                project_flag,
+                primary_project,
+                "--tags",
+                "codex",
+                "--tags",
+                "auto-capture",
+            ],
+            input=content,
+            capture_output=True,
+            text=True,
+            timeout=60,
+        )
+    except Exception:
+        return 0
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())