1#!/usr/bin/env python3
2"""
3Session Catchup Script for planning-with-files
4 
5Analyzes the previous session to find unsynced context after the last
6planning file update. Designed to run on SessionStart.
7 
8Usage: python3 session-catchup.py [project-path]
9"""
10 
11import json
12import sys
13import os
14from pathlib import Path
15from typing import Any, Dict, Iterable, List, Optional, Tuple
16 
17try:
18    import orjson
19except ImportError:
20    orjson = None
21 
22PLANNING_FILES = ['task_plan.md', 'progress.md', 'findings.md']
23MIN_SESSION_BYTES = 5000
24 
25 
26def json_loads(line: str) -> Optional[Dict[str, Any]]:
27    """Prefer optional orjson while keeping the hook dependency-free."""
28    try:
29        if orjson is not None:
30            data = orjson.loads(line)
31        else:
32            data = json.loads(line)
33    except (ValueError, TypeError, UnicodeDecodeError):
34        return None
35    return data if isinstance(data, dict) else None
36 
37 
38def normalize_for_compare(path_value: str) -> str:
39    expanded = os.path.expanduser(path_value)
40    try:
41        return str(Path(expanded).resolve())
42    except (OSError, ValueError):
43        return os.path.abspath(expanded)
44 
45 
46def normalize_path(project_path: str) -> str:
47    """Normalize project path to match Claude Code's internal representation.
48 
49    Claude Code stores session directories using the Windows-native path
50    (e.g., C:\\Users\\...) sanitized with separators replaced by dashes.
51    Git Bash passes /c/Users/... which produces a DIFFERENT sanitized
52    string. This function converts Git Bash paths to Windows paths first.
53    """
54    p = project_path
55 
56    # Git Bash / MSYS2: /c/Users/... -> C:/Users/...
57    if len(p) >= 3 and p[0] == '/' and p[2] == '/':
58        p = p[1].upper() + ':' + p[2:]
59 
60    # Resolve to absolute path to handle relative paths and symlinks
61    try:
62        resolved = str(Path(p).resolve())
63        # On Windows, resolve() returns C:\Users\... which is what we want
64        if os.name == 'nt' or '\\' in resolved:
65            p = resolved
66    except (OSError, ValueError):
67        pass
68 
69    return p
70 
71 
72def get_claude_project_dir(project_path: str) -> Path:
73    """Resolve Claude Code's project-specific session storage path."""
74    normalized = normalize_path(project_path)
75 
76    # Claude Code's sanitization: replace path separators and : with -
77    sanitized = normalized.replace('\\', '-').replace('/', '-').replace(':', '-')
78    sanitized = sanitized.replace('_', '-')
79    # Strip leading dash if present (Unix absolute paths start with /)
80    if sanitized.startswith('-'):
81        sanitized = sanitized[1:]
82 
83    return Path.home() / '.claude' / 'projects' / sanitized
84 
85 
86def get_sessions_sorted(project_dir: Path) -> List[Path]:
87    """Get all session files sorted by modification time (newest first)."""
88    sessions = list(project_dir.glob('*.jsonl'))
89    main_sessions = [s for s in sessions if not s.name.startswith('agent-')]
90    return sorted(main_sessions, key=safe_stat_mtime, reverse=True)
91 
92 
93def safe_stat_mtime(path: Path) -> float:
94    try:
95        return path.stat().st_mtime
96    except OSError:
97        return 0.0
98 
99 
100def is_substantial_session(session: Path) -> bool:
101    try:
102        return session.stat().st_size > MIN_SESSION_BYTES
103    except OSError:
104        return False
105 
106 
107def read_codex_meta(session_file: Path) -> Optional[Dict[str, Any]]:
108    """Read the first session_meta; later meta records may be copied parent context."""
109    try:
110        with open(session_file, 'r', encoding='utf-8', errors='replace') as f:
111            for line in f:
112                data = json_loads(line)
113                if not data or data.get('type') != 'session_meta':
114                    continue
115                payload = data.get('payload')
116                return payload if isinstance(payload, dict) else None
117    except OSError:
118        return None
119    return None
120 
121 
122def codex_meta_cwd(meta: Dict[str, Any]) -> Optional[str]:
123    cwd = meta.get('cwd')
124    return cwd if isinstance(cwd, str) else None
125 
126 
127def find_current_codex_session(sessions: List[Path]) -> Optional[Path]:
128    thread_id = os.getenv('CODEX_THREAD_ID', '').strip()
129    if not thread_id:
130        return None
131 
132    for session in sessions:
133        if thread_id in session.name:
134            return session
135    return None
136 
137 
138def is_codex_project_session(session: Path, project_cmp: str) -> bool:
139    if not is_substantial_session(session):
140        return False
141 
142    meta = read_codex_meta(session)
143    if not meta:
144        return False
145    source = meta.get('source')
146    if isinstance(source, dict) and 'subagent' in source:
147        return False
148    cwd = codex_meta_cwd(meta)
149    return bool(cwd and normalize_for_compare(cwd) == project_cmp)
150 
151 
152def get_codex_sessions(project_path: str) -> Iterable[Path]:
153    sessions_dir = Path(os.path.expanduser(os.getenv('CODEX_SESSIONS_DIR', '~/.codex/sessions')))
154    if not sessions_dir.exists():
155        return
156 
157    project_cmp = normalize_for_compare(project_path)
158    sessions = sorted(sessions_dir.rglob('rollout-*.jsonl'), key=safe_stat_mtime, reverse=True)
159    current = find_current_codex_session(sessions)
160    if current and is_codex_project_session(current, project_cmp):
161        yield current
162 
163    for session in sessions:
164        if session == current:
165            continue
166        if is_codex_project_session(session, project_cmp):
167            yield session
168 
169 
170def get_session_candidates(project_path: str) -> Tuple[str, Iterable[Path]]:
171    if '/.codex/' in Path(__file__).resolve().as_posix().lower():
172        return 'codex', get_codex_sessions(project_path)
173 
174    claude_project_dir = get_claude_project_dir(project_path)
175    if claude_project_dir.exists():
176        return 'claude', get_sessions_sorted(claude_project_dir)
177    return 'claude', []
178 
179 
180def parse_session_messages(session_file: Path) -> List[Dict[str, Any]]:
181    """Parse all messages from a session file, preserving order."""
182    messages = []
183    with open(session_file, 'r', encoding='utf-8', errors='replace') as f:
184        for line_num, line in enumerate(f):
185            data = json_loads(line)
186            if data is not None:
187                data['_line_num'] = line_num
188                messages.append(data)
189    return messages
190 
191 
192def planning_file_from_path(path_value: Any) -> Optional[str]:
193    if not isinstance(path_value, str):
194        return None
195    for pf in PLANNING_FILES:
196        if path_value.endswith(pf):
197            return pf
198    return None
199 
200 
201def planning_file_from_paths(paths: Iterable[Any]) -> Optional[str]:
202    matches = {pf for path in paths if (pf := planning_file_from_path(path))}
203    for pf in PLANNING_FILES:
204        if pf in matches:
205            return pf
206    return None
207 
208 
209def codex_planning_update(payload: Dict[str, Any]) -> Optional[str]:
210    """Use Codex's structured apply_patch result instead of parsing tool text."""
211    if payload.get('type') != 'patch_apply_end' or payload.get('success') is not True:
212        return None
213    changes = payload.get('changes')
214    return planning_file_from_paths(changes.keys()) if isinstance(changes, dict) else None
215 
216 
217def find_last_planning_update(messages: List[Dict[str, Any]]) -> Tuple[int, Optional[str]]:
218    """
219    Find the last time a planning file was written/edited.
220    Returns (line_number, filename) or (-1, None) if not found.
221    """
222    last_update_line = -1
223    last_update_file = None
224 
225    for msg in messages:
226        line_num = msg.get('_line_num')
227        if not isinstance(line_num, int):
228            continue
229        msg_type = msg.get('type')
230 
231        if msg_type == 'assistant':
232            content = msg.get('message', {}).get('content', [])
233            if isinstance(content, list):
234                for item in content:
235                    if item.get('type') == 'tool_use':
236                        tool_name = item.get('name', '')
237                        tool_input = item.get('input', {})
238                        if not isinstance(tool_input, dict):
239                            tool_input = {}
240 
241                        if tool_name in ('Write', 'Edit'):
242                            planning_file = planning_file_from_path(tool_input.get('file_path', ''))
243                            if planning_file:
244                                last_update_line = line_num
245                                last_update_file = planning_file
246 
247        elif msg_type == 'event_msg':
248            payload = msg.get('payload')
249            if isinstance(payload, dict):
250                planning_file = codex_planning_update(payload)
251                if planning_file:
252                    last_update_line = line_num
253                    last_update_file = planning_file
254 
255    return last_update_line, last_update_file
256 
257 
258def text_content(content: Any) -> str:
259    if isinstance(content, str):
260        return content
261    if not isinstance(content, list):
262        return ''
263    return '\n'.join(
264        item.get('text', '')
265        for item in content
266        if isinstance(item, dict) and isinstance(item.get('text'), str)
267    )
268 
269 
270def parse_codex_tool_args(payload: Dict[str, Any]) -> Tuple[Dict[str, Any], str]:
271    raw_args = payload.get('arguments', payload.get('input', ''))
272    if isinstance(raw_args, dict):
273        return raw_args, json.dumps(raw_args, ensure_ascii=True)
274    if not isinstance(raw_args, str):
275        return {}, ''
276    decoded = json_loads(raw_args)
277    return (decoded, raw_args) if isinstance(decoded, dict) else ({}, raw_args)
278 
279 
280def summarize_codex_tool(payload: Dict[str, Any]) -> str:
281    tool_name = payload.get('name', 'tool')
282    tool_args, raw_args = parse_codex_tool_args(payload)
283    if tool_name == 'exec_command':
284        command = tool_args.get('cmd', raw_args)
285        if isinstance(command, str):
286            return f"exec_command: {command[:80]}"
287    return str(tool_name)
288 
289 
290def extract_messages_after(messages: List[Dict[str, Any]], after_line: int) -> List[Dict[str, Any]]:
291    """Extract conversation messages after a certain line number."""
292    result = []
293    for msg in messages:
294        line_num = msg.get('_line_num')
295        if not isinstance(line_num, int) or line_num <= after_line:
296            continue
297 
298        msg_type = msg.get('type')
299        is_meta = msg.get('isMeta', False)
300 
301        if msg_type == 'user' and not is_meta:
302            content = text_content(msg.get('message', {}).get('content', ''))
303 
304            if content:
305                if content.startswith(('<local-command', '<command-', '<task-notification')):
306                    continue
307                if len(content) > 20:
308                    result.append({'role': 'user', 'content': content, 'line': line_num})
309 
310        elif msg_type == 'assistant':
311            msg_content = msg.get('message', {}).get('content', '')
312            text = text_content(msg_content)
313            tool_uses = []
314 
315            if isinstance(msg_content, list):
316                for item in msg_content:
317                    if isinstance(item, dict) and item.get('type') == 'tool_use':
318                        tool_name = item.get('name', '')
319                        tool_input = item.get('input', {})
320                        if not isinstance(tool_input, dict):
321                            tool_input = {}
322                        if tool_name == 'Edit':
323                            tool_uses.append(f"Edit: {tool_input.get('file_path', 'unknown')}")
324                        elif tool_name == 'Write':
325                            tool_uses.append(f"Write: {tool_input.get('file_path', 'unknown')}")
326                        elif tool_name == 'Bash':
327                            cmd = tool_input.get('command', '')[:80]
328                            tool_uses.append(f"Bash: {cmd}")
329                        else:
330                            tool_uses.append(f"{tool_name}")
331 
332            if text or tool_uses:
333                result.append({
334                    'role': 'assistant',
335                    'content': text[:600] if text else '',
336                    'tools': tool_uses,
337                    'line': line_num
338                })
339 
340        elif msg_type == 'response_item':
341            payload = msg.get('payload')
342            if not isinstance(payload, dict):
343                continue
344 
345            payload_type = payload.get('type')
346            if payload_type == 'message':
347                role = payload.get('role')
348                if role not in ('user', 'assistant'):
349                    continue
350                content = text_content(payload.get('content'))
351                if role == 'user':
352                    if content.startswith(('<local-command', '<command-', '<task-notification')):
353                        continue
354                    if len(content) > 20:
355                        result.append({'role': 'user', 'content': content, 'line': line_num})
356                elif content:
357                    result.append({
358                        'role': 'assistant',
359                        'content': content[:600],
360                        'tools': [],
361                        'line': line_num
362                    })
363            elif payload_type in ('function_call', 'custom_tool_call'):
364                result.append({
365                    'role': 'assistant',
366                    'content': '',
367                    'tools': [summarize_codex_tool(payload)],
368                    'line': line_num
369                })
370 
371    return result
372 
373 
374def main():
375    project_path = sys.argv[1] if len(sys.argv) > 1 else os.getcwd()
376 
377    # Check if planning files exist (indicates active task)
378    has_planning_files = any(
379        Path(project_path, f).exists() for f in PLANNING_FILES
380    )
381    if not has_planning_files:
382        # No planning files in this project; skip catchup to avoid noise.
383        return
384 
385    runtime_name, sessions = get_session_candidates(project_path)
386 
387    # Find a substantial previous session
388    target_session = None
389    for session in sessions:
390        if runtime_name == 'claude' and not is_substantial_session(session):
391            continue
392        target_session = session
393        break
394 
395    if not target_session:
396        return
397 
398    messages = parse_session_messages(target_session)
399    last_update_line, last_update_file = find_last_planning_update(messages)
400 
401    # No planning updates in the target session; skip catchup output.
402    if last_update_line < 0:
403        return
404 
405    # Only output if there's unsynced content
406    messages_after = extract_messages_after(messages, last_update_line)
407 
408    if not messages_after:
409        return
410 
411    # Output catchup report
412    print("\n[planning-with-files] SESSION CATCHUP DETECTED")
413    print(f"Previous session: {target_session.stem}")
414    print(f"Runtime: {runtime_name}")
415 
416    print(f"Last planning update: {last_update_file} at message #{last_update_line}")
417    print(f"Unsynced messages: {len(messages_after)}")
418 
419    print("\n--- UNSYNCED CONTEXT ---")
420    assistant_label = 'CODEX' if runtime_name == 'codex' else 'CLAUDE'
421    for msg in messages_after[-15:]:  # Last 15 messages
422        if msg['role'] == 'user':
423            print(f"USER: {msg['content'][:300]}")
424        else:
425            if msg.get('content'):
426                print(f"{assistant_label}: {msg['content'][:300]}")
427            if msg.get('tools'):
428                print(f"  Tools: {', '.join(msg['tools'][:4])}")
429 
430    print("\n--- RECOMMENDED ---")
431    print("1. Run: git diff --stat")
432    print("2. Read: task_plan.md, progress.md, findings.md")
433    print("3. Update planning files based on above context")
434    print("4. Continue with task")
435 
436 
437if __name__ == '__main__':
438    main()
439