diff --git a/scripts/codex_session.py b/scripts/codex_session.py index 68a922e..0e2a841 100644 --- a/scripts/codex_session.py +++ b/scripts/codex_session.py @@ -79,10 +79,11 @@ def _ps_codex_procs(ps_output: Optional[str] = None) -> list[CodexProc]: Looks for processes whose command is `codex` or starts with `codex ` (the CLI), excluding anything that just *mentions* codex (grep, editors, etc.). """ + lookup_tty_per_pid = ps_output is None if ps_output is None: try: ps_output = subprocess.check_output( - ["ps", "-axo", "pid=,tty=,lstart=,command="], + ["ps", "-axo", "pid=,lstart=,command="], stderr=subprocess.DEVNULL, text=True, ) @@ -94,29 +95,66 @@ def _ps_codex_procs(ps_output: Optional[str] = None) -> list[CodexProc]: line = line.strip() if not line: continue - # Format: " " + # Test fixtures may use the older inline-tty shape: + # " " + # Live discovery deliberately omits tty from the bulk ps scan and + # resolves it per PID below, avoiding stale/cached tty reuse. parts = line.split(None, 7) - if len(parts) < 8: - continue - pid_s, tty_s = parts[0], parts[1] - # lstart is 5 whitespace-separated fields (Day Mon DD HH:MM:SS YYYY) - lstart = " ".join(parts[2:7]) - command = parts[7] - # Skip ttys we can't address - if tty_s in ("?", "??", "-"): + if len(parts) >= 8 and _looks_like_tty_field(parts[1]): + pid_s = parts[0] + tty_s = parts[1] + lstart = " ".join(parts[2:7]) + command = parts[7] + elif len(parts) >= 7: + pid_s = parts[0] + tty_s = None + lstart = " ".join(parts[1:6]) + command = parts[6] + else: continue + if not _is_codex_command(command): continue try: pid = int(pid_s) except ValueError: continue + tty_full = _tty_for_pid(pid) if lookup_tty_per_pid else _normalize_tty(tty_s) + if not tty_full: + continue started = _parse_lstart(lstart) - tty_full = f"/dev/{tty_s}" if not tty_s.startswith("/dev/") else tty_s procs.append(CodexProc(pid=pid, tty=tty_full, started=started)) return procs +def _looks_like_tty_field(value: str) -> bool: + return value in ("?", "??", "-") or value.startswith(("tty", "/dev/tty")) + + +def _normalize_tty(tty_s: str | None) -> str | None: + if tty_s is None: + return None + tty_s = tty_s.strip() + if tty_s in ("", "?", "??", "-"): + return None + if any(ch.isspace() for ch in tty_s): + return None + return tty_s if tty_s.startswith("/dev/") else f"/dev/{tty_s}" + + +def _tty_for_pid(pid: int) -> str | None: + try: + out = subprocess.check_output( + ["ps", "-p", str(pid), "-o", "tty="], + stderr=subprocess.DEVNULL, + text=True, + ) + except (subprocess.CalledProcessError, FileNotFoundError): + return None + first = out.splitlines()[0] if out.splitlines() else "" + return _normalize_tty(first) + + def _is_codex_command(command: str) -> bool: """True iff this command line is a codex CLI invocation we should track.""" # First token = program path. Strip path. diff --git a/tests/test_classify_codex.py b/tests/test_classify_codex.py index 4aa90eb..9e256b9 100644 --- a/tests/test_classify_codex.py +++ b/tests/test_classify_codex.py @@ -209,11 +209,17 @@ def test_sweep_emits_signal_for_codex_proc(tmp_path, monkeypatch): ) fake_started = time.time() - 30 fake_ps = ( - f" 9999 ttys001 {time.strftime('%a %b %d %H:%M:%S %Y', time.localtime(fake_started))} codex\n" - ) - monkeypatch.setattr( - cs.subprocess, "check_output", lambda *a, **k: fake_ps + f" 9999 {time.strftime('%a %b %d %H:%M:%S %Y', time.localtime(fake_started))} codex\n" ) + + def fake_check_output(cmd, **kwargs): + if cmd == ["ps", "-axo", "pid=,lstart=,command="]: + return fake_ps + if cmd == ["ps", "-p", "9999", "-o", "tty="]: + return "ttys001\n" + raise AssertionError(f"unexpected command: {cmd!r}") + + monkeypatch.setattr(cs.subprocess, "check_output", fake_check_output) signal_dir = tmp_path / "signals" written = cs.sweep_once(str(signal_dir), str(tmp_path / "sessions")) assert len(written) == 1 diff --git a/tests/test_codex_session_tty.py b/tests/test_codex_session_tty.py new file mode 100644 index 0000000..c8662e0 --- /dev/null +++ b/tests/test_codex_session_tty.py @@ -0,0 +1,58 @@ +"""Regression tests for per-process Codex tty lookup.""" + +from __future__ import annotations + +import sys +from pathlib import Path + +import pytest + +SCRIPTS = Path(__file__).resolve().parent.parent / "scripts" +sys.path.insert(0, str(SCRIPTS)) + +import codex_session as cs # noqa: E402 + + +def test_live_codex_process_discovery_reads_tty_per_pid(monkeypatch): + bulk_ps = ( + " 1111 Mon May 28 19:20:01 2026 codex\n" + " 2222 Mon May 28 19:21:01 2026 codex --resume foo\n" + ) + calls = [] + + def fake_check_output(cmd, **kwargs): + calls.append(cmd) + if cmd == ["ps", "-axo", "pid=,lstart=,command="]: + return bulk_ps + if cmd == ["ps", "-p", "1111", "-o", "tty="]: + return "ttys010\n" + if cmd == ["ps", "-p", "2222", "-o", "tty="]: + return "ttys011\n" + raise AssertionError(f"unexpected command: {cmd!r}") + + monkeypatch.setattr(cs.subprocess, "check_output", fake_check_output) + + procs = cs._ps_codex_procs() + + assert [(p.pid, p.tty) for p in procs] == [ + (1111, "/dev/ttys010"), + (2222, "/dev/ttys011"), + ] + assert ["ps", "-p", "1111", "-o", "tty="] in calls + assert ["ps", "-p", "2222", "-o", "tty="] in calls + + +@pytest.mark.parametrize("raw_tty", ["?", "??", "-", ""]) +def test_live_codex_process_discovery_skips_pid_without_tty(monkeypatch, raw_tty): + bulk_ps = " 3333 Mon May 28 19:22:01 2026 codex\n" + + def fake_check_output(cmd, **kwargs): + if cmd == ["ps", "-axo", "pid=,lstart=,command="]: + return bulk_ps + if cmd == ["ps", "-p", "3333", "-o", "tty="]: + return raw_tty + "\n" + raise AssertionError(f"unexpected command: {cmd!r}") + + monkeypatch.setattr(cs.subprocess, "check_output", fake_check_output) + + assert cs._ps_codex_procs() == []