From c73c61f496ad5851a0abfdab38962e2094e3dc66 Mon Sep 17 00:00:00 2001 From: olegbrok Date: Wed, 5 Aug 2026 12:11:41 -0700 Subject: [PATCH] Fix Codex tmux queued wake delivery --- src/pinky_daemon/codex_tmux_session.py | 62 ++++++++++++++++++++ src/pinky_daemon/codex_tmux_transcript.py | 12 ++++ tests/test_codex_tmux_session.py | 71 +++++++++++++++++++++++ tests/test_codex_tmux_transcript.py | 32 ++++++++++ 4 files changed, 177 insertions(+) diff --git a/src/pinky_daemon/codex_tmux_session.py b/src/pinky_daemon/codex_tmux_session.py index db8ba471..f0f64f5e 100644 --- a/src/pinky_daemon/codex_tmux_session.py +++ b/src/pinky_daemon/codex_tmux_session.py @@ -312,6 +312,11 @@ async def _start_tailer(self) -> None: agent_name=self.agent_name, model=self._codex_model, path_discovery=self._discover_transcript_path, + # Scheduler receipts use the same exact transcript-observation + # contract as Claude tmux. Codex records acceptance as an + # event_msg/user_message entry instead of Claude's user or + # queue-operation rows. + on_entry=self._on_transcript_entry, ) if guessed is not None: # Warm-wake / resume: seek to EOF so we don't replay history. @@ -321,6 +326,63 @@ async def _start_tailer(self) -> None: pass await super()._start_tailer() + # ── seam: scheduler queued-route delivery (#1006) ────────────────────── + def _scheduler_pane_busy(self, candidate=None) -> bool: + """Use Codex-local turn state instead of Claude hook status. + + The base tmux implementation deliberately requires a fresh persisted + ``idle`` row from Claude Code's working/idle hooks before it pastes a + scheduler turn. Codex does not run those hooks, so that row is stale + or absent and every trigger wake waits forever despite an idle pane. + + Keep the same conservative no-overlap rule using the evidence Codex + does own: ordinary work in hand/queued, transcript-backed inflight + metadata, tool activity, and the rollout tailer's active-turn flag. + The inherited scheduler task, REPL lock, logs, exact receipt, and + retirement paths remain unchanged. + """ + candidate_in_worker = ( + candidate is not None and self._inflight_turn is candidate + ) + if ( + self._inflight_tool_calls + or ( + self._inflight_turn is not None + and not candidate_in_worker + ) + or ( + not self._message_queue.empty() + and not candidate_in_worker + ) + or self._inflight_metas + ): + return True + return bool(getattr(self._tailer, "_active", False)) + + def _on_transcript_entry(self, entry: dict) -> None: + """Map Codex rollout acceptance onto the shared exact-receipt path.""" + if entry.get("type") == "event_msg": + payload = entry.get("payload") or {} + if payload.get("type") == "user_message": + prompt = payload.get("message") + if isinstance(prompt, str): + turn = self._match_acceptance_turn(prompt) + # The rollout tailer can observe user_message and a very + # fast task_complete in one read while paste_text's final + # tmux subprocess is still returning. Reserve FIFO + # metadata before resolving the exact scheduler receipt so + # that same-read completion has a head to retire. The + # normal post-paste path is idempotent on this flag. + if ( + turn is not None + and turn.scheduler_delivery is not None + and not turn.pane_delivery_recorded + ): + self._finish_turn_delivery(turn) + self._mark_transport_accepted(turn) + return + super()._on_transcript_entry(entry) + # ── seam: cold-start (codex trust pre-seed + NUX dismissal + readiness) ── async def _spawn_tmux_repl(self) -> None: cwd = str(Path(self._config.working_dir or ".").resolve()) diff --git a/src/pinky_daemon/codex_tmux_transcript.py b/src/pinky_daemon/codex_tmux_transcript.py index 7bcef818..a238be5f 100644 --- a/src/pinky_daemon/codex_tmux_transcript.py +++ b/src/pinky_daemon/codex_tmux_transcript.py @@ -293,6 +293,7 @@ def __init__( fallback_poll_sec: float = _FALLBACK_POLL_SEC, active_poll_sec: float = _ACTIVE_POLL_SEC, path_discovery: Callable[[], Path | None] | None = None, + on_entry: Callable[[dict], None] | None = None, ) -> None: self._path = Path(transcript_path) self._on_turn_complete = on_turn_complete @@ -303,6 +304,7 @@ def __init__( # Self-heal: when the watched path doesn't exist, call this to # scan for the real rollout by cwd match (mirrors #515). self._path_discovery = path_discovery + self._on_entry = on_entry self._offset: int = 0 # Bumped on every path-changing ``set_transcript_path``. Lets @@ -609,6 +611,16 @@ async def _read_and_dispatch(self) -> int: if not self.session_cwd: self.session_cwd = payload.get("cwd", "") + if self._on_entry is not None: + try: + self._on_entry(dict(entry)) + except Exception as e: + self._stats["callback_errors"] += 1 + _log( + f"codex_tailer[{self._agent_name}]: on_entry raised " + f"({type(e).__name__}: {e}); continuing" + ) + closes_turn = False aborted = False try: diff --git a/tests/test_codex_tmux_session.py b/tests/test_codex_tmux_session.py index 5993088b..d252b3c7 100644 --- a/tests/test_codex_tmux_session.py +++ b/tests/test_codex_tmux_session.py @@ -317,6 +317,77 @@ async def test_oauth_watcher_is_noop(): ss._tmux.capture_pane.assert_not_called() +# ── #1006 scheduler queued-route delivery ────────────────────────────────── +@pytest.mark.asyncio +async def test_scheduler_queued_route_reaches_codex_pane_with_stale_cc_status( + capsys, +): + """Codex has no Claude working/idle hooks, so their persisted status is + not a delivery gate. A scheduler wake must use Codex-local in-flight + state, reach the pane, and keep the inherited exact-receipt contract.""" + tmux = _mock_tmux() + ss = _session(tmux=tmux) + ss._state_machine._state = SessionState.CONNECTED + ss._config.live_status_fn = lambda: { + "status": "working", + "last_updated": time.time(), + } + + receipt = await ss.send_scheduler_prompt("scheduled codex wake") + for _ in range(100): + if tmux.paste_text.await_count == 1: + break + await asyncio.sleep(0.01) + + tmux.paste_text.assert_awaited_once_with( + "scheduled codex wake", enter=True + ) + assert not receipt.done(), "pane keystrokes alone are not acceptance" + assert "tmux[murzik]: queued message (chat=)" in capsys.readouterr().err + + # Codex's rollout user_message is the exact transport-acceptance edge. + ss._on_transcript_entry({ + "type": "event_msg", + "payload": { + "type": "user_message", + "message": "scheduled codex wake", + }, + }) + assert await receipt is True + + +@pytest.mark.asyncio +async def test_codex_acceptance_reserves_meta_before_same_read_completion(): + """A user_message + task_complete pair can beat paste_text's return.""" + ss = _session() + ss._state_machine._state = SessionState.CONNECTED + receipt = asyncio.get_running_loop().create_future() + turn = _QueuedTurn( + prompt="fast scheduled wake", + scheduler_delivery=receipt, + scheduler_serialized=True, + pane_delivery_started=True, + ) + ss._scheduler_pending_turns.append(turn) + + ss._on_transcript_entry({ + "type": "event_msg", + "payload": { + "type": "user_message", + "message": "fast scheduled wake", + }, + }) + + assert await receipt is True + assert turn.pane_delivery_recorded is True + assert len(ss._inflight_metas) == 1 + + await ss._handle_turn_complete( + TurnResponse(text="done", stop_reason="task_complete") + ) + assert len(ss._inflight_metas) == 0 + + # ── StopFailure hook is a no-op for codex (#795 P2) ───────────────────────── @pytest.mark.asyncio async def test_handle_stop_failure_is_noop_for_codex(): diff --git a/tests/test_codex_tmux_transcript.py b/tests/test_codex_tmux_transcript.py index 49a09737..82c60c62 100644 --- a/tests/test_codex_tmux_transcript.py +++ b/tests/test_codex_tmux_transcript.py @@ -19,6 +19,7 @@ - Cold-start seek-to-EOF vs first-bind seek-to-0. - ``path_discovery``: cwd-filtered selection from a temp tree of fixtures. - Background loop: wake() short-circuits poll; fallback poll makes progress. + - Raw rollout entries are forwarded to the session acceptance observer. - Standard stats shape; self_heal_repoints counter; callback errors. """ @@ -84,6 +85,20 @@ def _agent_message(text: str = "Hello") -> dict: } +def _user_message(text: str = "queued wake") -> dict: + return { + "timestamp": "2026-06-17T10:00:01.500Z", + "type": "event_msg", + "payload": { + "type": "user_message", + "message": text, + "images": [], + "local_images": [], + "text_elements": [], + }, + } + + def _token_count_null() -> dict: """token_count with info=null (rate-limit-only event, no usage).""" return { @@ -347,6 +362,23 @@ async def test_nonexistent_file_no_error(self, tmp_path): consumed = await tailer.read_once() assert consumed == 0 + @pytest.mark.asyncio + async def test_forwards_raw_user_message_to_entry_observer(self, transcript): + cb = _Captor() + observed: list[dict] = [] + entry = _user_message("scheduled codex wake") + tailer = CodexTmuxTranscriptTailer( + transcript, + cb, + on_entry=lambda raw: observed.append(raw), + ) + _write_jsonl(transcript, [entry]) + + await tailer.read_once() + + assert observed == [entry] + assert cb.responses == [] + @pytest.mark.asyncio async def test_single_turn_fires_callback_once(self, transcript): """task_complete fires exactly one callback; text==last_agent_message."""