diff --git a/CHANGELOG.md b/CHANGELOG.md index c5dc10e8..2daf5c02 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,12 @@ GitHub Releases page; `0.8.0` is the new starting line. ## Unreleased +- Fix the Windows shell UI going blank mid-session (transcript and input box + disappearing until terminal restart): child processes no longer attach to the + interactive console (`CREATE_NO_WINDOW` on Shell-tool, background-task, and + `!` command spawns), and the prompt renderer now forces an absolute repaint + after terminal resizes and failed scrollback handoffs instead of diffing + against a stale frame. - Fix Workflow progress rendering duplicating agents truncated by the per-phase display cap, close leaked `agent()` coroutines when `parallel()` rejects its arguments, and add a 1000-agent lifetime backstop against runaway workflow loops. diff --git a/docs/en/reference/keyboard.md b/docs/en/reference/keyboard.md index ccca682c..8ca0a131 100644 --- a/docs/en/reference/keyboard.md +++ b/docs/en/reference/keyboard.md @@ -17,6 +17,7 @@ Pythinker Code shell mode supports the following keyboard shortcuts. | `Ctrl-V` | Paste (supports images and video files) | | `Ctrl-E` | Expand full approval request content | | `Ctrl-T` | Show/hide the pinned todo list (during a running turn) | +| `Ctrl-L` | Clear and repaint the screen (recovers a blank/corrupted display) | | `1`–`4` | Quick select approval option (`4` for decline with feedback) | | `1`–`5` | Select question option by number | | `Ctrl-D` | Exit Pythinker Code | diff --git a/packages/pythinker-host/src/pythinker_host/local.py b/packages/pythinker-host/src/pythinker_host/local.py index 3dee3358..d703c69b 100644 --- a/packages/pythinker-host/src/pythinker_host/local.py +++ b/packages/pythinker-host/src/pythinker_host/local.py @@ -3,7 +3,6 @@ import asyncio import os import signal -import subprocess from asyncio.subprocess import Process as AsyncioProcess from collections.abc import AsyncGenerator from pathlib import Path, PurePath @@ -30,6 +29,7 @@ StrOrHostPath, ) from pythinker_host.path import HostPath +from pythinker_host.windows import windows_console_detach_flags if TYPE_CHECKING: @@ -198,9 +198,12 @@ async def exec( process_options: dict[str, Any] = {} if os.name == "nt": - process_options["creationflags"] = getattr( - subprocess, "CREATE_NEW_PROCESS_GROUP", 0x00000200 - ) + # CREATE_NO_WINDOW detaches the child from the interactive console + # (it gets its own hidden one): console-API writes, `cls`, or + # SetConsoleMode calls from the child would otherwise bypass the + # stdio pipes and corrupt the parent TUI until terminal restart. + # CREATE_NEW_PROCESS_GROUP keeps kill() semantics unchanged. + process_options["creationflags"] = windows_console_detach_flags() else: process_options["start_new_session"] = True diff --git a/packages/pythinker-host/src/pythinker_host/windows.py b/packages/pythinker-host/src/pythinker_host/windows.py new file mode 100644 index 00000000..2c7555e0 --- /dev/null +++ b/packages/pythinker-host/src/pythinker_host/windows.py @@ -0,0 +1,26 @@ +from __future__ import annotations + +import subprocess + +_CREATE_NO_WINDOW = 0x08000000 +_CREATE_NEW_PROCESS_GROUP = 0x00000200 + + +def windows_console_detach_flags(*, new_process_group: bool = True) -> int: + """Build Windows ``creationflags`` that keep a spawned child off the console. + + ``CREATE_NO_WINDOW`` gives the child its own hidden console: without it, a + child that touches the Win32 console API (``cls``, ``SetConsoleMode``, ...) + bypasses redirected stdio and can blank the parent TUI until the terminal is + restarted. ``CREATE_NEW_PROCESS_GROUP`` is included by default so existing + ``kill()``/``taskkill`` semantics against the child are unaffected; pass + ``new_process_group=False`` for callers that only need console detachment. + + Centralized here so the fallback constants (used when the real Win32 + constants aren't defined, e.g. off-Windows) cannot drift between the + several process-spawn sites that need them. + """ + flags = getattr(subprocess, "CREATE_NO_WINDOW", _CREATE_NO_WINDOW) + if new_process_group: + flags |= getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", _CREATE_NEW_PROCESS_GROUP) + return flags diff --git a/packages/pythinker-host/tests/test_local_host.py b/packages/pythinker-host/tests/test_local_host.py index c1dc6911..8c3c4430 100644 --- a/packages/pythinker-host/tests/test_local_host.py +++ b/packages/pythinker-host/tests/test_local_host.py @@ -257,3 +257,46 @@ def _record_killpg(pgid: int, sig: int) -> None: await process.wait() assert sent and sent[0] == signal_any.SIGKILL + + +async def test_exec_windows_child_gets_hidden_console( + local_host: LocalHost, monkeypatch: pytest.MonkeyPatch +): + """On Windows the child must not attach to the interactive console. + + A child sharing the TUI's console can clear it or reset its modes through + the Win32 console API (bypassing the stdio pipes), blanking the shell UI + until the terminal is restarted. CREATE_NO_WINDOW gives the child its own + hidden console; CREATE_NEW_PROCESS_GROUP keeps kill() semantics unchanged. + """ + from types import SimpleNamespace + + from pythinker_host import local as local_module + + captured: dict[str, Any] = {} + + class _FakePipe: + pass + + class _FakeProcess: + stdin: Any = _FakePipe() + stdout: Any = _FakePipe() + stderr: Any = _FakePipe() + + async def _fake_create_subprocess_exec(*args: Any, **kwargs: Any) -> Any: + captured.update(kwargs) + return _FakeProcess() + + # Swap the module's `os` reference rather than mutating the global + # `os.name`, which corrupts pathlib/pytest path handling mid-run. + monkeypatch.setattr(local_module, "os", SimpleNamespace(name="nt")) + monkeypatch.setattr( + local_module.asyncio, "create_subprocess_exec", _fake_create_subprocess_exec + ) + + await local_host.exec("cmd", "/c", "echo hi") + + flags = captured["creationflags"] + assert flags & 0x00000200 # CREATE_NEW_PROCESS_GROUP + assert flags & 0x08000000 # CREATE_NO_WINDOW + assert "start_new_session" not in captured diff --git a/src/pythinker_code/background/manager.py b/src/pythinker_code/background/manager.py index a1f3c4f8..8da9bde3 100644 --- a/src/pythinker_code/background/manager.py +++ b/src/pythinker_code/background/manager.py @@ -12,6 +12,7 @@ from typing import TYPE_CHECKING, Any, Literal from pythinker_host.local import local_host +from pythinker_host.windows import windows_console_detach_flags from pythinker_code.config import BackgroundConfig from pythinker_code.notifications import NotificationEvent, NotificationManager @@ -243,7 +244,10 @@ def _launch_worker(self, task_dir: Path) -> int: "cwd": str(task_dir), } if os.name == "nt": - kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + # CREATE_NO_WINDOW: don't share the interactive console — a child + # touching it via the Win32 console API bypasses DEVNULL stdio and + # can blank the parent TUI until terminal restart. + kwargs["creationflags"] = windows_console_detach_flags() else: kwargs["start_new_session"] = True diff --git a/src/pythinker_code/background/worker.py b/src/pythinker_code/background/worker.py index 690e0d70..fd3a42d0 100644 --- a/src/pythinker_code/background/worker.py +++ b/src/pythinker_code/background/worker.py @@ -9,6 +9,8 @@ from pathlib import Path from typing import Any +from pythinker_host.windows import windows_console_detach_flags + from pythinker_code.utils.logging import logger from pythinker_code.utils.subprocess_env import get_clean_env, scrub_secret_env @@ -185,7 +187,10 @@ async def _input_loop() -> None: "env": scrub_secret_env(get_clean_env()) if spec.scrub_secrets else get_clean_env(), } if os.name == "nt": - spawn_kwargs["creationflags"] = getattr(subprocess, "CREATE_NEW_PROCESS_GROUP", 0) + # CREATE_NO_WINDOW: don't share the interactive console — a child + # touching it via the Win32 console API bypasses the redirected + # stdio and can blank the parent TUI until terminal restart. + spawn_kwargs["creationflags"] = windows_console_detach_flags() else: spawn_kwargs["start_new_session"] = True diff --git a/src/pythinker_code/ui/shell/__init__.py b/src/pythinker_code/ui/shell/__init__.py index fcb34d08..c28de8cf 100644 --- a/src/pythinker_code/ui/shell/__init__.py +++ b/src/pythinker_code/ui/shell/__init__.py @@ -4,6 +4,7 @@ import asyncio import contextlib import json +import os import re import shlex import textwrap @@ -24,6 +25,7 @@ APITimeoutError, ChatProviderError, ) +from pythinker_host.windows import windows_console_detach_flags from rich import box from rich.align import Align from rich.cells import cell_len @@ -1271,11 +1273,20 @@ def _handler(): try: # TODO: For the sake of simplicity, we now use `create_subprocess_shell`. # Later we should consider making this behave like a real shell. + spawn_kwargs: dict[str, Any] = {} + if os.name == "nt": + # CREATE_NO_WINDOW: don't share the interactive console — a child + # touching it via the Win32 console API bypasses the pipes and + # can blank the TUI until terminal restart. + spawn_kwargs["creationflags"] = windows_console_detach_flags( + new_process_group=False + ) proc = await asyncio.create_subprocess_shell( command, env=get_clean_env(), stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, + **spawn_kwargs, ) stdout_task = asyncio.create_task(_read_stream_limited(proc.stdout, max_output_bytes)) stderr_task = asyncio.create_task(_read_stream_limited(proc.stderr, max_output_bytes)) diff --git a/src/pythinker_code/ui/shell/prompt.py b/src/pythinker_code/ui/shell/prompt.py index f378d961..44268255 100644 --- a/src/pythinker_code/ui/shell/prompt.py +++ b/src/pythinker_code/ui/shell/prompt.py @@ -2554,6 +2554,11 @@ def _(event: KeyPressEvent) -> None: track("shortcut_editor") self._open_in_external_editor(event) + @_kb.add("c-l", eager=True) + def _(event: KeyPressEvent) -> None: + """Erase and fully repaint the screen (recovery from console damage).""" + self._hard_repaint(event) + def _has_staged_suggestion_prefill() -> bool: return bool(getattr(self, "_staged_suggestion_prefill", None)) @@ -3092,6 +3097,21 @@ def _render_shell_prompt_message(self) -> FormattedText: fragments.append(("bold", f"{PROMPT_SYMBOL_SHELL} ")) return fragments + def _hard_repaint(self, event: KeyPressEvent) -> None: + """Erase the screen and absolutely repaint the prompt (Ctrl+L escape hatch). + + Pins prompt_toolkit's default clear-screen behavior explicitly: when the + real screen has diverged from the renderer's frame model (Windows ConPTY + replay, a child process writing to the shared console), the differential + renderer keeps emitting empty diffs and the UI looks blank; this forces + an absolute frame. Explicit so future custom bindings cannot silently + shadow the recovery path. + """ + try: + event.app.renderer.clear() + except Exception as exc: # noqa: BLE001 — recovery must never crash the prompt + logger.debug("Hard repaint (ctrl-l) failed: {}", exc) + def _open_in_external_editor(self, event: KeyPressEvent) -> None: """Open the current buffer content in an external editor.""" from prompt_toolkit.application.run_in_terminal import run_in_terminal diff --git a/src/pythinker_code/ui/shell/slash.py b/src/pythinker_code/ui/shell/slash.py index 707d2b2b..262be8ff 100644 --- a/src/pythinker_code/ui/shell/slash.py +++ b/src/pythinker_code/ui/shell/slash.py @@ -81,6 +81,7 @@ def slash_command_arg_suggestions() -> dict[str, tuple[str, ...]]: ("Ctrl-O", "Edit in external editor ($VISUAL/$EDITOR)"), ("Ctrl-J / Alt-Enter", "Insert newline"), ("Ctrl-V", "Paste (supports images)"), + ("Ctrl-L", "Repaint the screen (recover a blank/corrupted display)"), ("Ctrl-D", "Exit"), ("Ctrl-C", "Interrupt"), ] diff --git a/src/pythinker_code/ui/shell/sync_output.py b/src/pythinker_code/ui/shell/sync_output.py index 2f152303..118fe5b1 100644 --- a/src/pythinker_code/ui/shell/sync_output.py +++ b/src/pythinker_code/ui/shell/sync_output.py @@ -31,7 +31,11 @@ def install_synchronized_output(output: Output) -> bool: """Bracket every flushed frame of *output* in synchronized-update marks. Returns ``True`` when installed (or already installed). Outputs without - the vt100 list buffer (Windows console, dummy outputs) are left untouched. + the vt100 list buffer (legacy ``Win32Output``, dummy outputs) are left + untouched. Note that on VT-capable Windows consoles prompt_toolkit uses + ``Windows10_Output``, which delegates ``_buffer`` to its inner vt100 + output — so those ARE patched, same as Unix. Safe either way: Windows + Terminal expires an unmatched begin-mark after 100ms, xterm.js after 5s. """ if getattr(output, _INSTALLED_MARKER, False): return True diff --git a/src/pythinker_code/ui/shell/visualize/_interactive.py b/src/pythinker_code/ui/shell/visualize/_interactive.py index f876b042..bb1e24ac 100644 --- a/src/pythinker_code/ui/shell/visualize/_interactive.py +++ b/src/pythinker_code/ui/shell/visualize/_interactive.py @@ -196,11 +196,34 @@ def _tick_resize_recovery(self) -> None: self._resize_recovery_remaining = _RESIZE_RECOVERY_FRAMES self._force_refresh = True _handoff_trace(f"RESIZE\t{columns}x{rows}") + self._reset_prompt_renderer("resize") else: _handoff_trace(f"RESIZE_IGNORE\t{columns}x{rows}") if self._resize_recovery_remaining > 0: self._resize_recovery_remaining -= 1 + def _reset_prompt_renderer(self, reason: str) -> None: + """Drop the prompt renderer's diff state so the next redraw repaints every row. + + When the real screen diverges from prompt_toolkit's frame model — + ConPTY rewraps lines on resize, a terminal replays/clears the + viewport, a scrollback handoff dies mid-erase — the differential + renderer keeps emitting empty diffs over a blank screen and the UI + never comes back. Resetting makes the next frame absolute. + """ + from prompt_toolkit.application import get_app_or_none + + app = get_app_or_none() + if app is None: + return + try: + app.renderer.reset() + except Exception as exc: # noqa: BLE001 — recovery must never take down the UI loop + _handoff_trace(f"RENDERER_RESET_FAIL\t{reason}\t{type(exc).__name__}:{exc}") + logger.debug("Prompt renderer reset failed ({}): {}", reason, exc) + else: + _handoff_trace(f"RENDERER_RESET\t{reason}") + def _defer_scrollback_handoff(self) -> bool: """Backpressure: defer permanent scrollback while preamble geometry is unstable.""" return self._resize_recovery_remaining > 0 @@ -244,6 +267,9 @@ async def _run_scrollback_handoff(self, emit: Callable[[], None], *, reason: str emit() except Exception as exc: _handoff_trace(f"HANDOFF_FAIL\t{reason}\t{type(exc).__name__}:{exc}") + # The teardown/erase may have half-completed; force an absolute + # repaint so the prompt recovers instead of diffing a wrong model. + self._reset_prompt_renderer("handoff-fail") raise finally: self._scrollback_handoff_depth -= 1 diff --git a/tests/background/test_manager.py b/tests/background/test_manager.py index 5d2b9f35..3ae41587 100644 --- a/tests/background/test_manager.py +++ b/tests/background/test_manager.py @@ -1622,3 +1622,45 @@ def fail_write_runtime(*_a, **_k): assert events == ["lock_enter", "read", "write_unlocked", "lock_exit"] # The mutation still landed: state reflects the terminal status. assert real_read(task_id).status == "completed" + + +def test_launch_worker_windows_child_gets_hidden_console(monkeypatch): + """The detached worker must not attach to the interactive console on Windows. + + A worker sharing the TUI's console can corrupt it via the Win32 console + API even with DEVNULL stdio; CREATE_NO_WINDOW detaches it. + """ + from pathlib import Path as _Path + + manager = object.__new__(manager_module.BackgroundTaskManager) + monkeypatch.setattr( + manager_module.BackgroundTaskManager, + "_worker_command", + lambda self, task_dir: ["worker"], + ) + + captured: dict[str, object] = {} + + class _FakeProcess: + pid = 4242 + + def _fake_popen(cmd, **kwargs): + captured.update(kwargs) + return _FakeProcess() + + from types import SimpleNamespace + + monkeypatch.setattr(manager_module.subprocess, "Popen", _fake_popen) + # Swap the module's `os` reference rather than mutating the global + # `os.name`, which corrupts pathlib/pytest path handling mid-run. + monkeypatch.setattr(manager_module, "os", SimpleNamespace(name="nt")) + + pid = manager._launch_worker(_Path("/tmp/task")) + + assert pid == 4242 + flags = captured["creationflags"] + assert isinstance(flags, int) + # CREATE_NEW_PROCESS_GROUP resolves to 0 off-Windows (getattr fallback); + # this test only owns the console-detachment bit. + assert flags & 0x08000000 # CREATE_NO_WINDOW + assert "start_new_session" not in captured diff --git a/tests/background/test_worker.py b/tests/background/test_worker.py index 2172fb16..ded013fa 100644 --- a/tests/background/test_worker.py +++ b/tests/background/test_worker.py @@ -171,3 +171,51 @@ def _run(args, **kwargs): ["taskkill", "/PID", "1234", "/T"], ["taskkill", "/PID", "1234", "/T", "/F"], ] + + +@pytest.mark.asyncio +async def test_worker_spawn_detaches_windows_console(runtime, monkeypatch): + """The spawned child must not attach to the interactive console on Windows. + + A child sharing the TUI's console can corrupt it via the Win32 console + API even with redirected stdio; CREATE_NO_WINDOW detaches it. + """ + import os as real_os + from types import SimpleNamespace + + from pythinker_code.background import worker as worker_module + + store = BackgroundTaskStore(runtime.session.context_file.parent / "tasks") + spec = TaskSpec( + id="b5555555", + kind="bash", + session_id=runtime.session.id, + description="echo hello", + tool_call_id="tool-9", + command="echo hello", + shell_name="bash", + shell_path="/bin/bash", + cwd=str(runtime.session.work_dir), + timeout_s=60, + ) + store.create_task(spec) + + captured: dict[str, object] = {} + + async def _capture_and_abort(*args, **kwargs): + captured.update(kwargs) + raise RuntimeError("spawn intercepted by test") + + monkeypatch.setattr(asyncio, "create_subprocess_exec", _capture_and_abort) + # Swap the module's `os` reference rather than mutating the global + # `os.name`, which corrupts pathlib/pytest path handling mid-run. + monkeypatch.setattr(worker_module, "os", SimpleNamespace(name="nt", getpid=real_os.getpid)) + + await run_background_task_worker(store.task_dir(spec.id), heartbeat_interval_ms=50) + + flags = captured["creationflags"] + assert isinstance(flags, int) + assert flags & 0x08000000 # CREATE_NO_WINDOW + assert "start_new_session" not in captured + # The intercepted spawn surfaced as an explicit failure, not silent success. + assert store.merged_view(spec.id).runtime.status == "failed" diff --git a/tests/ui_and_conv/test_prompt_hard_repaint.py b/tests/ui_and_conv/test_prompt_hard_repaint.py new file mode 100644 index 00000000..7eec0d18 --- /dev/null +++ b/tests/ui_and_conv/test_prompt_hard_repaint.py @@ -0,0 +1,36 @@ +from __future__ import annotations + +from types import SimpleNamespace +from typing import cast + +from prompt_toolkit.key_binding import KeyPressEvent + +from pythinker_code.ui.shell import prompt as shell_prompt + + +def test_hard_repaint_clears_renderer() -> None: + """Ctrl+L must trigger an erase + absolute repaint via renderer.clear().""" + prompt_session = object.__new__(shell_prompt.CustomPromptSession) + + calls: list[str] = [] + + class _Renderer: + def clear(self) -> None: + calls.append("clear") + + event = SimpleNamespace(app=SimpleNamespace(renderer=_Renderer())) + prompt_session._hard_repaint(cast(KeyPressEvent, event)) + + assert calls == ["clear"] + + +def test_hard_repaint_survives_renderer_failure() -> None: + """A failing renderer must not crash the prompt (recovery path stays safe).""" + prompt_session = object.__new__(shell_prompt.CustomPromptSession) + + class _BoomRenderer: + def clear(self) -> None: + raise RuntimeError("boom") + + event = SimpleNamespace(app=SimpleNamespace(renderer=_BoomRenderer())) + prompt_session._hard_repaint(cast(KeyPressEvent, event)) # must not raise diff --git a/tests/ui_and_conv/test_shell_bang_spawn.py b/tests/ui_and_conv/test_shell_bang_spawn.py new file mode 100644 index 00000000..e0d7eda1 --- /dev/null +++ b/tests/ui_and_conv/test_shell_bang_spawn.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +import asyncio +from types import SimpleNamespace + +import pytest + +import pythinker_code.ui.shell as shell_module + + +@pytest.mark.asyncio +async def test_bang_command_detaches_windows_console(monkeypatch) -> None: + """`!` foreground commands must not attach to the interactive console on Windows. + + A child sharing the TUI's console can clear it or reset its modes via the + Win32 console API (bypassing the stdio pipes), blanking the shell UI until + the terminal is restarted; CREATE_NO_WINDOW detaches it. + """ + shell = object.__new__(shell_module.Shell) + + captured: dict[str, object] = {} + + async def _capture_and_abort(command: str, **kwargs): + captured.update(kwargs) + raise RuntimeError("spawn intercepted by test") + + monkeypatch.setattr(asyncio, "create_subprocess_shell", _capture_and_abort) + monkeypatch.setattr(shell_module, "install_sigint_handler", lambda loop, handler: lambda: None) + # Swap the module's `os` reference rather than mutating the global + # `os.name`, which corrupts pathlib/pytest path handling mid-run. + monkeypatch.setattr(shell_module, "os", SimpleNamespace(name="nt")) + + await shell._run_shell_command("echo hi") + + flags = captured["creationflags"] + assert isinstance(flags, int) + assert flags & 0x08000000 # CREATE_NO_WINDOW diff --git a/tests/ui_and_conv/test_visualize_running_prompt.py b/tests/ui_and_conv/test_visualize_running_prompt.py index bd09f5b2..2c5a6d73 100644 --- a/tests/ui_and_conv/test_visualize_running_prompt.py +++ b/tests/ui_and_conv/test_visualize_running_prompt.py @@ -2489,3 +2489,81 @@ async def runner(call) -> None: assert "menu for /version" in visible assert "❯ /version" in visible # echo lives inside the panel too assert "menu for /version" not in capsys.readouterr().out + + +def test_reset_prompt_renderer_resets_app_renderer(monkeypatch) -> None: + import prompt_toolkit.application as pt_application + + view = object.__new__(_PromptLiveView) + resets: list[str] = [] + + class _Renderer: + def reset(self) -> None: + resets.append("reset") + + class _App: + renderer = _Renderer() + + monkeypatch.setattr(pt_application, "get_app_or_none", lambda: _App()) + view._reset_prompt_renderer("test") + assert resets == ["reset"] + + +def test_reset_prompt_renderer_survives_no_app_and_renderer_failure(monkeypatch) -> None: + import prompt_toolkit.application as pt_application + + view = object.__new__(_PromptLiveView) + + monkeypatch.setattr(pt_application, "get_app_or_none", lambda: None) + view._reset_prompt_renderer("no-app") # must not raise + + class _BoomRenderer: + def reset(self) -> None: + raise RuntimeError("boom") + + class _App: + renderer = _BoomRenderer() + + monkeypatch.setattr(pt_application, "get_app_or_none", lambda: _App()) + view._reset_prompt_renderer("boom") # must not raise + + +def test_resize_change_forces_absolute_prompt_repaint(monkeypatch) -> None: + view = object.__new__(_PromptLiveView) + view._last_terminal_size = (80, 24) + view._resize_recovery_remaining = 0 + view._force_refresh = False + view._current_terminal_size = lambda: (100, 30) # type: ignore[method-assign] + + resets: list[str] = [] + monkeypatch.setattr(view, "_reset_prompt_renderer", lambda reason: resets.append(reason)) + + view._tick_resize_recovery() + + assert resets == ["resize"] + + +@pytest.mark.asyncio +async def test_handoff_failure_forces_absolute_prompt_repaint(monkeypatch) -> None: + class _PromptSession: + def invalidate(self) -> None: + pass + + async def _run_in_terminal(func, *args, **kwargs): # noqa: ANN001, ANN002, ANN003 + raise RuntimeError("terminal suspended") + + monkeypatch.setattr(_interactive_mod, "run_in_terminal", _run_in_terminal) + monkeypatch.setattr(_live_view_mod.console, "_force_terminal", True) + + view = _PromptLiveView( + StatusUpdate(), + prompt_session=cast(Any, _PromptSession()), + steer=lambda _content: None, + ) + resets: list[str] = [] + monkeypatch.setattr(view, "_reset_prompt_renderer", lambda reason: resets.append(reason)) + view._pending_scrollback.append((Text("must survive"), True)) + + await view._flush_pending_scrollback() + + assert resets == ["handoff-fail"]