1#!/usr/bin/env python3
2from __future__ import annotations
3 
4import argparse
5import json
6import sys
7from dataclasses import dataclass
8from pathlib import Path
9from typing import Any
10 
11DEFAULT_SESSIONS_ROOT = Path.home() / ".codex" / "sessions"
12 
13 
14@dataclass
15class SessionMeta:
16    path: Path
17    session_id: str | None
18    timestamp: str | None
19    cwd: str | None
20    source: str | None
21    cli_version: str | None
22    model_provider: str | None
23 
24 
25@dataclass
26class TextMessage:
27    role: str
28    text: str
29    timestamp: str | None
30    phase: str | None = None
31 
32 
33@dataclass
34class SessionRecord:
35    meta: SessionMeta
36    messages: list[TextMessage]
37 
38 
39def parse_args() -> argparse.Namespace:
40    parser = argparse.ArgumentParser(
41        description=(
42            "Find and parse Codex JSONL sessions under ~/.codex/sessions, filter them by repo path, "
43            "and extract only plain user/assistant text without tool calls."
44        )
45    )
46    parser.add_argument(
47        "--sessions-root",
48        default=str(DEFAULT_SESSIONS_ROOT),
49        help="Root directory that contains Codex session JSONL files (default: ~/.codex/sessions)",
50    )
51 
52    subparsers = parser.add_subparsers(dest="command", required=True)
53 
54    list_parser = subparsers.add_parser("list", help="List matching sessions with short previews")
55    add_session_filters(list_parser)
56    list_parser.add_argument("--limit", type=int, default=10, help="Maximum number of sessions to print")
57    list_parser.add_argument(
58        "--preview-chars",
59        type=int,
60        default=140,
61        help="Maximum preview length for last user/assistant messages",
62    )
63    list_parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON")
64 
65    messages_parser = subparsers.add_parser(
66        "messages",
67        help="Print extracted user/assistant messages for one or more sessions",
68    )
69    add_session_filters(messages_parser)
70    messages_parser.add_argument(
71        "--session",
72        help="Exact session file path or unique filename fragment to inspect directly",
73    )
74    messages_parser.add_argument(
75        "--limit-sessions",
76        type=int,
77        default=1,
78        help="Maximum number of sessions to print when using repo/path filters",
79    )
80    messages_parser.add_argument(
81        "--tail",
82        type=int,
83        default=8,
84        help="Number of trailing text messages to print per session (0 = all)",
85    )
86    messages_parser.add_argument(
87        "--role",
88        choices=["all", "user", "assistant"],
89        default="all",
90        help="Restrict output to one side of the conversation",
91    )
92    messages_parser.add_argument(
93        "--assistant-phase",
94        choices=["all", "commentary", "final_answer"],
95        default="all",
96        help="Restrict assistant output to commentary or final answers when event metadata is present",
97    )
98    messages_parser.add_argument("--json", action="store_true", help="Emit machine-readable JSON")
99 
100    return parser.parse_args()
101 
102 
103def add_session_filters(parser: argparse.ArgumentParser) -> None:
104    parser.add_argument(
105        "--repo",
106        help="Exact repo/worktree path to match against session_meta.payload.cwd",
107    )
108    parser.add_argument(
109        "--cwd-contains",
110        help="Substring match against session_meta.payload.cwd",
111    )
112 
113 
114def iter_session_paths(root: Path):
115    if not root.exists():
116        return
117    for path in sorted(root.rglob("*.jsonl"), reverse=True):
118        if path.is_file():
119            yield path
120 
121 
122def load_session_meta(path: Path) -> SessionMeta:
123    payload: dict[str, Any] = {}
124    with path.open("r", encoding="utf-8") as handle:
125        for line in handle:
126            if not line.strip():
127                continue
128            record = json.loads(line)
129            if record.get("type") == "session_meta":
130                payload = record.get("payload") or {}
131                break
132 
133    return SessionMeta(
134        path=path,
135        session_id=payload.get("id"),
136        timestamp=payload.get("timestamp"),
137        cwd=payload.get("cwd"),
138        source=payload.get("source") or payload.get("originator"),
139        cli_version=payload.get("cli_version"),
140        model_provider=payload.get("model_provider"),
141    )
142 
143 
144def matches_filters(meta: SessionMeta, repo: str | None, cwd_contains: str | None) -> bool:
145    cwd = meta.cwd or ""
146    if repo and cwd != repo:
147        return False
148    if cwd_contains and cwd_contains not in cwd:
149        return False
150    return True
151 
152 
153def content_to_text(content: Any) -> str:
154    if isinstance(content, str):
155        return content.strip()
156    if not isinstance(content, list):
157        return ""
158 
159    parts: list[str] = []
160    for item in content:
161        if not isinstance(item, dict):
162            continue
163        item_type = item.get("type")
164        if item_type in {"input_text", "output_text", "text"}:
165            text = item.get("text")
166            if isinstance(text, str) and text.strip():
167                parts.append(text.strip())
168    return "\n\n".join(parts).strip()
169 
170 
171def extract_messages(path: Path) -> list[TextMessage]:
172    event_messages: list[TextMessage] = []
173    fallback_messages: list[TextMessage] = []
174 
175    with path.open("r", encoding="utf-8") as handle:
176        for line in handle:
177            if not line.strip():
178                continue
179            record = json.loads(line)
180            record_type = record.get("type")
181            payload = record.get("payload") or {}
182            timestamp = record.get("timestamp")
183 
184            if record_type == "event_msg":
185                event_type = payload.get("type")
186                if event_type == "user_message":
187                    text = (payload.get("message") or "").strip()
188                    if text:
189                        event_messages.append(TextMessage(role="user", text=text, timestamp=timestamp))
190                elif event_type == "agent_message":
191                    text = (payload.get("message") or "").strip()
192                    if text:
193                        event_messages.append(
194                            TextMessage(
195                                role="assistant",
196                                text=text,
197                                timestamp=timestamp,
198                                phase=payload.get("phase"),
199                            )
200                        )
201                continue
202 
203            if record_type != "response_item":
204                continue
205            if payload.get("type") != "message":
206                continue
207            role = payload.get("role")
208            if role not in {"user", "assistant"}:
209                continue
210            text = content_to_text(payload.get("content"))
211            if text:
212                fallback_messages.append(TextMessage(role=role, text=text, timestamp=timestamp))
213 
214    return event_messages or fallback_messages
215 
216 
217def load_session_record(path: Path) -> SessionRecord:
218    return SessionRecord(meta=load_session_meta(path), messages=extract_messages(path))
219 
220 
221def find_matching_sessions(root: Path, repo: str | None, cwd_contains: str | None) -> list[SessionMeta]:
222    matches: list[SessionMeta] = []
223    for path in iter_session_paths(root):
224        meta = load_session_meta(path)
225        if matches_filters(meta, repo=repo, cwd_contains=cwd_contains):
226            matches.append(meta)
227    matches.sort(key=lambda item: (item.timestamp or "", str(item.path)), reverse=True)
228    return matches
229 
230 
231def resolve_session_target(root: Path, target: str) -> Path:
232    candidate = Path(target).expanduser()
233    if candidate.exists() and candidate.is_file():
234        return candidate
235 
236    matches = [path for path in iter_session_paths(root) if target in path.name or target in str(path)]
237    if not matches:
238        raise SystemExit(f"No session file matched: {target}")
239    if len(matches) > 1:
240        joined = "\n".join(str(path) for path in matches[:10])
241        extra = "" if len(matches) <= 10 else f"\n... and {len(matches) - 10} more"
242        raise SystemExit(f"Session target is ambiguous: {target}\n{joined}{extra}")
243    return matches[0]
244 
245 
246def preview(text: str, limit: int) -> str:
247    compact = " ".join(text.split())
248    if len(compact) <= limit:
249        return compact
250    return compact[: max(limit - 1, 0)].rstrip() + "…"
251 
252 
253def filter_messages(
254    messages: list[TextMessage],
255    role: str,
256    assistant_phase: str,
257    tail: int,
258) -> list[TextMessage]:
259    filtered = messages
260    if role != "all":
261        filtered = [message for message in filtered if message.role == role]
262    if assistant_phase != "all":
263        filtered = [
264            message
265            for message in filtered
266            if message.role != "assistant" or (message.phase or "commentary") == assistant_phase
267        ]
268    if tail > 0:
269        filtered = filtered[-tail:]
270    return filtered
271 
272 
273def session_to_dict(record: SessionRecord, preview_chars: int | None = None) -> dict[str, Any]:
274    last_user = next((message for message in reversed(record.messages) if message.role == "user"), None)
275    last_assistant = next((message for message in reversed(record.messages) if message.role == "assistant"), None)
276 
277    payload: dict[str, Any] = {
278        "path": str(record.meta.path),
279        "session_id": record.meta.session_id,
280        "timestamp": record.meta.timestamp,
281        "cwd": record.meta.cwd,
282        "source": record.meta.source,
283        "cli_version": record.meta.cli_version,
284        "model_provider": record.meta.model_provider,
285        "message_count": len(record.messages),
286        "messages": [
287            {
288                "role": message.role,
289                "timestamp": message.timestamp,
290                "phase": message.phase,
291                "text": message.text,
292            }
293            for message in record.messages
294        ],
295    }
296    if preview_chars is not None:
297        payload["last_user_preview"] = preview(last_user.text, preview_chars) if last_user else None
298        payload["last_assistant_preview"] = preview(last_assistant.text, preview_chars) if last_assistant else None
299    return payload
300 
301 
302def command_list(args: argparse.Namespace) -> int:
303    root = Path(args.sessions_root).expanduser()
304    matches = find_matching_sessions(root, repo=args.repo, cwd_contains=args.cwd_contains)[: args.limit]
305    records = [load_session_record(meta.path) for meta in matches]
306 
307    if args.json:
308        json.dump([session_to_dict(record, preview_chars=args.preview_chars) for record in records], sys.stdout, ensure_ascii=False, indent=2)
309        sys.stdout.write("\n")
310        return 0
311 
312    if not records:
313        print("No matching Codex sessions found.")
314        return 0
315 
316    for index, record in enumerate(records, 1):
317        data = session_to_dict(record, preview_chars=args.preview_chars)
318        print(f"[{index}] {data['timestamp'] or '-'}  {data['session_id'] or record.meta.path.name}")
319        print(f"  path: {data['path']}")
320        print(f"  cwd:  {data['cwd'] or '-'}")
321        print(f"  msgs: {data['message_count']}")
322        print(f"  user: {data['last_user_preview'] or '-'}")
323        print(f"  asst: {data['last_assistant_preview'] or '-'}")
324        if index != len(records):
325            print()
326    return 0
327 
328 
329def command_messages(args: argparse.Namespace) -> int:
330    root = Path(args.sessions_root).expanduser()
331 
332    if args.session:
333        session_paths = [resolve_session_target(root, args.session)]
334    else:
335        matches = find_matching_sessions(root, repo=args.repo, cwd_contains=args.cwd_contains)
336        session_paths = [meta.path for meta in matches[: args.limit_sessions]]
337 
338    records = [load_session_record(path) for path in session_paths]
339    filtered_records: list[dict[str, Any]] = []
340 
341    for record in records:
342        filtered_messages = filter_messages(
343            record.messages,
344            role=args.role,
345            assistant_phase=args.assistant_phase,
346            tail=args.tail,
347        )
348        filtered_records.append(
349            {
350                **session_to_dict(record),
351                "messages": [
352                    {
353                        "role": message.role,
354                        "timestamp": message.timestamp,
355                        "phase": message.phase,
356                        "text": message.text,
357                    }
358                    for message in filtered_messages
359                ],
360            }
361        )
362 
363    if args.json:
364        json.dump(filtered_records, sys.stdout, ensure_ascii=False, indent=2)
365        sys.stdout.write("\n")
366        return 0
367 
368    if not filtered_records:
369        print("No matching Codex sessions found.")
370        return 0
371 
372    for record_index, record in enumerate(filtered_records, 1):
373        print(f"[{record_index}] {record['timestamp'] or '-'}  {record['session_id'] or Path(record['path']).name}")
374        print(f"path: {record['path']}")
375        print(f"cwd:  {record['cwd'] or '-'}")
376        print()
377        for message in record["messages"]:
378            phase_suffix = ""
379            if message["role"] == "assistant" and message.get("phase"):
380                phase_suffix = f" ({message['phase']})"
381            print(f"[{message['role']}{phase_suffix}]")
382            print(message["text"].rstrip())
383            print()
384        if record_index != len(filtered_records):
385            print("=" * 80)
386    return 0
387 
388 
389def main() -> int:
390    args = parse_args()
391    if args.command == "list":
392        return command_list(args)
393    if args.command == "messages":
394        return command_messages(args)
395    raise SystemExit(f"Unsupported command: {args.command}")
396 
397 
398if __name__ == "__main__":
399    raise SystemExit(main())
400