diff --git a/frontend-svelte/src/pages/Agents.svelte b/frontend-svelte/src/pages/Agents.svelte index b97bcf38..5a2ae3f9 100644 --- a/frontend-svelte/src/pages/Agents.svelte +++ b/frontend-svelte/src/pages/Agents.svelte @@ -261,7 +261,7 @@ function normalizeStreamingSession(agentName, ss) { const stats = ss.stats || {}; const state = ss.connected - ? ((stats.pending_responses || 0) > 0 ? 'busy' : 'connected') + ? (((stats.pending_responses || 0) + (stats.inflight_turns || 0)) > 0 ? 'busy' : 'connected') : 'sleeping'; return { id: `${agentName}-${ss.label}`, diff --git a/frontend-svelte/src/pages/Dashboard.svelte b/frontend-svelte/src/pages/Dashboard.svelte index ba304b8e..d484f616 100644 --- a/frontend-svelte/src/pages/Dashboard.svelte +++ b/frontend-svelte/src/pages/Dashboard.svelte @@ -138,7 +138,7 @@ // Determine status — streaming session is the source of truth let status; - if (mainSession?.connected && (stats.pending_responses || 0) > 0) { + if (mainSession?.connected && ((stats.pending_responses || 0) + (stats.inflight_turns || 0)) > 0) { status = 'online'; // actively processing } else if (mainSession?.connected) { status = 'idle'; // connected but not processing diff --git a/src/pinky_daemon/tmux_dream_runner.py b/src/pinky_daemon/tmux_dream_runner.py index 114978ca..f7c4865c 100644 --- a/src/pinky_daemon/tmux_dream_runner.py +++ b/src/pinky_daemon/tmux_dream_runner.py @@ -44,6 +44,10 @@ def _log(msg: str) -> None: print(msg, file=sys.stderr, flush=True) +class _ReplExitedError(Exception): + """The dream REPL process exited before writing a result file.""" + + @dataclass class TmuxDreamConfig: """Configuration for the tmux-based dream runner.""" @@ -115,14 +119,30 @@ def session_name(self) -> str: # ── tmux plumbing ───────────────────────────────────────── - async def _tmux(self, *args: str) -> tuple[int, str]: - """Run a tmux command; returns (returncode, combined output).""" + async def _tmux(self, *args: str, timeout: float = 5.0) -> tuple[int, str]: + """Run a tmux command; returns (returncode, combined output). + + ``timeout`` defends against a hung tmux server (mirrors the + rails' ``_TmuxControl._run``). On expiry the subprocess is + killed and a nonzero rc is returned so callers handle it like + any other tmux failure instead of hanging the dream task + forever (which would also wedge the #704 overlap guard for + every subsequent nightly fire). + """ proc = await asyncio.create_subprocess_exec( "tmux", *args, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.STDOUT, ) - out, _ = await proc.communicate() + try: + out, _ = await asyncio.wait_for(proc.communicate(), timeout=timeout) + except asyncio.TimeoutError: + try: + proc.kill() + except ProcessLookupError: + pass # process exited on its own between timeout and kill + await proc.wait() + return 124, f"tmux {args[0] if args else ''} timed out after {timeout}s" return proc.returncode or 0, out.decode(errors="replace") def _resolve_binary(self) -> str: @@ -189,6 +209,15 @@ async def run(self, prompt: str, *, system_prompt: str = "") -> RunResult: f"(model={self._config.model or 'default'}, prompt={prompt_path.name})" ) + # Keep the pane around when the claude process exits (crash, bad + # --model, expired auth) - without this tmux reaps the session on + # command exit, so the post-mortem capture-pane below would lose + # the one clue about WHY it died. Best-effort. + await self._tmux( + "set-option", "-w", "-t", self.session_name, "remain-on-exit", "on" + ) + + success = False try: if not await self._wait_ready(): # Proceed anyway: the pty buffers typed input, and CC reads it @@ -219,8 +248,16 @@ async def run(self, prompt: str, *, system_prompt: str = "") -> RunResult: elapsed_ms = int((time.time() - start) * 1000) _log(f"tmux-dream: {self.session_name} done in {elapsed_ms}ms, " f"report={len(output)} chars") + success = True return RunResult(output=output.strip(), exit_code=0, duration_ms=elapsed_ms) + except _ReplExitedError as e: + return RunResult( + output="", + exit_code=1, + error=str(e), + duration_ms=int((time.time() - start) * 1000), + ) except TimeoutError: _, pane = await self._tmux("capture-pane", "-p", "-t", self.session_name) tail = pane.strip()[-500:] @@ -235,6 +272,16 @@ async def run(self, prompt: str, *, system_prompt: str = "") -> RunResult: ) finally: await self._tmux("kill-session", "-t", self.session_name) + # The report is persisted to the dream DB by the caller; the + # prompt file carries raw conversation history. Delete both on + # success so nightly dreams don't grow dreams/ unbounded; keep + # them on failure for post-mortem. + if success: + for p in (prompt_path, result_path): + try: + p.unlink(missing_ok=True) + except OSError as e: + _log(f"tmux-dream: cleanup of {p} failed (ignored): {e}") def _seed_trust(self, project_dir: str) -> bool: """Seed first-run trust flags; seam for tests.""" @@ -267,14 +314,62 @@ async def _wait_ready(self) -> bool: await asyncio.sleep(1.0) return False + async def _repl_alive(self) -> bool | None: + """Tri-state liveness probe for the claude process in the dream pane. + + Returns True while the process is running and False on definitive + death evidence: ``pane_dead=1`` (a dead command under + ``remain-on-exit on``) or the session/server being gone entirely. + Returns None when the probe itself failed (e.g. rc=124 from the + subprocess timeout on a hung tmux server) - indeterminate, NOT + death; one transiently hung probe must not kill the night's dream. + """ + rc, out = await self._tmux( + "display-message", "-p", "-t", self.session_name, "#{pane_dead}" + ) + if rc == 0: + return out.strip() != "1" + low = out.lower() + if "can't find" in low or "no server running" in low: + return False + return None + async def _wait_for_result(self, path: Path, *, deadline: float) -> str: - """Poll for the result file; require a stable non-empty size before reading.""" + """Poll for the result file; require a stable non-empty size before reading. + + Also checks REPL liveness whenever the file isn't growing (absent, + empty, or stalled): a claude process that exits early (binary + crash, expired OAuth, bad --model, OOM) can never produce the + result file, so waiting out the full timeout would declare the + night lost an hour after it actually died. Declaring death needs + definitive probe evidence (pane_dead / session gone) or two + consecutive failed probes - a single indeterminate probe (hung + tmux server) is not enough to abort the run. + """ last_size = -1 + dead_probes = 0 while time.time() < deadline: - if path.exists(): - size = path.stat().st_size - if size > 0 and size == last_size: - return path.read_text(encoding="utf-8", errors="replace") - last_size = size + size = path.stat().st_size if path.exists() else -1 + if size > 0 and size == last_size: + return path.read_text(encoding="utf-8", errors="replace") + growing = size > last_size + last_size = size + if growing: + dead_probes = 0 + else: + alive = await self._repl_alive() + if alive is True: + dead_probes = 0 + else: + dead_probes += 1 + if alive is False or dead_probes >= 2: + _, pane = await self._tmux( + "capture-pane", "-p", "-t", self.session_name + ) + tail = pane.strip()[-500:] + raise _ReplExitedError( + f"dream REPL exited before writing a result file; " + f"pane tail: {tail}" + ) await asyncio.sleep(self._config.poll_interval_s) raise TimeoutError(f"no result file at {path}") diff --git a/src/pinky_daemon/tmux_session.py b/src/pinky_daemon/tmux_session.py index 8029af7d..94e6eaa5 100644 --- a/src/pinky_daemon/tmux_session.py +++ b/src/pinky_daemon/tmux_session.py @@ -267,6 +267,24 @@ def _seed_claude_trust_file(config_path: Path, project_dir: str) -> bool: # either succeeds or hits a permanent failure. _TRANSIENT_RETRY_BACKOFF_SEC = 2.0 +# Bounded retry budget for per-turn delivery attempts that died on the +# tmux command timeout (``_TmuxControl._run``'s 5s subprocess ceiling). +# A momentarily busy tmux server / loaded host is transient; treating it +# as permanent silently dropped the user's message. Kept small because a +# retry after a timeout that landed AFTER the paste could double-paste +# the prompt into the input area. +_DELIVERY_TIMEOUT_RETRY_LIMIT = 3 + +# Capture-pane double-submit guard (see ``_timed_out_turn_landed``). +# ``_PANE_MARKER_CHARS`` is how much of the prompt's first line we look +# for in the pane -- short enough to survive an 80-col pane without +# wrapping, long enough to be distinctive. Markers shorter than +# ``_PANE_MARKER_MIN_CHARS`` are too ambiguous to trust (a false match +# would silently drop the message), so the guard declines and the worker +# falls back to a plain retry. +_PANE_MARKER_CHARS = 40 +_PANE_MARKER_MIN_CHARS = 12 + # Sentinel path used by ``_start_tailer`` when the transcript JSONL # doesn't exist yet (cold-start). The tailer's ``read_once`` treats # the non-existent file as "no data" and waits; once the SessionStart @@ -1401,10 +1419,19 @@ def state(self) -> SessionState: @property def stats(self) -> dict: """Operational snapshot. Keeps the keys callers actually read.""" + # ``pending_responses`` counts ONLY undelivered queue backlog -- + # it is the key session_watchdog's require_backlog gate reads, and + # an in-flight turn must not arm that outer watchdog: it has none + # of ``_inflight_watchdog``'s liveness carve-outs (transcript + # growth, recent background tasks, live_status floor), so counting + # a running turn there would warn/auto-recover mid-turn on any + # long turn. ``inflight_turns`` exposes the pasted-awaiting-stop + # span separately for busy-state consumers (UI badge). return { **self._stats, "state": self.state.value, - "pending_responses": self._processing, + "pending_responses": self._message_queue.qsize(), + "inflight_turns": len(self._inflight_metas), "current_activity": self._current_activity, "current_thinking": self._current_thinking, "activity_log": list(self._activity_log[-20:]), @@ -4055,6 +4082,7 @@ async def _message_worker(self) -> None: (splash dismisses on input focus); we trust that path. """ _log(f"tmux[{self.agent_name}]: message worker started") + delivery_timeouts = 0 try: while self.state == SessionState.CONNECTED: # Only pull a new turn when nothing is inflight. After @@ -4063,6 +4091,7 @@ async def _message_worker(self) -> None: # silently dropped (Murzik #522 round-1). if self._inflight_turn is None: self._inflight_turn = await self._message_queue.get() + delivery_timeouts = 0 turn = self._inflight_turn try: self._processing = True @@ -4090,6 +4119,50 @@ async def _message_worker(self) -> None: await asyncio.sleep(_TRANSIENT_RETRY_BACKOFF_SEC) continue except Exception as e: + # A tmux command timeout (``_run``'s 5s subprocess + # ceiling) is transient - a busy tmux server must not + # cost the user their message. Keep the turn in hand + # and retry with a bounded budget; ``_deliver_turn`` + # raised before appending any meta, so the retry is + # state-clean. + if ( + isinstance(e, TimeoutError) + and delivery_timeouts + 1 < _DELIVERY_TIMEOUT_RETRY_LIMIT + ): + delivery_timeouts += 1 + # DUPLICATE-SUBMIT WINDOW: a timeout on the final + # send-keys Enter can expire after tmux already + # processed the paste+submit; re-pasting would + # then run a side-effecting turn twice. Check the + # pane for the pasted prompt first -- if it is + # there, finish bookkeeping instead of re-pasting + # (an extra Enter submits a parked prompt and is + # a no-op on an empty input box). + if await self._timed_out_turn_landed(turn): + _log( + f"tmux[{self.agent_name}]: delivery timed " + f"out but the prompt reached the pane; " + f"recording delivery instead of re-pasting" + ) + try: + await self._tmux.send_keys("", enter=True) + except Exception as enter_e: + _log( + f"tmux[{self.agent_name}]: post-timeout " + f"submit Enter failed: {enter_e}" + ) + self._finish_turn_delivery(turn) + self._stats["turns"] += 1 + self._inflight_turn = None + continue + _log( + f"tmux[{self.agent_name}]: turn delivery timed " + f"out (attempt {delivery_timeouts}/" + f"{_DELIVERY_TIMEOUT_RETRY_LIMIT}); retrying in " + f"{_TRANSIENT_RETRY_BACKOFF_SEC}s" + ) + await asyncio.sleep(_TRANSIENT_RETRY_BACKOFF_SEC) + continue # Permanent failure (paste-buffer/send-keys failed, # dead-pane, tailer-state corruption, etc.). Drop # the inflight turn so we don't redeliver into a @@ -4113,6 +4186,11 @@ async def _message_worker(self) -> None: and not turn.completion_event.is_set() ): turn.completion_event.set() + # The message is being dropped; tell the chat that + # sent it instead of leaving the user with dead + # silence (daemon-log-only failures are invisible + # from Telegram/Discord). + await self._notify_delivery_failure(turn) self._inflight_turn = None # Task #90: dead-pane/dead-container already scheduled # disconnect from inside _deliver_turn. Exit the worker @@ -4127,6 +4205,73 @@ async def _message_worker(self) -> None: except Exception as e: _log(f"tmux[{self.agent_name}]: worker error: {e}") + async def _notify_delivery_failure(self, turn: _QueuedTurn) -> None: + """Route a delivery-failure notice back to the chat that sent + ``turn``. + + Called when the worker gives up on an external turn (permanent + paste failure or exhausted timeout retries). The message was + already popped from ``_message_queue`` and will not be + redelivered; without this the sender gets no signal at all. + Internal turns have no chat target, so they are skipped. + Failure-tolerant: a broken callback must not take the worker + down with it. + """ + if turn.internal or not self._response_callback: + return + notice = TurnResponse( + agent_name=self.agent_name, + session_id=self.id, + platform=turn.platform, + chat_id=turn.chat_id, + message_id=turn.message_id, + text=( + "[delivery error] Your message could not be delivered to " + "the agent's session and was dropped. Please resend it." + ), + stop_reason="delivery_error", + ) + try: + result = self._response_callback(notice) + if asyncio.iscoroutine(result): + await result + except Exception as e: + _log( + f"tmux[{self.agent_name}]: delivery-failure notice " + f"callback raised: {e}" + ) + + async def _timed_out_turn_landed(self, turn: _QueuedTurn) -> bool: + """Capture-pane check: did a timed-out delivery actually land? + + A tmux command timeout can expire AFTER tmux processed the + command -- notably ``paste_text``'s final send-keys Enter -- so + blindly re-pasting would submit the turn a second time and + side-effecting instructions would run twice. Look for the head + of the prompt's first line in the pane: if it is visible, the + paste reached the pane (parked in the input area or already + submitted into the scrollback) and the worker must NOT re-paste. + + Returns False when the probe fails or the marker is too short + to be unambiguous -- the worker then falls back to a plain + retry, accepting the narrow duplicate window over the certainty + of a dropped message. Best-effort by design: a capture-pane + that itself times out yields False, never an exception. + """ + marker = "" + for line in turn.prompt.splitlines(): + line = line.strip() + if line: + marker = line[:_PANE_MARKER_CHARS] + break + if len(marker) < _PANE_MARKER_MIN_CHARS: + return False + try: + result = await self._tmux.capture_pane() + except Exception: + return False + return result.ok and marker in (result.stdout or "") + def _transcript_recently_grew(self, now: float, window: float) -> bool: """True if the transcript file was written within ``window`` seconds. @@ -4840,13 +4985,23 @@ async def _deliver_turn(self, turn: _QueuedTurn) -> None: f"stderr={result.stderr.strip()!r}" ) + self._finish_turn_delivery(turn) + + def _finish_turn_delivery(self, turn: _QueuedTurn) -> None: + """Post-paste bookkeeping for a turn that reached the pane. + + Factored out of ``_deliver_turn`` so the worker's timeout-retry + path can mark a turn delivered when the capture-pane guard + (``_timed_out_turn_landed``) shows a timed-out paste actually + landed -- without re-pasting it. + """ # #591 P1#2 (Murzik round-2): paste landed. Fire the optional # post-delivery callback (set on wake turns by # ``_enqueue_wake_prompt`` so ``agent_wake`` is logged AFTER the # prompt actually reached the REPL — not at enqueue time). This # guarantees the cycle-gate boundary advances only on confirmed - # delivery: paste-failure (the ``if not result.ok`` branch above) - # raises BEFORE this point, so a wedged paste leaves the + # delivery: paste-failure (``_deliver_turn``'s ``not result.ok`` + # branch) raises BEFORE this point, so a wedged paste leaves the # boundary intact and the next attempt re-emits the directive. # Failure-tolerant: a misbehaving callback must not strand the # delivery, so wrap in try/except. @@ -5195,6 +5350,19 @@ async def attempt_reconnect(self, *, trigger: Trigger = Trigger.BROKER) -> None: SessionState.CONNECTED, trigger=Trigger.INTERNAL, ) + # Re-prime with an orientation wake prompt BEFORE the + # worker starts draining, mirroring force_restart (#589). + # Without this a heartbeat-resurrected agent comes back + # on a session with no saved-state / current-time / + # channel orientation. Reason derivation matches + # force_restart's launch-signal mapping. + if self._last_launch_forced_fresh: + wake_reason = WakeReason.CONTEXT_RESTART + elif self._last_launch_had_prior_transcript: + wake_reason = WakeReason.RESUME + else: + wake_reason = WakeReason.NEW_SESSION + await self._enqueue_wake_prompt(wake_reason, front=True) # Respawn the worker — disconnect() above cancelled it, so # the queue would otherwise have no drainer on success. if not self._worker_task or self._worker_task.done(): @@ -5202,7 +5370,10 @@ async def attempt_reconnect(self, *, trigger: Trigger = Trigger.BROKER) -> None: # Respawn the watchdog too (#560). if not self._watchdog_task or self._watchdog_task.done(): self._watchdog_task = asyncio.create_task(self._inflight_watchdog()) - _log(f"tmux[{self.agent_name}]: reconnected successfully") + _log( + f"tmux[{self.agent_name}]: reconnected successfully " + f"(wake_reason={wake_reason.value})" + ) return except Exception as e: last_error = e diff --git a/src/pinky_daemon/tmux_transcript.py b/src/pinky_daemon/tmux_transcript.py index c34e3253..62ae54eb 100644 --- a/src/pinky_daemon/tmux_transcript.py +++ b/src/pinky_daemon/tmux_transcript.py @@ -287,6 +287,13 @@ def __init__( self._path_discovery = path_discovery self._offset: int = 0 + # Bumped by every path-changing ``set_transcript_path``. Lets + # ``_read_and_dispatch`` detect a concurrent swap that landed + # while it was parked in an awaited turn callback, so it can + # discard the rest of the old file's chunk instead of feeding it + # into the freshly-drained buffer and adding the old chunk's + # byte length to the NEW file's offset. + self._swap_generation: int = 0 self._buffer = _TurnBuffer() self._wake_event = asyncio.Event() self._task: asyncio.Task | None = None @@ -386,6 +393,7 @@ def set_transcript_path( """ if Path(path) != self._path: self._path = Path(path) + self._swap_generation += 1 if seek_to_start: self._offset = 0 else: @@ -565,6 +573,12 @@ async def _read_and_dispatch(self) -> int: if not self._path.exists(): return 0 + # Snapshot for the mid-chunk swap check below. The only awaits in + # this method are the turn callbacks; everything else is sync, so + # a concurrent ``set_transcript_path`` can only land while a + # callback is in flight. + generation = self._swap_generation + size = self._path.stat().st_size if size < self._offset: # File truncated or rotated underneath us. Reset to 0 and @@ -646,6 +660,16 @@ async def _read_and_dispatch(self) -> int: self._active = False self._stats["turns_fired"] += 1 await self._safe_callback(response) + if self._swap_generation != generation: + # ``set_transcript_path`` swapped the watched file + # while the callback was awaited. The rest of this + # chunk belongs to the OLD file and ``_offset`` now + # refers to the NEW one: feeding more lines would + # repollute the drained buffer (#496 Case 2 leak) + # and the offset advance below would corrupt the + # new file's position. Discard and return; the + # swap already armed the wake event. + return bytes_read elif closes_turn: # Cold-start replay: stop_hook_summary appeared but the # buffer is empty (we entered mid-transcript). Drain diff --git a/tests/test_tmux_dream_runner.py b/tests/test_tmux_dream_runner.py index acb9ef43..ab1c87c8 100644 --- a/tests/test_tmux_dream_runner.py +++ b/tests/test_tmux_dream_runner.py @@ -63,11 +63,19 @@ async def test_happy_path_prompt_file_instruction_result(self): fake = _FakeTmux() runner = _runner(tmp, fake) + prompt_content: list[str] = [] + async def write_result(): # Wait until the instruction was sent, then play the dream agent while not fake.named("send-keys"): await asyncio.sleep(0.01) instruction = fake.named("send-keys")[0][-1] + # Snapshot the prompt file mid-run - both files are cleaned + # up after a successful run. + prompt_file = next( + tok for tok in instruction.split() if "prompt-" in tok + ) + prompt_content.append(open(prompt_file).read()) result_path = next( tok for tok in instruction.split() if "result-" in tok ) @@ -82,15 +90,16 @@ async def write_result(): assert result.output == "FINAL DREAM REPORT" # Prompt file holds system prompt + prompt; REPL never saw them - dreams = os.listdir(os.path.join(tmp, "dreams")) - prompt_file = next(f for f in dreams if f.startswith("prompt-")) - content = open(os.path.join(tmp, "dreams", prompt_file)).read() - assert "DREAM INSTRUCTIONS" in content - assert "consolidate the night" in content + assert "DREAM INSTRUCTIONS" in prompt_content[0] + assert "consolidate the night" in prompt_content[0] instruction = fake.named("send-keys")[0][-1] assert "consolidate the night" not in instruction assert "prompt-" in instruction and "result-" in instruction + # Successful runs leave no prompt/result artifacts behind - + # nightly dreams must not grow dreams/ unbounded. + assert os.listdir(os.path.join(tmp, "dreams")) == [] + # Spawn used the model + bypassPermissions; session torn down spawn = fake.named("new-session")[0] assert "--model" in spawn and "claude-sonnet-4-6" in spawn @@ -134,6 +143,203 @@ def test_session_name_is_distinct_from_main_rails(self): runner = TmuxDreamRunner(TmuxDreamConfig(), agent_name="ivan") assert runner.session_name == "pinky-dream-ivan" + @pytest.mark.asyncio + async def test_spawn_sets_remain_on_exit_for_postmortem_capture(self): + """Without remain-on-exit tmux reaps the session the moment the + claude process exits, so the death-diagnostic capture-pane would + report nothing.""" + with tempfile.TemporaryDirectory() as tmp: + fake = _FakeTmux() + runner = _runner(tmp, fake, timeout_s=0.2) + await runner.run("x") + opts = fake.named("set-option") + assert opts and "remain-on-exit" in opts[0] + + @pytest.mark.asyncio + async def test_repl_death_fails_fast_with_pane_diagnostic(self): + """An early claude exit must fail the run promptly with the pane + tail (the clue about WHY it died), not burn the full timeout + polling for a result file that can never appear.""" + + class _DeadReplTmux(_FakeTmux): + async def __call__(self, *args: str) -> tuple[int, str]: + if args[0] == "display-message": + self.calls.append(args) + return 0, "1" # pane_dead=1: process exited + if args[0] == "capture-pane": + self.calls.append(args) + return 0, "Welcome to Claude Code\nerror: invalid API key" + return await super().__call__(*args) + + with tempfile.TemporaryDirectory() as tmp: + fake = _DeadReplTmux() + runner = _runner(tmp, fake, timeout_s=30.0) + import time + + start = time.time() + result = await runner.run("x") + elapsed = time.time() - start + + assert result.exit_code == 1 + assert "exited" in result.error + assert "invalid API key" in result.error + assert elapsed < 5.0, "death must be detected long before timeout" + assert fake.named("kill-session") + # Failure keeps the prompt file for post-mortem. + dreams = os.listdir(os.path.join(tmp, "dreams")) + assert any(f.startswith("prompt-") for f in dreams) + + @pytest.mark.asyncio + async def test_tmux_subprocess_timeout_is_bounded(self, monkeypatch): + """A hung tmux server must not hang the dream task forever: the + subprocess is killed and a nonzero rc comes back.""" + procs = [] + + class _HungProc: + returncode = None + + def __init__(self): + self.killed = False + procs.append(self) + + async def communicate(self): + await asyncio.Event().wait() # never returns + + def kill(self): + self.killed = True + + async def wait(self): + return 0 + + async def fake_exec(*args, **kwargs): + return _HungProc() + + monkeypatch.setattr(asyncio, "create_subprocess_exec", fake_exec) + runner = TmuxDreamRunner(TmuxDreamConfig(), agent_name="ivan") + rc, out = await runner._tmux("has-session", timeout=0.05) + assert rc != 0 + assert "timed out" in out + assert procs and procs[0].killed + + @pytest.mark.asyncio + async def test_single_probe_timeout_does_not_abort_run(self): + """One transiently hung liveness probe (rc=124) is indeterminate, + not death: the run must keep waiting and still pick up the + result.""" + + class _FlakyProbeTmux(_FakeTmux): + def __init__(self): + super().__init__() + self.probes = 0 + + async def __call__(self, *args: str) -> tuple[int, str]: + if args[0] == "display-message": + self.calls.append(args) + self.probes += 1 + if self.probes == 1: + return 124, "tmux display-message timed out after 5.0s" + return 0, "0" # alive + return await super().__call__(*args) + + with tempfile.TemporaryDirectory() as tmp: + fake = _FlakyProbeTmux() + runner = _runner(tmp, fake, timeout_s=30.0) + + async def write_result(): + while not fake.named("send-keys"): + await asyncio.sleep(0.01) + # Let the poller swallow the rc=124 probe before the + # result appears. + while fake.probes < 1: + await asyncio.sleep(0.01) + instruction = fake.named("send-keys")[0][-1] + result_path = next( + tok for tok in instruction.split() if "result-" in tok + ) + with open(result_path, "w") as f: + f.write("REPORT AFTER FLAKY PROBE") + + writer = asyncio.create_task(write_result()) + result = await runner.run("x") + await writer + + assert result.ok + assert result.output == "REPORT AFTER FLAKY PROBE" + assert fake.probes >= 1 + + @pytest.mark.asyncio + async def test_two_consecutive_dead_probes_abort_run(self): + """A tmux server that stays hung is indistinguishable from a dead + session; two consecutive failed probes count as death.""" + + class _AlwaysDeadProbeTmux(_FakeTmux): + async def __call__(self, *args: str) -> tuple[int, str]: + if args[0] == "display-message": + self.calls.append(args) + return 124, "tmux display-message timed out after 5.0s" + return await super().__call__(*args) + + with tempfile.TemporaryDirectory() as tmp: + fake = _AlwaysDeadProbeTmux() + runner = _runner(tmp, fake, timeout_s=30.0) + import time + + start = time.time() + result = await runner.run("x") + elapsed = time.time() - start + + assert result.exit_code == 1 + assert "exited" in result.error + assert elapsed < 5.0, "death must be detected long before timeout" + assert len(fake.named("display-message")) == 2 + + @pytest.mark.asyncio + async def test_zero_byte_result_with_dead_repl_is_detected(self): + """A REPL that creates an empty result file and then dies must be + detected promptly: liveness is probed whenever the file isn't + growing, not only while it is absent.""" + + class _DeadAfterEmptyResultTmux(_FakeTmux): + def __init__(self, dreams_dir: str): + super().__init__() + self.dreams_dir = dreams_dir + + async def __call__(self, *args: str) -> tuple[int, str]: + if args[0] == "display-message": + self.calls.append(args) + results = [ + f for f in os.listdir(self.dreams_dir) + if f.startswith("result-") + ] + # Alive until the 0-byte result appears, then dead. + return (0, "1") if results else (0, "0") + return await super().__call__(*args) + + with tempfile.TemporaryDirectory() as tmp: + fake = _DeadAfterEmptyResultTmux(os.path.join(tmp, "dreams")) + runner = _runner(tmp, fake, timeout_s=30.0) + + async def touch_empty_result(): + while not fake.named("send-keys"): + await asyncio.sleep(0.01) + instruction = fake.named("send-keys")[0][-1] + result_path = next( + tok for tok in instruction.split() if "result-" in tok + ) + open(result_path, "w").close() # 0 bytes, then death + + toucher = asyncio.create_task(touch_empty_result()) + import time + + start = time.time() + result = await runner.run("x") + elapsed = time.time() - start + await toucher + + assert result.exit_code == 1 + assert "exited before writing a result file" in result.error + assert elapsed < 5.0, "death must be detected long before timeout" + @pytest.mark.asyncio async def test_allowlist_is_the_primary_tool_boundary(self): """#708 review (Murzik): the dream prompt embeds raw conversation diff --git a/tests/test_tmux_session.py b/tests/test_tmux_session.py index d99705fb..ffdfcd74 100644 --- a/tests/test_tmux_session.py +++ b/tests/test_tmux_session.py @@ -7430,3 +7430,304 @@ async def test_deliver_turn_no_on_delivered_is_safe() -> None: await ss._deliver_turn(turn) tmux.paste_text.assert_awaited_once() # delivered cleanly, no crash + + +# -------------------------------------------------------------------------- +# Worker transient-timeout retry + delivery-failure notice +# -------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_worker_retries_turn_after_tmux_timeout(monkeypatch) -> None: + """A tmux command timeout (asyncio.TimeoutError from _TmuxControl._run's + 5s subprocess ceiling) is transient: the worker must keep the turn in + hand and retry instead of silently dropping the user's message.""" + monkeypatch.setattr(tmux_session, "_TRANSIENT_RETRY_BACKOFF_SEC", 0) + ss, _ = _make_session(state=SessionState.CONNECTED) + + attempts: list[str] = [] + + async def flaky_deliver(turn): + attempts.append(turn.prompt) + if len(attempts) < 3: + raise asyncio.TimeoutError("tmux server busy") + + ss._deliver_turn = flaky_deliver + ss._message_queue.put_nowait( + _QueuedTurn(prompt="keep me", platform="telegram", chat_id="c", message_id="m") + ) + + worker = asyncio.create_task(ss._message_worker()) + try: + for _ in range(200): + await asyncio.sleep(0.005) + if len(attempts) >= 3 and ss._inflight_turn is None: + break + finally: + worker.cancel() + try: + await worker + except asyncio.CancelledError: + pass + + assert attempts == ["keep me"] * 3, "same turn must be retried, not dropped" + assert ss._stats["turns"] == 1 + assert ss._inflight_turn is None + + +@pytest.mark.asyncio +async def test_worker_gives_up_after_timeout_budget_and_notifies_chat(monkeypatch) -> None: + """When the timeout retry budget is exhausted, the turn is dropped but + the sending chat gets a delivery-failure notice instead of dead + silence.""" + monkeypatch.setattr(tmux_session, "_TRANSIENT_RETRY_BACKOFF_SEC", 0) + ss, _ = _make_session(state=SessionState.CONNECTED) + + notices: list[TurnResponse] = [] + + async def record_notice(resp): + notices.append(resp) + + ss._response_callback = record_notice + + attempts = 0 + + async def always_timeout(turn): + nonlocal attempts + attempts += 1 + raise asyncio.TimeoutError("tmux server busy") + + ss._deliver_turn = always_timeout + ss._message_queue.put_nowait( + _QueuedTurn(prompt="lost", platform="telegram", chat_id="c1", message_id="m1") + ) + + worker = asyncio.create_task(ss._message_worker()) + try: + for _ in range(200): + await asyncio.sleep(0.005) + if notices: + break + finally: + worker.cancel() + try: + await worker + except asyncio.CancelledError: + pass + + assert attempts == tmux_session._DELIVERY_TIMEOUT_RETRY_LIMIT + assert ss._inflight_turn is None + assert len(notices) == 1 + assert notices[0].platform == "telegram" + assert notices[0].chat_id == "c1" + assert notices[0].message_id == "m1" + assert notices[0].stop_reason == "delivery_error" + assert "delivery" in notices[0].text.lower() + + +@pytest.mark.asyncio +async def test_worker_timeout_after_landed_paste_does_not_repaste(monkeypatch) -> None: + """A tmux timeout that expires AFTER the paste+submit actually landed + must not re-paste the turn -- a side-effecting instruction would run + twice. The capture-pane guard sees the prompt in the pane, so the + worker records the delivery and moves on.""" + monkeypatch.setattr(tmux_session, "_TRANSIENT_RETRY_BACKOFF_SEC", 0) + prompt = "please deploy release 26.06.001 to production now" + tmux = _make_mock_tmux() + tmux.paste_text = AsyncMock(side_effect=asyncio.TimeoutError("tmux busy")) + tmux.capture_pane = AsyncMock( + return_value=TmuxCommandResult( + returncode=0, stdout=f"> {prompt}\nesc to interrupt", stderr="" + ) + ) + ss, _ = _make_session(state=SessionState.CONNECTED, tmux=tmux) + ss._message_queue.put_nowait( + _QueuedTurn(prompt=prompt, platform="telegram", chat_id="c", message_id="m") + ) + + worker = asyncio.create_task(ss._message_worker()) + try: + for _ in range(200): + await asyncio.sleep(0.005) + if ss._inflight_metas and ss._inflight_turn is None: + break + finally: + worker.cancel() + try: + await worker + except asyncio.CancelledError: + pass + + assert tmux.paste_text.await_count == 1, "must not re-paste a landed turn" + assert len(ss._inflight_metas) == 1 + assert ss._inflight_metas[0].meta["chat_id"] == "c" + assert ss._stats["turns"] == 1 + assert ss._inflight_turn is None + + +@pytest.mark.asyncio +async def test_worker_timeout_without_landed_paste_retries(monkeypatch) -> None: + """When the pane shows no trace of the prompt after a timeout, the + paste never landed: the worker must re-paste rather than treat the + turn as delivered (a false 'landed' verdict would drop the message).""" + monkeypatch.setattr(tmux_session, "_TRANSIENT_RETRY_BACKOFF_SEC", 0) + prompt = "please deploy release 26.06.001 to production now" + tmux = _make_mock_tmux() + tmux.paste_text = AsyncMock( + side_effect=[asyncio.TimeoutError("tmux busy"), _ok()] + ) + tmux.capture_pane = AsyncMock( + return_value=TmuxCommandResult(returncode=0, stdout="unrelated pane", stderr="") + ) + ss, _ = _make_session(state=SessionState.CONNECTED, tmux=tmux) + ss._message_queue.put_nowait( + _QueuedTurn(prompt=prompt, platform="telegram", chat_id="c", message_id="m") + ) + + worker = asyncio.create_task(ss._message_worker()) + try: + for _ in range(200): + await asyncio.sleep(0.005) + if ss._inflight_metas and ss._inflight_turn is None: + break + finally: + worker.cancel() + try: + await worker + except asyncio.CancelledError: + pass + + assert tmux.paste_text.await_count == 2, "unlanded paste must be retried" + assert len(ss._inflight_metas) == 1 + assert ss._stats["turns"] == 1 + + +@pytest.mark.asyncio +async def test_worker_notifies_chat_on_permanent_delivery_failure() -> None: + """A permanent delivery failure (paste-buffer/send-keys error) must + route a delivery-failure notice to the external sender.""" + ss, _ = _make_session(state=SessionState.CONNECTED) + + notices: list[TurnResponse] = [] + + async def record_notice(resp): + notices.append(resp) + + ss._response_callback = record_notice + + async def boom(turn): + raise RuntimeError("tmux paste-buffer / send-keys failed: rc=1") + + ss._deliver_turn = boom + ss._message_queue.put_nowait( + _QueuedTurn(prompt="x", platform="discord", chat_id="c2", message_id="m2") + ) + + worker = asyncio.create_task(ss._message_worker()) + try: + for _ in range(200): + await asyncio.sleep(0.005) + if notices: + break + finally: + worker.cancel() + try: + await worker + except asyncio.CancelledError: + pass + + assert len(notices) == 1 + assert notices[0].chat_id == "c2" + assert ss._inflight_turn is None + + +@pytest.mark.asyncio +async def test_delivery_failure_notice_skips_internal_turns() -> None: + """Internal turns have no chat target; the notice helper must not + route anything for them.""" + ss, _ = _make_session(state=SessionState.CONNECTED) + + notices: list[TurnResponse] = [] + + async def record_notice(resp): + notices.append(resp) + + ss._response_callback = record_notice + await ss._notify_delivery_failure( + _QueuedTurn(prompt="wake", internal=True, reason="wake_resume") + ) + assert notices == [] + + +# -------------------------------------------------------------------------- +# attempt_reconnect wake-prompt re-prime (#589 parity) +# -------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_attempt_reconnect_enqueues_resume_wake_prompt() -> None: + """attempt_reconnect (the heartbeat-resurrect path) must re-prime the + agent with an orientation wake prompt after respawn, exactly like + connect() and force_restart() do. Pre-fix it respawned the REPL and + restarted the worker but never enqueued a wake prompt, so a + resurrected agent came back orientationless (the #589 symptom on a + third lifecycle path).""" + ss, _ = _make_session(state=SessionState.DEAD) + ss._skip_wake_prompt_for_tests = False + ss._has_prior_transcript = lambda: True + + enqueued: list[tuple[str, bool]] = [] + + async def _record( + prompt, + *, + reason, + wait_for_completion=False, + timeout_sec=None, + front=False, + on_delivered=None, + ): + enqueued.append((reason, front)) + return None + + ss._enqueue_internal_prompt = _record + + import pinky_daemon.tmux_session as ts_mod + original_backoff = ts_mod._RECONNECT_BACKOFF + ts_mod._RECONNECT_BACKOFF = (0,) + try: + await ss.attempt_reconnect() + assert ss.state == SessionState.CONNECTED + finally: + ts_mod._RECONNECT_BACKOFF = original_backoff + + wake = [e for e in enqueued if e[0].startswith("wake_")] + assert len(wake) == 1, f"expected exactly one wake prompt, got {enqueued}" + assert wake[0][0] == "wake_resume" + assert wake[0][1] is True, "reconnect wake prompt must be front-enqueued" + await ss.disconnect() + + +# -------------------------------------------------------------------------- +# stats: pending_responses is backlog-only; inflight_turns is separate +# -------------------------------------------------------------------------- + + +def test_stats_pending_responses_excludes_inflight_turns() -> None: + """pending_responses is the key session_watchdog's require_backlog + gate reads, so it must count ONLY undelivered queue backlog. An + in-flight turn must not arm the outer watchdog (it lacks the inner + _inflight_watchdog's liveness carve-outs and would warn/auto-recover + mid-turn on any long turn); the running span is exposed separately + as inflight_turns for busy-state consumers.""" + ss, _ = _make_session(state=SessionState.CONNECTED) + assert ss.stats["pending_responses"] == 0 + assert ss.stats["inflight_turns"] == 0 + + _seed_inflight(ss, meta={"platform": "t", "chat_id": "c", "message_id": "m"}) + assert ss.stats["pending_responses"] == 0 + assert ss.stats["inflight_turns"] == 1 + + ss._message_queue.put_nowait(_QueuedTurn(prompt="queued")) + assert ss.stats["pending_responses"] == 1 + assert ss.stats["inflight_turns"] == 1 diff --git a/tests/test_tmux_transcript.py b/tests/test_tmux_transcript.py index fe74aa79..71f565e0 100644 --- a/tests/test_tmux_transcript.py +++ b/tests/test_tmux_transcript.py @@ -618,6 +618,58 @@ async def test_set_transcript_path_drains_buffer_across_swap( f"regression has reopened" ) + @pytest.mark.asyncio + async def test_swap_during_turn_callback_aborts_old_chunk( + self, transcript, tmp_path, + ): + """A path-changing ``set_transcript_path`` can land from another + task while ``_read_and_dispatch`` is parked in an awaited turn + callback (late SessionStart hook, #565 first-bind recovery). + + Pre-fix, the read loop kept feeding the REST of the old file's + chunk into the buffer the swap just drained (dead-session text + leaking into the new session) and then added the old chunk's + byte length to the offset the swap just set for the NEW file + (offset corruption: skipped bytes or a bogus shrank-branch + replay). The swap-generation check discards the remainder of + the chunk and leaves the swapped-in offset untouched. + """ + # File A (dying session): two complete turns in one chunk. + _write_jsonl(transcript, [ + _assistant(text="A1"), + _stop_hook_summary(), + _assistant(text="A2 dead-session text"), + _stop_hook_summary(), + ]) + # File B (new session): one turn, to be read from byte 0. + new_path = tmp_path / "session_b.jsonl" + _write_jsonl(new_path, [ + _assistant(text="B1"), + _stop_hook_summary(), + ]) + + responses: list[str] = [] + box: dict = {} + + async def cb(response: TurnResponse) -> None: + responses.append(response.text) + if len(responses) == 1: + # Simulate the concurrent swap landing mid-callback. + box["tailer"].set_transcript_path(new_path, seek_to_start=True) + + tailer = TmuxTranscriptTailer(transcript, cb) + box["tailer"] = tailer + + await tailer.read_once() + # Turn A2 belongs to the dead session and must NOT fire; the + # offset must stay where the swap put it for file B. + assert responses == ["A1"] + assert tailer.offset == 0 + + await tailer.read_once() + assert responses == ["A1", "B1"] + assert tailer.offset == new_path.stat().st_size + class TestTailerBackgroundLoop: """Drive the actual asyncio loop end-to-end (with shortened cadences)."""