|
#!/usr/bin/env python3 |
|
import json |
|
import os |
|
import sys |
|
from pathlib import Path |
|
from typing import Optional |
|
|
|
|
|
def latest_session_files(sessions_dir: Path, limit: Optional[int] = None) -> list[Path]: |
|
items = [] |
|
for root, _dirs, files in os.walk(sessions_dir): |
|
for name in files: |
|
if not name.endswith('.jsonl'): |
|
continue |
|
path = Path(root) / name |
|
try: |
|
mtime = path.stat().st_mtime |
|
except OSError: |
|
continue |
|
items.append((mtime, path)) |
|
items.sort(key=lambda item: item[0], reverse=True) |
|
paths = [path for _mtime, path in items] |
|
if limit is None: |
|
return paths |
|
return paths[:limit] |
|
|
|
|
|
def extract_message_text(content) -> str: |
|
if not isinstance(content, list): |
|
return '' |
|
parts = [] |
|
for item in content: |
|
if isinstance(item, dict): |
|
text = item.get('text') |
|
if isinstance(text, str) and text: |
|
parts.append(text) |
|
return ' '.join(parts).strip() |
|
|
|
|
|
def format_dir_label(cwd: Optional[str]) -> str: |
|
if not cwd: |
|
return 'unknown' |
|
name = os.path.basename(cwd.rstrip('/')) or cwd |
|
if len(name) <= 6: |
|
return name |
|
return f'{name[:3]}…{name[-3:]}' |
|
|
|
|
|
def escape_pango(text: str) -> str: |
|
return ( |
|
text.replace('&', '&') |
|
.replace('<', '<') |
|
.replace('>', '>') |
|
) |
|
|
|
|
|
def read_cmdline(cmdline_path: Path) -> list[str]: |
|
try: |
|
raw = cmdline_path.read_bytes() |
|
except OSError: |
|
return [] |
|
if not raw: |
|
return [] |
|
return [part.decode(errors='ignore') for part in raw.split(b'\0') if part] |
|
|
|
|
|
def is_codex_process(cmdline: list[str]) -> bool: |
|
if not cmdline: |
|
return False |
|
joined = ' '.join(cmdline) |
|
if 'codex_status.py' in joined: |
|
return False |
|
return ( |
|
'codex/bin/codex.js' in joined |
|
or 'vendor/x86_64-unknown-linux-musl/codex/codex' in joined |
|
or (cmdline and os.path.basename(cmdline[0]) == 'codex') |
|
) |
|
|
|
|
|
def extract_session_id(cmdline: list[str]) -> Optional[str]: |
|
# Example: codex.js resume <session_id> --search --yolo |
|
for idx, token in enumerate(cmdline): |
|
if token == 'resume' and idx + 1 < len(cmdline): |
|
return cmdline[idx + 1] |
|
return None |
|
|
|
|
|
def running_codex_processes() -> tuple[set[str], set[str]]: |
|
session_ids: set[str] = set() |
|
cwds: set[str] = set() |
|
proc_dir = Path('/proc') |
|
for entry in proc_dir.iterdir(): |
|
if not entry.name.isdigit(): |
|
continue |
|
pid = entry.name |
|
cmdline_path = proc_dir / pid / 'cmdline' |
|
cwd_path = proc_dir / pid / 'cwd' |
|
cmdline = read_cmdline(cmdline_path) |
|
if not is_codex_process(cmdline): |
|
continue |
|
session_id = extract_session_id(cmdline) |
|
if session_id: |
|
session_ids.add(session_id) |
|
try: |
|
cwd = os.readlink(cwd_path) |
|
except OSError: |
|
continue |
|
if cwd: |
|
cwds.add(cwd) |
|
return session_ids, cwds |
|
|
|
|
|
def read_session_state(session_path: Path) -> dict: |
|
session_id = None |
|
session_cwd = None |
|
last_role = None |
|
last_text = None |
|
last_ts = None |
|
pending_approval = False |
|
pending_approval_ts = None |
|
try: |
|
mtime = session_path.stat().st_mtime |
|
except OSError: |
|
mtime = 0.0 |
|
|
|
try: |
|
with session_path.open('r', encoding='utf-8') as handle: |
|
for line in handle: |
|
line = line.strip() |
|
if not line: |
|
continue |
|
try: |
|
item = json.loads(line) |
|
except json.JSONDecodeError: |
|
continue |
|
|
|
if item.get('type') == 'session_meta': |
|
payload = item.get('payload', {}) |
|
session_id = payload.get('id') or session_id |
|
session_cwd = payload.get('cwd') or session_cwd |
|
continue |
|
|
|
if item.get('type') == 'response_item': |
|
payload = item.get('payload', {}) |
|
if payload.get('type') == 'function_call': |
|
args = payload.get('arguments', '') |
|
if isinstance(args, str) and 'require_escalated' in args: |
|
pending_approval = True |
|
pending_approval_ts = item.get('timestamp') |
|
if payload.get('type') == 'function_call_output': |
|
if pending_approval and pending_approval_ts: |
|
pending_approval = False |
|
pending_approval_ts = None |
|
if payload.get('type') == 'message': |
|
last_role = payload.get('role') |
|
last_text = extract_message_text(payload.get('content')) |
|
last_ts = item.get('timestamp') |
|
except OSError: |
|
return { |
|
'path': session_path, |
|
'read_error': True, |
|
} |
|
|
|
return { |
|
'path': session_path, |
|
'session_id': session_id, |
|
'cwd': session_cwd, |
|
'last_role': last_role, |
|
'last_text': last_text, |
|
'last_ts': last_ts, |
|
'mtime': mtime, |
|
'pending_approval': pending_approval, |
|
'read_error': False, |
|
} |
|
|
|
|
|
def main() -> int: |
|
codex_home = Path(os.environ.get('CODEX_HOME', str(Path.home() / '.codex'))) |
|
sessions_dir = codex_home / 'sessions' |
|
|
|
if not sessions_dir.exists(): |
|
print(json.dumps({ |
|
'text': '', |
|
'class': 'inactive', |
|
'tooltip': '' |
|
})) |
|
return 0 |
|
|
|
max_sessions = os.environ.get('CODEX_STATUS_MAX') |
|
limit = None |
|
if max_sessions: |
|
try: |
|
limit = max(1, int(max_sessions)) |
|
except ValueError: |
|
limit = None |
|
|
|
session_paths = latest_session_files(sessions_dir, limit=limit) |
|
if not session_paths: |
|
print(json.dumps({ |
|
'text': '', |
|
'class': 'inactive', |
|
'tooltip': '' |
|
})) |
|
return 0 |
|
|
|
running_session_ids, running_cwds = running_codex_processes() |
|
if not (running_session_ids or running_cwds): |
|
print(json.dumps({ |
|
'text': '', |
|
'class': 'inactive', |
|
'tooltip': '', |
|
})) |
|
return 0 |
|
|
|
states = [read_session_state(path) for path in session_paths] |
|
by_id = {state.get('session_id'): state for state in states if state.get('session_id')} |
|
selected = [] |
|
seen_paths = set() |
|
|
|
for session_id in running_session_ids: |
|
state = by_id.get(session_id) |
|
if state and state['path'] not in seen_paths: |
|
selected.append(state) |
|
seen_paths.add(state['path']) |
|
|
|
if running_cwds: |
|
by_cwd = {} |
|
for state in states: |
|
cwd = state.get('cwd') |
|
if not cwd or cwd not in running_cwds: |
|
continue |
|
if state.get('session_id') in running_session_ids: |
|
continue |
|
existing = by_cwd.get(cwd) |
|
if existing is None or state.get('mtime', 0) > existing.get('mtime', 0): |
|
by_cwd[cwd] = state |
|
for state in by_cwd.values(): |
|
if state['path'] not in seen_paths: |
|
selected.append(state) |
|
seen_paths.add(state['path']) |
|
|
|
states = selected |
|
icon_waiting = '○' |
|
icon_idle = '●' |
|
color_waiting = '#9aa0a6' |
|
color_idle = '#7ccf7c' |
|
color_approval = '#ff9900' |
|
|
|
if not states: |
|
print(json.dumps({ |
|
'text': '', |
|
'class': 'inactive', |
|
'tooltip': '', |
|
})) |
|
return 0 |
|
|
|
icons = [] |
|
tooltip_lines = [] |
|
for idx, state in enumerate(states, start=1): |
|
path = state['path'] |
|
if state.get('read_error'): |
|
icons.append(f'<span foreground="{color_idle}">{icon_idle}</span>') |
|
tooltip_lines.append(f'{idx}. {path.name}: read error') |
|
continue |
|
last_role = state.get('last_role') |
|
waiting_for_user = last_role == 'assistant' |
|
pending_approval = state.get('pending_approval', False) |
|
if pending_approval: |
|
icon = '⚠' |
|
color = color_approval |
|
else: |
|
icon = icon_waiting if waiting_for_user else icon_idle |
|
color = color_waiting if waiting_for_user else color_idle |
|
label = escape_pango(format_dir_label(state.get('cwd'))) |
|
icon_span = f'<span foreground="{color}" font_family="0xProto Nerd Font">{icon}</span>' |
|
label_span = f'<span foreground="{color}" font_family="0xProto Nerd Font">{label}</span>' |
|
icons.append(f'{label_span} {icon_span}') |
|
|
|
label = f'{idx}. {path.name}' |
|
session_id = state.get('session_id') |
|
if session_id: |
|
label += f' ({session_id})' |
|
tooltip_lines.append(label) |
|
cwd = state.get('cwd') |
|
if cwd: |
|
tooltip_lines.append(f' CWD: {cwd}') |
|
last_ts = state.get('last_ts') |
|
if last_ts and last_role: |
|
tooltip_lines.append(f' Last: {last_role} at {last_ts}') |
|
last_text = state.get('last_text') |
|
if last_text: |
|
snippet = last_text |
|
if len(snippet) > 120: |
|
snippet = snippet[:117] + '...' |
|
tooltip_lines.append(f' Msg: {snippet}') |
|
|
|
print(json.dumps({ |
|
'text': ' '.join(icons), |
|
'class': 'codex', |
|
'tooltip': '\n'.join(tooltip_lines), |
|
})) |
|
return 0 |
|
|
|
|
|
if __name__ == '__main__': |
|
sys.exit(main()) |