Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ GitHub Releases page; `0.8.0` is the new starting line.

## Unreleased

- Fix queued follow-up input showing a bordered ghost; pressing Enter during an active turn now shows one intentional queued row.

## 0.60.0 (2026-07-18)

- **Leaf subagent prompt profile.** All 12 built-in subagent roles (implementer,
Expand Down
2 changes: 2 additions & 0 deletions docs/en/release-notes/changelog.md
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,8 @@ GitHub Releases page; `0.8.0` is the new starting line.

## Unreleased

- Fix queued follow-up input showing a bordered ghost; pressing Enter during an active turn now shows one intentional queued row.

## 0.60.0 (2026-07-18)

- **Leaf subagent prompt profile.** All 12 built-in subagent roles (implementer,
Expand Down
5 changes: 2 additions & 3 deletions src/pythinker_code/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -987,9 +987,8 @@ class TUIConfig(BaseModel):
sticky_input: bool = Field(
default=True,
description=(
"Keep the prompt composer pinned to the bottom of the terminal while "
"an agent turn is running. Disable for terminals that mishandle "
"prompt_toolkit fullscreen rendering."
"Legacy compatibility setting retained for backward compatibility. "
"The running prompt composer stays inline and available during active turns."
),
)
code_theme: str = Field(
Expand Down
27 changes: 1 addition & 26 deletions src/pythinker_code/ui/shell/prompt.py
Original file line number Diff line number Diff line change
Expand Up @@ -2349,7 +2349,6 @@ def __init__(
# _input_card_hidden_pre_stream.
self._turn_starting: bool = False
self._sticky_input = sticky_input
self._previous_full_screen: bool | None = None
self._latest_todos: tuple[TodoDisplayItem, ...] = ()
self._modal_delegates: list[RunningPromptDelegate] = []
self._shortcut_help_open = False
Expand Down Expand Up @@ -3168,27 +3167,7 @@ def _apply_mode(self, event: KeyPressEvent | None = None) -> None:
def _sync_erase_when_done(self) -> None:
app = getattr(self._session, "app", None)
if app is not None:
app.erase_when_done = getattr(
self, "_mode", PromptMode.AGENT
) == PromptMode.AGENT and not getattr(app, "full_screen", False)

def _set_running_fullscreen(self, active: bool) -> None:
if not getattr(self, "_sticky_input", True):
return
app = getattr(getattr(self, "_session", None), "app", None)
if app is None:
return
if active:
if getattr(self, "_previous_full_screen", None) is None:
self._previous_full_screen = bool(getattr(app, "full_screen", False))
app.full_screen = True
self._sync_erase_when_done()
return
previous = getattr(self, "_previous_full_screen", None)
self._previous_full_screen = None
if previous is not None:
app.full_screen = previous
self._sync_erase_when_done()
app.erase_when_done = getattr(self, "_mode", PromptMode.AGENT) == PromptMode.AGENT

def _active_modal_delegate(self) -> RunningPromptDelegate | None:
modal_delegates = getattr(self, "_modal_delegates", [])
Expand Down Expand Up @@ -4009,7 +3988,6 @@ def mark_turn_starting(self) -> None:
# not cost an extra repaint.
if not self._turn_starting:
self._turn_starting = True
self._set_running_fullscreen(True)
self.invalidate()

def clear_turn_starting(self) -> None:
Expand All @@ -4021,7 +3999,6 @@ def clear_turn_starting(self) -> None:
without reaching into the private ``_turn_starting`` attribute.
"""
self._turn_starting = False
self._set_running_fullscreen(False)
self.invalidate()

def attach_running_prompt(self, delegate: RunningPromptDelegate) -> None:
Expand All @@ -4035,7 +4012,6 @@ def attach_running_prompt(self, delegate: RunningPromptDelegate) -> None:
self._turn_starting = False
self._mode = PromptMode.AGENT
self._apply_mode()
self._set_running_fullscreen(True)
self.invalidate()

def detach_running_prompt(self, delegate: RunningPromptDelegate) -> None:
Expand All @@ -4048,7 +4024,6 @@ def detach_running_prompt(self, delegate: RunningPromptDelegate) -> None:
if previous_mode is not None:
self._mode = previous_mode
self._apply_mode()
self._set_running_fullscreen(False)
self.invalidate()

def attach_modal(self, delegate: RunningPromptDelegate) -> None:
Expand Down
105 changes: 102 additions & 3 deletions tests/e2e/test_shell_pty_prompt_layout_e2e.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,9 +2,9 @@

Unlike the byte-stream PTY helpers, these feed the raw terminal bytes to a pyte
virtual screen so assertions run against the *rendered* frame — the only place
an incomplete-erase "ghost"/duplicate row is visible. They pin Focus TUI
fossilization behavior and the normal prompt card's visible loading/mid-turn
contract.
an incomplete-erase accepted-buffer "ghost"/duplicate row is visible. They pin
Focus TUI fossilization behavior and the normal prompt card's visible
loading/mid-turn contract.

This is a manual/local check, not a CI-enforced one — it is skipped on CI (see
``pytestmark`` below: scripted_echo + prompt_toolkit hang on GitHub Actions'
Expand All @@ -30,6 +30,7 @@
import pytest

from tests.e2e.shell_pty_helpers import (
list_turn_begin_inputs,
make_home_dir,
make_work_dir,
read_until_prompt_ready,
Expand All @@ -49,6 +50,7 @@

_COLS, _ROWS = 120, 40
_PROMPT_TEXT = "this is a prompt to the agent"
_QUEUED_FOLLOW_UP = "queued follow-up ghost regression 7f3a"


def _render(chunks: list[bytes]):
Expand Down Expand Up @@ -83,6 +85,30 @@ def _has_fossil_border_above_content(rows: list[str]) -> bool:
return any(_is_input_card_border(rows[i]) for i in range(echo_i + 1, content_i))


def _queued_text_fossilized_as_card(rows: list[str], text: str) -> bool:
"""True if the queued follow-up rendered as a fossilized accepted-input card.

The queued-input ghost commits the accepted follow-up into scrollback as a
bordered ``● <effort>`` input card, so its text ends up wedged between a card
border directly above and committed ``⏺`` turn content directly below. That
signature excludes the two healthy renderings: the live queued display (text
plus the ``↑ to edit`` hint, with no card border directly above) and the
execute-echo once the queued turn drains (committed like any turn input, again
with no card border directly above). A/B-verified: fires on the pre-fix code
and stays silent on the fixed code across every rendered frame.
"""
for i, row in enumerate(rows):
if text not in row:
continue
border_above = any(_is_input_card_border(rows[j]) for j in range(max(0, i - 2), i))
content_below = any(
rows[j].strip().startswith("⏺") for j in range(i + 1, min(len(rows), i + 3))
)
if border_above and content_below:
return True
return False


def test_focus_tui_hides_files_and_never_fossilizes_prompt(tmp_path: Path) -> None:
write = {
"id": "w1",
Expand Down Expand Up @@ -193,3 +219,76 @@ def test_input_card_stays_visible_during_initial_loading_and_mid_turn(tmp_path:
)
finally:
shell.close()


def test_mid_turn_queued_input_renders_once_and_executes_once(tmp_path: Path) -> None:
slow = {
"id": "queued-slow",
"name": "Shell",
"arguments": json.dumps({"command": "sleep 3"}),
}
config_path = write_scripted_config(
tmp_path,
[
f"tool_call: {json.dumps(slow)}",
"text: First turn finished.",
"text: Queued follow-up executed.",
],
capabilities=["thinking"],
)
work_dir = make_work_dir(tmp_path)
home_dir = make_home_dir(tmp_path)
shell = start_shell_pty(
config_path=config_path,
work_dir=work_dir,
home_dir=home_dir,
yolo=True,
columns=_COLS,
lines=_ROWS,
)
try:
shell.read_until_contains("think first, then code")
read_until_prompt_ready(shell, after=shell.mark())
assert any(_is_input_card_border(row) for row in _render(shell._raw_chunks))

first_turn_mark = shell.mark()
shell.send_line(_PROMPT_TEXT)
shell.read_until_contains("Bash(sleep 3", after=first_turn_mark, timeout=15.0)
shell.send_line(_QUEUED_FOLLOW_UP)

# The queued follow-up must render as the intentional ``❯ … / ↑ to edit``
# row and never fossilize into a bordered accepted-input card (the ghost).
# Steady state legitimately shows the text twice — the live queued display
# and, after drain, the execute-echo — so the guard is "never a fossilized
# accepted-input card", not a raw occurrence count.
queued_hint_seen = False
deadline = time.monotonic() + 12.0
while time.monotonic() < deadline:
shell.read_available(timeout=0.08)
rows = _render(shell._raw_chunks)
joined = "\n".join(rows)

assert not _queued_text_fossilized_as_card(rows, _QUEUED_FOLLOW_UP), (
"queued follow-up fossilized as a bordered ghost card"
)
if _QUEUED_FOLLOW_UP in joined and "↑ to edit · ctrl-s to send immediately" in joined:
queued_hint_seen = True
if "First turn finished." in shell.normalized_text():
break

assert queued_hint_seen, "intentional queued-message row was never rendered"
shell.read_until_contains("Queued follow-up executed.", timeout=15.0)
shell.wait_for_quiet(timeout=6.0, quiet_period=0.3)

assert not _queued_text_fossilized_as_card(_render(shell._raw_chunks), _QUEUED_FOLLOW_UP), (
"queued follow-up fossilized as a bordered ghost card in the settled frame"
)

turn_inputs = list_turn_begin_inputs(home_dir, work_dir)
assert turn_inputs == [_PROMPT_TEXT, _QUEUED_FOLLOW_UP]
assert turn_inputs.count(_QUEUED_FOLLOW_UP) == 1
assert any(_is_input_card_border(row) for row in _render(shell._raw_chunks)), (
"idle input-card border did not return after the queued turn ended"
)
finally:
shell.close()
54 changes: 24 additions & 30 deletions tests/ui_and_conv/test_prompt_tips.py
Original file line number Diff line number Diff line change
Expand Up @@ -1546,7 +1546,7 @@ def test_apply_mode_syncs_erase_when_done_with_current_mode() -> None:
prompt_session._session = cast(
Any,
SimpleNamespace(
app=SimpleNamespace(erase_when_done=False),
app=SimpleNamespace(erase_when_done=False, full_screen=True),
default_buffer=SimpleNamespace(completer=None),
),
)
Expand All @@ -1558,72 +1558,66 @@ def test_apply_mode_syncs_erase_when_done_with_current_mode() -> None:

assert prompt_session._session.default_buffer.completer is prompt_session._agent_mode_completer
assert prompt_session._session.app.erase_when_done is True
assert prompt_session._session.app.full_screen is True

prompt_session._mode = PromptMode.SHELL
prompt_session._apply_mode()

assert prompt_session._session.default_buffer.completer is prompt_session._shell_mode_completer
assert prompt_session._session.app.erase_when_done is False
assert prompt_session._session.app.full_screen is True


def test_attach_running_prompt_enables_erase_when_done_and_detach_restores_state() -> None:
prompt_session = object.__new__(CustomPromptSession)
prompt_session._mode = PromptMode.SHELL
prompt_session._running_prompt_delegate = None
prompt_session._running_prompt_previous_mode = None
prompt_session._sticky_input = False
prompt_session._session = cast(Any, SimpleNamespace(app=SimpleNamespace(erase_when_done=False)))
prompt_session._turn_starting = False
app = SimpleNamespace(
erase_when_done=False,
full_screen=True,
renderer=SimpleNamespace(full_screen=False),
)
prompt_session._session = cast(
Any,
SimpleNamespace(app=app, default_buffer=SimpleNamespace(completer=None)),
)
prompt_session._agent_mode_completer = cast(Any, object())
prompt_session._shell_mode_completer = cast(Any, object())

delegate = _DummyRunningPrompt()
trace: list[tuple[str, object, object, object]] = []

def fake_apply_mode(event=None) -> None:
prompt_session._session.app.erase_when_done = prompt_session._mode == PromptMode.AGENT
trace.append(
(
"apply",
prompt_session._mode,
prompt_session._session.app.erase_when_done,
prompt_session._running_prompt_delegate,
)
)
invalidations: list[tuple[PromptMode, bool, object | None]] = []
initial_screen_modes = (app.full_screen, app.renderer.full_screen)

def fake_invalidate() -> None:
trace.append(
invalidations.append(
(
"invalidate",
prompt_session._mode,
prompt_session._session.app.erase_when_done,
prompt_session._running_prompt_delegate,
)
)

async def fake_prompt_once(*, append_history: bool) -> UserInput:
trace.append(
(
"prompt",
append_history,
prompt_session._session.app.erase_when_done,
prompt_session._running_prompt_delegate,
)
)
return UserInput(mode=PromptMode.AGENT, command="hi", resolved_command="hi", content=[])

prompt_session._apply_mode = fake_apply_mode
prompt_session.invalidate = fake_invalidate

prompt_session.attach_running_prompt(delegate)

assert prompt_session._mode == PromptMode.AGENT
assert prompt_session._running_prompt_delegate is delegate
assert prompt_session._session.app.erase_when_done is True
assert (app.full_screen, app.renderer.full_screen) == initial_screen_modes

prompt_session.detach_running_prompt(delegate)

assert prompt_session._mode == PromptMode.SHELL
assert prompt_session._running_prompt_delegate is None
assert prompt_session._session.app.erase_when_done is False
assert [entry[0] for entry in trace] == ["apply", "invalidate", "apply", "invalidate"]
assert (app.full_screen, app.renderer.full_screen) == initial_screen_modes
assert invalidations == [
(PromptMode.AGENT, True, delegate),
(PromptMode.SHELL, False, None),
]


# ── Prompt async contract ─────────────────────────────────────────────────────
Expand Down
Loading
Loading