test(cli): prove /compress type-ahead queue-drain; map contributor email

- tests/cli/test_compress_type_ahead.py: end-to-end proof of the PR #68284
  docstring claim — a prompt queued into _pending_input while /compress runs
  survives compaction untouched and is the next item process_loop drains,
  i.e. it is processed against the compacted history. Plus a structural
  guard that handle_enter never gates on _command_running /
  _command_blocks_input (read-only enforcement belongs solely to the
  TextArea Condition; a busy-gate in handle_enter would silently drop
  type-ahead input).
- tests/test_cli_manual_compress.py: update the one remaining _busy_command
  stub (added on main after the PR branched) to accept the new
  blocks_input kwarg.
- contributors/emails: map lucas@policastromd.com -> enzo2.

contributors/emails/lucas@policastromd.com

@@ -0,0 +1 @@
+enzo2

tests/cli/test_compress_type_ahead.py

@@ -0,0 +1,150 @@
+"""Type-ahead queue-drain proof for /compress (issue #61042, PR #68284).
+
+PR #68284 made the classic prompt_toolkit CLI keep the composer editable
+while ``/compress`` runs (``_busy_command(..., blocks_input=False)``), with
+the docstring claim that "the queued input is processed against the
+compacted history after the command completes."
+
+These tests pin that claim end-to-end for the classic CLI:
+
+1. ``test_type_ahead_queued_during_compress_becomes_next_prompt`` — runs
+   ``_manual_compress`` on a worker thread (matching the real topology:
+   slash commands execute on ``process_loop``, which is blocked inside
+   ``process_command`` for the duration), submits a type-ahead payload into
+   ``_pending_input`` mid-compression (what ``handle_enter``'s idle branch
+   does — ``_agent_running`` is False while a slash command runs), and
+   asserts that after compression commits the compacted history the queued
+   text is still the next item ``process_loop`` will drain. Nothing inside
+   the compression path may consume or drop it.
+
+2. ``test_handle_enter_never_gates_on_command_running`` — structural
+   invariant: ``handle_enter`` must not consult ``_command_running`` /
+   ``_command_blocks_input``. The composer's read-only state is enforced
+   solely by the TextArea's ``read_only=Condition(_command_blocks_input)``;
+   if a future edit added an early-return in ``handle_enter`` while a
+   command runs, type-ahead submissions would be silently dropped and the
+   drain contract above would break without any other test noticing.
+
+The Ink TUI (ui-tui) needs no equivalent fix: its ``/compress`` is an async
+``session.compress`` RPC and the composer is never made read-only while it
+runs; the gateway already resolves the concurrent-mutation race via the
+``history_version`` guard in ``_compress_session_history``.
+"""
+
+from __future__ import annotations
+
+import ast
+import queue as queue_mod
+import threading
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+
+from tests.cli.test_cli_init import _make_cli
+
+
+def _make_history() -> list[dict[str, str]]:
+    return [
+        {"role": "user", "content": "one"},
+        {"role": "assistant", "content": "two"},
+        {"role": "user", "content": "three"},
+        {"role": "assistant", "content": "four"},
+    ]
+
+
+def test_type_ahead_queued_during_compress_becomes_next_prompt():
+    """Text queued while /compress runs survives compaction and is the next prompt."""
+    shell = _make_cli()
+    history = _make_history()
+    compressed = [
+        {"role": "user", "content": "[summary]"},
+        history[-1],
+    ]
+    shell.conversation_history = history
+    shell.agent = MagicMock()
+    shell.agent.compression_enabled = True
+    shell.agent._cached_system_prompt = ""
+    shell.agent.tools = None
+    shell.agent.session_id = shell.session_id
+
+    compress_started = threading.Event()
+    release_compress = threading.Event()
+    mid_compress = {}
+
+    def _compress(*_args, **_kwargs):
+        compress_started.set()
+        # Hold the "compression in flight" window open until the test has
+        # simulated the user's type-ahead submission.
+        assert release_compress.wait(timeout=10), "test deadlock: never released"
+        return (list(compressed), "")
+
+    shell.agent._compress_context.side_effect = _compress
+
+    with patch("agent.model_metadata.estimate_request_tokens_rough", return_value=100):
+        worker = threading.Thread(target=shell._manual_compress, daemon=True)
+        worker.start()
+        assert compress_started.wait(timeout=10), "compression never started"
+
+        # Mid-compression: composer stays editable (spinner up, input open).
+        mid_compress["running"] = shell._command_running
+        mid_compress["blocks_input"] = shell._command_blocks_input
+
+        # The user types a follow-up and presses Enter. handle_enter's normal
+        # routing puts it on _pending_input (agent idle — slash commands run
+        # on process_loop, not the agent). process_loop is blocked inside
+        # process_command → _manual_compress, so the payload must sit queued.
+        shell._pending_input.put("follow-up prompt drafted during compaction")
+
+        release_compress.set()
+        worker.join(timeout=10)
+
+    assert not worker.is_alive(), "_manual_compress did not finish"
+    assert mid_compress == {"running": True, "blocks_input": False}
+
+    # Compaction committed the compressed transcript...
+    assert shell.conversation_history == compressed
+    # ...and busy state was fully unwound so the next turn renders normally.
+    assert shell._command_running is False
+    assert shell._command_blocks_input is False
+
+    # The queued type-ahead is exactly the next item process_loop drains —
+    # i.e. it becomes the next prompt, processed against the compacted
+    # history. Compression must not consume, reorder, or drop it.
+    assert (
+        shell._pending_input.get_nowait()
+        == "follow-up prompt drafted during compaction"
+    )
+    with pytest.raises(queue_mod.Empty):
+        shell._pending_input.get_nowait()
+
+
+def test_handle_enter_never_gates_on_command_running():
+    """handle_enter must not consult the busy-command flags (drop-proof routing).
+
+    Enter-key routing while a slash command runs must keep flowing into
+    ``_pending_input``; read-only enforcement belongs exclusively to the
+    TextArea's ``read_only=Condition(...)``. A ``_command_running`` /
+    ``_command_blocks_input`` check inside ``handle_enter`` would let a
+    future edit silently drop type-ahead submissions during /compress.
+    """
+    cli_path = Path(__file__).resolve().parents[2] / "cli.py"
+    tree = ast.parse(cli_path.read_text(encoding="utf-8"))
+
+    target = None
+    for node in ast.walk(tree):
+        if isinstance(node, ast.FunctionDef) and node.name == "handle_enter":
+            target = node
+            break
+    assert target is not None, "handle_enter closure not found in cli.py"
+
+    offenders = [
+        node.attr
+        for node in ast.walk(target)
+        if isinstance(node, ast.Attribute)
+        and node.attr in {"_command_running", "_command_blocks_input"}
+    ]
+    assert not offenders, (
+        "handle_enter references busy-command state — Enter routing while a "
+        f"slash command runs risks dropping type-ahead input: {offenders}"
+    )

tests/test_cli_manual_compress.py

@@ -110,7 +110,7 @@ def test_manual_compress_flush_failure_discards_notification(monkeypatch):
     cli.agent.flush_error = RuntimeError("synthetic child flush failure")
     cli.session_id = "old-session"
     cli._pending_title = "old title"
-    cli._busy_command = lambda _message: nullcontext()
+    cli._busy_command = lambda _message, **_kwargs: nullcontext()
 
     cli._manual_compress("/compress")