From 7227b98321b87847aa559e8faf87c732f62ab38d Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Fri, 11 Sep 2026 09:04:00 +0100 Subject: [PATCH 1/2] Read the Codex composer through its idle animation Codex draws an idle animation of braille particles across the composer box, and in the observed layout the particle row above the prompt became the first line of the trailing input block, so no live prompt was recognised and sends were refused with an unrecognisable-prompt error. Drop rows that hold nothing but particles, so they cannot hide the prompt row, and ignore particles when comparing the placeholder that marks an empty composer. Retained composer text keeps every character, braille included, so ownership checks, cleanup and delivery verification are unaffected. --- cookbook/two-agent-chat/peer-chat.py | 11 +++++- cookbook/two-agent-chat/test_peer_chat.py | 47 +++++++++++++++++++++++ 2 files changed, 57 insertions(+), 1 deletion(-) diff --git a/cookbook/two-agent-chat/peer-chat.py b/cookbook/two-agent-chat/peer-chat.py index ab8fc59d..e3a5c85e 100755 --- a/cookbook/two-agent-chat/peer-chat.py +++ b/cookbook/two-agent-chat/peer-chat.py @@ -42,6 +42,12 @@ CODEX_PROMPT_RE = re.compile(r"^[›»][\s ]*(.*?)\s*$") CODEX_SHELL_PROMPT_RE = re.compile(r"^![\s ]*(.*?)\s*$") CODEX_CHOICE_RE = re.compile(r"^\d+\.\s") +# Codex sprays an idle animation of braille particles (U+2800-U+28FF) across the composer +# box, prompt row included. Particles are tolerated in exactly two places: rows that hold +# nothing else are dropped so they cannot hide the prompt row, and the empty-placeholder +# comparison ignores them. Composer content itself stays verbatim, so ownership checks, +# cleanup and delivery verification still see every character a person typed. +CODEX_PARTICLE_RE = re.compile(r"[\u2800-\u28FF]") CODEX_EMPTY_PROMPT = "Ask Codex to do anything" # Codex prefixes footer rows with two spaces. Only the final row is stripped: a # multi-row shortcut overlay is indistinguishable from indented modal choices and @@ -477,6 +483,9 @@ def codex_live_prompt_text(text: str) -> str | None: block = trailing_input_block(text) if not block or any(CODEX_SHELL_PROMPT_RE.match(line) for line in block): return None + block = [line for line in block if CODEX_PARTICLE_RE.sub(" ", line).strip()] + if not block: + return None match = CODEX_PROMPT_RE.match(block[0]) if ( not match @@ -522,7 +531,7 @@ def composer_is_empty(profile: Profile, content: str) -> bool: """Recognise known empty-input content for cleanup, acceptance and Codex preflight.""" joined = " ".join(content.splitlines()) if profile.agent == "codex": - return joined == CODEX_EMPTY_PROMPT + return " ".join(CODEX_PARTICLE_RE.sub(" ", joined).split()) == CODEX_EMPTY_PROMPT return joined in CLAUDE_EMPTY_PROMPTS or bool( CLAUDE_STARTUP_HINT_RE.fullmatch(joined) ) diff --git a/cookbook/two-agent-chat/test_peer_chat.py b/cookbook/two-agent-chat/test_peer_chat.py index df81e42d..83b2f21a 100644 --- a/cookbook/two-agent-chat/test_peer_chat.py +++ b/cookbook/two-agent-chat/test_peer_chat.py @@ -256,6 +256,53 @@ def test_wrapped_empty_composer_placeholder_is_known_empty(self) -> None: self.assertEqual(content, "Ask Codex to do\nanything") self.assertTrue(COMPOSER_IS_EMPTY(PROFILES["codex"], content)) + def test_idle_animation_above_the_prompt_is_ignored(self) -> None: + screen = ( + " \u2801 \u2808 \u2804 \u2840 \u2808\n" + "\u203a\u2801Ask Codex to do anything\u2840 \u2808 \u2801\n" + " \u2840 \u2804 \u2802\n" + f"{CODEX_FOOTER}" + ) + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + self.assertIsNotNone(content) + self.assertTrue(COMPOSER_IS_EMPTY(PROFILES["codex"], content)) + + def test_idle_animation_does_not_hide_a_draft(self) -> None: + screen = ( + " \u2801 \u2808\n" + "\u203aChat from Claude: ping\n" + f"{CODEX_FOOTER}" + ) + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + self.assertEqual(content, "Chat from Claude: ping") + self.assertFalse(COMPOSER_IS_EMPTY(PROFILES["codex"], content)) + + def test_composer_content_keeps_every_typed_character(self) -> None: + screen = f"\u203aChat from Claude: explain \u2801 please\n{CODEX_FOOTER}" + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + self.assertEqual(content, "Chat from Claude: explain \u2801 please") + + def test_foreign_character_beside_owned_text_is_not_owned(self) -> None: + screen = f"\u203aChat from Claude: ping\u2801\n{CODEX_FOOTER}" + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: ping" + + self.assertFalse(COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned)) + + def test_idle_animation_over_a_dialog_is_refused(self) -> None: + screen = ( + "\u203a\u2801Ask Codex to do anything\u2840 \u2808\n" + "1. yes\u2801\n" + "2. no\n" + f"{CODEX_FOOTER}" + ) + + self.assertIsNone(LIVE_PROMPT_TEXT(PROFILES["codex"], screen)) + def test_shell_mode_is_not_a_prompt(self) -> None: screen = f"! ls -la\n{CODEX_FOOTER}" From ec99ee0f1ebe1bf6cfcd846a87ee0c37c265e829 Mon Sep 17 00:00:00 2001 From: Dmitry Verkhoturov Date: Fri, 11 Sep 2026 17:10:39 +0100 Subject: [PATCH 2/2] Verify typing through the Codex idle animation Dropping particle-only rows let the prompt be recognised, but a send still failed once text was in the composer: the animation keeps drawing single particles over typed characters for seconds at a time, so the typed chunk never matched, no two screen reads stayed identical long enough to count as settled, cleanup could not locate the owned text, and a queue notice with no blank row under it hid the prompt after submit. For the Codex pane only, compare screen rows with a particle standing for the one cell it covers, or for an empty cell when it trails the text; treat the cell after the prompt marker and the two-cell indent of wrapped rows as decoration; judge a typed chunk settled once the expected text has matched for the settle delay; judge cleanup by the owned spans and cursor column; read the empty placeholder through particles; and treat particle-only rows as blank when the trailing input block is delimited. Claude's composer is still compared verbatim. A particle can hide a wrong character until it moves, so the pre-submit check now verifies the visible body against the intended text again, asking only for the visible suffix the final chunk needed so a long body whose beginning scrolled away still submits; a wrong or foreign character that is visible still fails verification and cleanup. A particle that covers the only character of a wrapped row hides that row, and such a send fails closed until the particle moves. --- cookbook/two-agent-chat/peer-chat.py | 280 +++++++++++++--- cookbook/two-agent-chat/test_peer_chat.py | 368 ++++++++++++++++++++++ 2 files changed, 606 insertions(+), 42 deletions(-) diff --git a/cookbook/two-agent-chat/peer-chat.py b/cookbook/two-agent-chat/peer-chat.py index e3a5c85e..07b22668 100755 --- a/cookbook/two-agent-chat/peer-chat.py +++ b/cookbook/two-agent-chat/peer-chat.py @@ -13,8 +13,9 @@ import tempfile import time import unicodedata -from collections.abc import Iterator +from collections.abc import Callable, Iterator from dataclasses import dataclass, replace +from functools import partial from pathlib import Path from typing import Any @@ -39,15 +40,21 @@ # the glyph to `»` (codex-rs/tui/src/bottom_pane/effort_ignition.rs). Both prompt patterns # are anchored at column zero so prompt-shaped output cannot be mistaken for live input. # Shell mode shows `!` and is deliberately not matched, so nothing is ever typed there. -CODEX_PROMPT_RE = re.compile(r"^[›»][\s ]*(.*?)\s*$") +CODEX_PROMPT_RE = re.compile(r"^[›»][\s \u2800-\u28FF]?[\s ]*(.*?)\s*$") CODEX_SHELL_PROMPT_RE = re.compile(r"^![\s ]*(.*?)\s*$") CODEX_CHOICE_RE = re.compile(r"^\d+\.\s") # Codex sprays an idle animation of braille particles (U+2800-U+28FF) across the composer -# box, prompt row included. Particles are tolerated in exactly two places: rows that hold -# nothing else are dropped so they cannot hide the prompt row, and the empty-placeholder -# comparison ignores them. Composer content itself stays verbatim, so ownership checks, -# cleanup and delivery verification still see every character a person typed. +# box, prompt row included, and a single particle can sit in one cell for seconds, so no +# burst of reads sees the text underneath. For the Codex pane only: rows that hold nothing +# but particles count as blank, so they neither hide the prompt row nor glue a notice to +# it; the cell after the prompt marker and the two-cell indent of wrapped rows are +# decoration; everywhere else a particle stands for the one cell it covers, or for an +# empty cell when it trails the text (row_matches, trailing_particle_variants), and the +# empty placeholder is read the same way. A particle can hide a wrong character until it +# moves, which is why send() verifies the whole body again just before Return. Claude's +# composer shows no animation and is compared verbatim. CODEX_PARTICLE_RE = re.compile(r"[\u2800-\u28FF]") +CODEX_INDENT_RE = re.compile(r"^[\s\u2800-\u28FF]{0,2}") CODEX_EMPTY_PROMPT = "Ask Codex to do anything" # Codex prefixes footer rows with two spaces. Only the final row is stripped: a # multi-row shortcut overlay is indistinguishable from indented modal choices and @@ -381,6 +388,8 @@ def _pane_text_unchecked( ) + + def cursor_column( sid: str, profile: Profile, window: str | None = None ) -> int: @@ -465,16 +474,21 @@ def composer_probe_marker(text: str) -> str: index += 1 +def codex_row_is_blank(line: str) -> bool: + """A row holding nothing but whitespace and idle-animation particles.""" + return not CODEX_PARTICLE_RE.sub(" ", line).strip() + + def trailing_input_block(text: str) -> list[str]: lines = text.splitlines()[-BOX_LINES:] - while lines and not lines[-1].strip(): + while lines and codex_row_is_blank(lines[-1]): lines.pop() if lines and CODEX_FOOTER_RE.match(lines[-1]): lines.pop() - while lines and not lines[-1].strip(): + while lines and codex_row_is_blank(lines[-1]): lines.pop() start = len(lines) - while start and lines[start - 1].strip(): + while start and not codex_row_is_blank(lines[start - 1]): start -= 1 return lines[start:] @@ -483,17 +497,28 @@ def codex_live_prompt_text(text: str) -> str | None: block = trailing_input_block(text) if not block or any(CODEX_SHELL_PROMPT_RE.match(line) for line in block): return None - block = [line for line in block if CODEX_PARTICLE_RE.sub(" ", line).strip()] + block = [line for line in block if not codex_row_is_blank(line)] if not block: return None match = CODEX_PROMPT_RE.match(block[0]) if ( not match or CODEX_CHOICE_RE.match(match.group(1)) - or any(not line[:1].isspace() for line in block[1:]) + or any( + not (line[:1].isspace() or CODEX_PARTICLE_RE.match(line[:1])) + for line in block[1:] + ) ): return None - content = [match.group(1), *(line.strip() for line in block[1:])] + # Wrapped rows carry a two-cell indent, and whatever the animation drew there is + # decoration. A particle on the first text cell may cover a character, so it stays + # for row_matches to judge, and trailing particles stay for the same reason. A row + # whose only character is covered looks like an animation row and is dropped, so + # such a send fails closed until the particle moves; nothing tells the two apart. + content = [ + match.group(1), + *(CODEX_INDENT_RE.sub("", line).strip() for line in block[1:]), + ] return "\n".join(part for part in content if part) @@ -527,11 +552,39 @@ def live_prompt_text(profile: Profile, text: str) -> str | None: return claude_live_prompt_text(text) +def codex_placeholder_shown(content: str) -> bool: + """The empty-composer placeholder, read through the idle animation. + + Rows are joined with the space that wrapping consumed. A particle may stand for + the placeholder character under it or for an empty cell beside the text, so the + walk lets every particle take either role and every other cell must match. + """ + shown = " ".join(row.strip() for row in content.splitlines()) + expected = CODEX_EMPTY_PROMPT + reached = {0} + for char in shown: + advanced = set() + for index in reached: + if CODEX_PARTICLE_RE.match(char): + advanced.add(index) + if index < len(expected): + advanced.add(index + 1) + elif index < len(expected) and expected[index] == char: + advanced.add(index + 1) + elif index == len(expected) and char == " ": + # Spaces between the particles that trail the placeholder. + advanced.add(index) + reached = advanced + if not reached: + return False + return len(expected) in reached + + def composer_is_empty(profile: Profile, content: str) -> bool: """Recognise known empty-input content for cleanup, acceptance and Codex preflight.""" - joined = " ".join(content.splitlines()) if profile.agent == "codex": - return " ".join(CODEX_PARTICLE_RE.sub(" ", joined).split()) == CODEX_EMPTY_PROMPT + return codex_placeholder_shown(content) + joined = " ".join(content.splitlines()) return joined in CLAUDE_EMPTY_PROMPTS or bool( CLAUDE_STARTUP_HINT_RE.fullmatch(joined) ) @@ -555,19 +608,36 @@ def wait_for_composer_change( previous: tuple[str, int], settle_delay: float, window: str | None = None, + matches: Callable[[str], bool] | None = None, ) -> tuple[str, int] | None: - """Wait for a changed composer state to remain stable.""" + """Wait for a changed composer state to remain stable. + + With ``matches`` the composer counts as settled once the predicate has held + for ``settle_delay``, whatever the idle animation does to the text meanwhile. + Without it, a state that differs from ``previous`` has to stay the same for + that long, a particle on either side standing for the cell it covers. + """ deadline = time.monotonic() + PROBE_TIMEOUT + tolerant = profile.agent == "codex" stable_state: tuple[str, int] | None = None stable_since: float | None = None while True: state = composer_state(sid, profile, window) now = time.monotonic() - if state is not None and state != previous: - if state != stable_state: + if matches is not None: + good = state is not None and matches(state[0]) + else: + good = state is not None and not same_composer_state( + state, previous, tolerant + ) + if good: + if stable_since is None or ( + matches is None + and not same_composer_state(state, stable_state, tolerant) + ): stable_state = state stable_since = now - elif stable_since is not None and now - stable_since >= settle_delay: + elif now - stable_since >= settle_delay: return state else: stable_state = None @@ -589,6 +659,7 @@ def type_body( if progress is None: progress = BodyProgress() marker = composer_probe_marker(text) + tolerant = profile.agent == "codex" chunks = text_chunks(text, TYPE_CHUNK_BYTES - len(marker.encode("utf-8"))) state = initial expected = "" @@ -600,7 +671,17 @@ def type_body( try: type_text(sid, profile, chunk + marker, window) changed = wait_for_composer_change( - sid, profile, state, CHUNK_SETTLE_DELAY, window + sid, + profile, + state, + CHUNK_SETTLE_DELAY, + window, + matches=partial( + chunk_is_visible, + expected=marked, + chunk=chunk + marker, + tolerant=tolerant, + ), ) if changed is None: raise ComposerDirty( @@ -609,7 +690,7 @@ def type_body( marked, ) if not composer_has_expected_tail( - changed[0], marked, chunk + marker + changed[0], marked, chunk + marker, tolerant ): raise ComposerDirty( f"message chunk {index + 1}/{len(chunks)} is incomplete in " @@ -623,7 +704,18 @@ def type_body( else CHUNK_SETTLE_DELAY ) unmarked = wait_for_composer_change( - sid, profile, changed, settle_delay, window + sid, + profile, + changed, + settle_delay, + window, + matches=partial( + chunk_is_visible, + expected=attempted, + chunk=chunk, + tolerant=tolerant, + marker=marker, + ), ) if unmarked is None or marker in unmarked[0]: raise ComposerDirty( @@ -631,7 +723,9 @@ def type_body( "not confirmably removed; submit withheld", marked, ) - if not composer_has_expected_tail(unmarked[0], attempted, chunk): + if not composer_has_expected_tail( + unmarked[0], attempted, chunk, tolerant + ): raise ComposerDirty( f"message chunk {index + 1}/{len(chunks)} changed during " "marker removal; submit withheld", @@ -663,20 +757,107 @@ def type_body( return state +def trailing_particle_variants(row: str) -> list[str]: + """The row as shown, then with one more trailing particle removed each time. + + A trailing particle may cover the last typed character or an empty cell beside + the text, and only the source text tells which, so every reading is offered. + """ + row = row.rstrip() + variants = [row] + while row and CODEX_PARTICLE_RE.match(row[-1]): + # Particles drawn beyond the text sit after the spaces of empty cells. + row = row[:-1].rstrip() + variants.append(row) + return variants + + +def row_readings(row: str, tolerant: bool) -> list[str]: + return trailing_particle_variants(row) if tolerant else [row] + + +def row_matches(expected: str, row: str, tolerant: bool = False) -> bool: + """Compare one screen row with source text of the same length. + + Codex draws its idle animation over the composer, so with ``tolerant`` a braille + particle stands for the one cell it covers; every other cell must match exactly, + and a row of the wrong length never matches. Claude's composer shows no + animation, so its rows are compared verbatim. + """ + if not tolerant: + return expected == row + if len(expected) != len(row): + return False + return all( + wanted == seen or CODEX_PARTICLE_RE.match(seen) is not None + for wanted, seen in zip(expected, row) + ) + + +def same_composer_text(first: str, second: str, tolerant: bool = False) -> bool: + """Equal composer text; with ``tolerant`` a particle on either side is its cell.""" + if not tolerant: + return first == second + rows_first = first.splitlines() or [first] + rows_second = second.splitlines() or [second] + if len(rows_first) != len(rows_second): + return False + return all( + any( + len(one) == len(other) + and all( + left == right + or CODEX_PARTICLE_RE.match(left) is not None + or CODEX_PARTICLE_RE.match(right) is not None + for left, right in zip(one, other) + ) + for one in trailing_particle_variants(row_one) + for other in trailing_particle_variants(row_other) + ) + for row_one, row_other in zip(rows_first, rows_second) + ) + + +def same_composer_state( + first: tuple[str, int] | None, + second: tuple[str, int] | None, + tolerant: bool = False, +) -> bool: + if first is None or second is None: + return first is second + return first[1] == second[1] and same_composer_text(first[0], second[0], tolerant) + + +def chunk_is_visible( + content: str, + expected: str, + chunk: str, + tolerant: bool = False, + marker: str = "", +) -> bool: + """The typed chunk shows in the composer, without a marker that should be gone.""" + if marker and marker in content: + return False + return composer_has_expected_tail(content, expected, chunk, tolerant) + + def composer_has_expected_tail( content: str, expected: str, chunk: str, + tolerant: bool = False, ) -> bool: """Match a source suffix while allowing visual line wrapping.""" positions = {len(expected)} rows = content.splitlines() or [content] for index in range(len(rows) - 1, -1, -1): - row = rows[index] + readings = row_readings(rows[index], tolerant) matched = { end - len(row) for end in positions - if end >= len(row) and expected[end - len(row) : end] == row + for row in readings + if end >= len(row) + and row_matches(expected[end - len(row) : end], row, tolerant) } if not matched: return False @@ -723,7 +904,10 @@ def chunk_tail_is_ambiguous(expected: str, chunk: str) -> bool: def composer_owned_spans( - content: str, owned_text: str, allowed_ends: set[int] + content: str, + owned_text: str, + allowed_ends: set[int], + tolerant: bool = False, ) -> set[tuple[int, int]]: """Locate visible contiguous owned text ending at an allowed prefix boundary.""" if not content: @@ -739,12 +923,13 @@ def composer_owned_spans( ) rows = content.splitlines() or [content] for index in range(len(rows) - 1, -1, -1): - row = rows[index] + readings = row_readings(rows[index], tolerant) matched = { (position - len(row), end) for position, end in positions + for row in readings if position >= len(row) - and owned_text[position - len(row) : position] == row + and row_matches(owned_text[position - len(row) : position], row, tolerant) } if not matched: return set() @@ -770,9 +955,8 @@ def wait_for_cleanup_state( ) -> tuple[tuple[str, int], set[tuple[int, int]]] | None: """Wait for a backspace batch to reach a stable observable state.""" deadline = time.monotonic() + PROBE_TIMEOUT - stable: tuple[ - tuple[str, int] | None, frozenset[tuple[int, int]] - ] | None = None + tolerant = profile.agent == "codex" + stable: tuple[int | None, frozenset[tuple[int, int]]] | None = None stable_since: float | None = None while True: state = composer_state(sid, profile, window) @@ -784,15 +968,17 @@ def wait_for_cleanup_state( if is_empty and accept_empty: return state, set() spans = ( - composer_owned_spans(state[0], owned_text, allowed_ends) + composer_owned_spans(state[0], owned_text, allowed_ends, tolerant) if state is not None else set() ) - if state == previous or (is_empty and not accept_empty): + if same_composer_state(state, previous, tolerant) or ( + is_empty and not accept_empty + ): stable = None stable_since = None else: - candidate = state, frozenset(spans) + candidate = (state[1] if state is not None else None), frozenset(spans) if candidate != stable: stable = candidate stable_since = now @@ -875,7 +1061,6 @@ def wait_for_accepted( ) -> bool: """Wait until submission clears the composed state and restores column 2.""" deadline = time.monotonic() + PROBE_TIMEOUT - stable_state: tuple[str, int] | None = None stable_since: float | None = None while True: state = composer_state(sid, profile, window) @@ -887,16 +1072,11 @@ def wait_for_accepted( and composer_is_empty(profile, state[0]) ) if accepted: - if state != stable_state: - stable_state = state + if stable_since is None: stable_since = now - elif ( - stable_since is not None - and now - stable_since >= CHUNK_SETTLE_DELAY - ): + elif now - stable_since >= CHUNK_SETTLE_DELAY: return True else: - stable_state = None stable_since = None if now >= deadline: return False @@ -995,8 +1175,24 @@ def send( composed = type_body( sid, profile, typed, initial, window, progress ) + # A particle can hide a wrong character until it moves, so the visible + # body is verified against the intended text once more with the latest + # reading. A long body may have scrolled its beginning out of the + # composer, so the check asks for the visible suffix the final chunk + # needed, no more. before_submit = composer_state(sid, profile, window) - if before_submit != composed: + tolerant = profile.agent == "codex" + final_chunk = text_chunks( + typed, + TYPE_CHUNK_BYTES + - len(composer_probe_marker(typed).encode("utf-8")), + )[-1] + if before_submit is None or not ( + same_composer_state(before_submit, composed, tolerant) + and composer_has_expected_tail( + before_submit[0], typed, final_chunk, tolerant + ) + ): raise ComposerDirty( "target composer changed before submit; submit withheld", typed, diff --git a/cookbook/two-agent-chat/test_peer_chat.py b/cookbook/two-agent-chat/test_peer_chat.py index 83b2f21a..e6c58190 100644 --- a/cookbook/two-agent-chat/test_peer_chat.py +++ b/cookbook/two-agent-chat/test_peer_chat.py @@ -20,6 +20,7 @@ CLEAR_COMPOSER = SCRIPT["clear_composer"] COMPOSER_DIRTY = SCRIPT["ComposerDirty"] COMPOSER_HAS_EXPECTED_TAIL = SCRIPT["composer_has_expected_tail"] +COMPOSER_OWNED_SPANS = SCRIPT["composer_owned_spans"] COMPOSER_IS_EMPTY = SCRIPT["composer_is_empty"] COMPOSER_PROBE_MARKER = SCRIPT["composer_probe_marker"] COMPOSER_STATE = SCRIPT["composer_state"] @@ -39,6 +40,7 @@ TYPE_CHUNK_BYTES = SCRIPT["TYPE_CHUNK_BYTES"] WAIT_FOR_ACCEPTED = SCRIPT["wait_for_accepted"] WAIT_FOR_COMPOSER_CHANGE = SCRIPT["wait_for_composer_change"] +WAIT_FOR_CLEANUP_STATE = SCRIPT["wait_for_cleanup_state"] CLAUDE_PROFILE = PROFILES["claude"] PARSE_ARGS = SCRIPT["parse_args"] PREPARE_MESSAGE = SCRIPT["prepare_message"] @@ -48,6 +50,7 @@ RUN_MAIN = SCRIPT["run_main"] TARGET_PROFILE = SCRIPT["target_profile"] PANE_TEXT = SCRIPT["pane_text"] +NORMALIZE = SCRIPT["normalize"] TREE = SCRIPT["tree"] TYPE_TEXT = SCRIPT["type_text"] RULE = "─" * 40 @@ -256,6 +259,24 @@ def test_wrapped_empty_composer_placeholder_is_known_empty(self) -> None: self.assertEqual(content, "Ask Codex to do\nanything") self.assertTrue(COMPOSER_IS_EMPTY(PROFILES["codex"], content)) + def test_placeholder_is_read_through_particles(self) -> None: + for shown in ( + "Ask Co\u2801ex to do anything", + "Ask Codex to do anything\u2801", + "\u2801sk Codex to do anything", + "Ask\u2802Codex to do anything", + "Ask Codex to do\u2801\nanything", + "Ask Codex to d\u2801\nanything", + ): + self.assertTrue(COMPOSER_IS_EMPTY(PROFILES["codex"], shown), shown) + for shown in ( + "Ask Codex to do anythin", + "Ask Codex to do anything!", + "Ask Codex to do\u2801\nanythin", + "Chat from Claude: ping", + ): + self.assertFalse(COMPOSER_IS_EMPTY(PROFILES["codex"], shown), shown) + def test_idle_animation_above_the_prompt_is_ignored(self) -> None: screen = ( " \u2801 \u2808 \u2804 \u2840 \u2808\n" @@ -268,6 +289,21 @@ def test_idle_animation_above_the_prompt_is_ignored(self) -> None: self.assertIsNotNone(content) self.assertTrue(COMPOSER_IS_EMPTY(PROFILES["codex"], content)) + def test_particle_rows_separate_the_prompt_from_a_queue_notice(self) -> None: + screen = ( + "\u2022 Messages to be submitted after next tool call\n" + " \u21b3 Chat from Claude: Review done\n" + " \u2026\n" + " \u2808 \u2801 \u2840\n" + "\u203a\u2801Ask Codex to do anything\u2840 \u2808\n" + " \u2840 \u2804\n" + f"{CODEX_FOOTER}" + ) + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + self.assertIsNotNone(content) + self.assertTrue(COMPOSER_IS_EMPTY(PROFILES["codex"], content)) + def test_idle_animation_does_not_hide_a_draft(self) -> None: screen = ( " \u2801 \u2808\n" @@ -286,12 +322,144 @@ def test_composer_content_keeps_every_typed_character(self) -> None: self.assertEqual(content, "Chat from Claude: explain \u2801 please") def test_foreign_character_beside_owned_text_is_not_owned(self) -> None: + screen = f"\u203aChat from Claude: ping!\n{CODEX_FOOTER}" + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: ping" + + self.assertFalse(COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned)) + + def test_particle_beside_owned_text_is_decoration(self) -> None: screen = f"\u203aChat from Claude: ping\u2801\n{CODEX_FOOTER}" content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) owned = "Chat from Claude: ping" + self.assertTrue( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True) + ) + self.assertEqual( + COMPOSER_OWNED_SPANS(content, owned, {len(owned)}, tolerant=True), + {(0, len(owned))}, + ) + + def test_space_separated_trailing_particles_are_decoration(self) -> None: + owned = "Chat from Claude: ping" + for shown in ( + f"\u203aChat from Claude: ping \u2801\n{CODEX_FOOTER}", + f"\u203aChat from Claude: ping\u2801 \u2802\n{CODEX_FOOTER}", + f"\u203aChat from Claude: pin\u2801 \u2802\n{CODEX_FOOTER}", + ): + content = LIVE_PROMPT_TEXT(PROFILES["codex"], shown) + self.assertTrue( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True), shown + ) + self.assertEqual( + COMPOSER_OWNED_SPANS(content, owned, {len(owned)}, tolerant=True), + {(0, len(owned))}, + shown, + ) + + def test_particle_beyond_the_row_end_is_decoration(self) -> None: + screen = ( + f"\u203aChat from Claude: a long reply that wraps onto\n" + f" a second row [peer-check:0] \u2808\n{CODEX_FOOTER}" + ) + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + marked = "Chat from Claude: a long reply that wraps onto a second row [peer-check:0]" + + self.assertTrue( + COMPOSER_HAS_EXPECTED_TAIL(content, marked, marked, tolerant=True) + ) + self.assertEqual( + COMPOSER_OWNED_SPANS(content, marked, {len(marked)}, tolerant=True), + {(0, len(marked))}, + ) + + def test_particle_over_the_last_typed_character_still_verifies(self) -> None: + screen = f"\u203aChat from Claude: pin\u2801\n{CODEX_FOOTER}" + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: ping" + + self.assertTrue( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True) + ) + + def test_particle_over_a_typed_character_still_verifies(self) -> None: + screen = f"\u203aChat from Cl\u2801ude: ping\n{CODEX_FOOTER}" + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: ping" + + self.assertTrue( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True) + ) + self.assertEqual( + COMPOSER_OWNED_SPANS(content, owned, {len(owned)}, tolerant=True), + {(0, len(owned))}, + ) + + def test_wrong_character_under_no_particle_still_fails(self) -> None: + screen = f"\u203aChat from Claude: pinf\n{CODEX_FOOTER}" + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: ping" + + self.assertFalse( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True) + ) + self.assertEqual( + COMPOSER_OWNED_SPANS(content, owned, {len(owned)}, tolerant=True), set() + ) + + def test_particle_cannot_hide_a_missing_character(self) -> None: + screen = f"\u203aChat from Claude: pin\u2801\n{CODEX_FOOTER}" + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: ping!" + + self.assertFalse( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True) + ) + + def test_claude_rows_are_compared_verbatim(self) -> None: + content = "Chat from Codex: p\u2801ng" + owned = "Chat from Codex: ping" + self.assertFalse(COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned)) + self.assertEqual(COMPOSER_OWNED_SPANS(content, owned, {len(owned)}), set()) + + def test_particles_in_the_prompt_gap_and_indent_are_ignored(self) -> None: + screen = ( + "\u203a\u2801Chat from Claude: a long\n" + "\u2810 line here\u2802\n" + f"{CODEX_FOOTER}" + ) + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: a long line here" + + self.assertEqual(content, "Chat from Claude: a long\nline here\u2802") + self.assertTrue( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True) + ) + + def test_particle_on_the_first_wrapped_cell_still_verifies(self) -> None: + screen = ( + "\u203a \u2801hat from Claude: a long\n" + " \u2810ine here\n" + f"{CODEX_FOOTER}" + ) + content = LIVE_PROMPT_TEXT(PROFILES["codex"], screen) + + owned = "Chat from Claude: a long line here" + + self.assertEqual(content, "\u2801hat from Claude: a long\n\u2810ine here") + self.assertTrue( + COMPOSER_HAS_EXPECTED_TAIL(content, owned, owned, tolerant=True) + ) def test_idle_animation_over_a_dialog_is_refused(self) -> None: screen = ( @@ -970,6 +1138,67 @@ def test_unaccepted_submit_is_reported(self) -> None: ): SEND("session-id", CLAUDE_PROFILE, "body") + def test_clipped_composer_before_submit_still_submits(self) -> None: + body = " ".join(f"word{index}" for index in range(120)) + typed = NORMALIZE(CLAUDE_PROFILE, body) + visible = typed[-300:] + type_text = Mock() + replacements = { + "pane_text": Mock(return_value=f"{RULE}\n\u276f \n{RULE}"), + "cursor_column": Mock(return_value=2), + "type_body": Mock(return_value=(visible, 2)), + "composer_state": Mock(return_value=(visible, 2)), + "type_text": type_text, + "wait_for_accepted": Mock(return_value=True), + } + + with patch.dict(SEND.__globals__, replacements): + self.assertEqual(SEND("session-id", CLAUDE_PROFILE, body), len(body)) + + type_text.assert_called_once_with( + "session-id", CLAUDE_PROFILE, CLAUDE_PROFILE.submit, None + ) + + def test_clipped_cyrillic_composer_needs_only_the_final_chunk(self) -> None: + body = " ".join(f"слово{index}" for index in range(90)) + typed = NORMALIZE(PROFILES["codex"], body) + visible = typed[-150:] + type_text = Mock() + replacements = { + "pane_text": Mock(return_value=f"» Ask Codex to do anything\n{CODEX_FOOTER}"), + "cursor_column": Mock(return_value=2), + "type_body": Mock(return_value=(visible, 2)), + "composer_state": Mock(return_value=(visible, 2)), + "type_text": type_text, + "wait_for_accepted": Mock(return_value=True), + } + + with patch.dict(SEND.__globals__, replacements): + self.assertEqual(SEND("session-id", PROFILES["codex"], body), len(body)) + + type_text.assert_called_once_with( + "session-id", PROFILES["codex"], PROFILES["codex"].submit, None + ) + + def test_wrong_character_revealed_before_submit_withholds_return(self) -> None: + typed = NORMALIZE(PROFILES["codex"], "ping") + type_text = Mock() + replacements = { + "pane_text": Mock(return_value=f"» Ask Codex to do anything\n{CODEX_FOOTER}"), + "cursor_column": Mock(return_value=2), + "type_body": Mock(return_value=(typed.replace("ping", "p\u2801ng"), 23)), + "composer_state": Mock(return_value=(typed.replace("ping", "pong"), 23)), + "clear_composer": Mock(return_value=True), + "type_text": type_text, + } + + with patch.dict(SEND.__globals__, replacements), self.assertRaisesRegex( + COMPOSER_DIRTY, "changed before submit" + ): + SEND("session-id", PROFILES["codex"], "ping") + + type_text.assert_not_called() + def test_changed_composer_before_submit_withholds_return(self) -> None: type_text = Mock() replacements = { @@ -1114,6 +1343,141 @@ def test_queued_placeholder_counts_as_accepted(self) -> None: ) +class TypeBodyThroughParticlesTests(unittest.TestCase): + def test_codex_chunk_verifies_and_unmarks_under_particles(self) -> None: + marker = COMPOSER_PROBE_MARKER("ping") + typed = "Chat from Claude: ping" + marked_seen = f"Chat from Cl\u2801ude: ping{marker}" + unmarked_seen = "Chat from Claude: p\u2802ng" + replacements = { + "type_text": Mock(), + "wait_for_composer_change": Mock( + side_effect=[(marked_seen, 40), (unmarked_seen, 23)] + ), + } + + with patch.dict(TYPE_BODY.__globals__, replacements): + self.assertEqual( + TYPE_BODY("session-id", PROFILES["codex"], typed, ("", 2)), + (unmarked_seen, 23), + ) + + def test_claude_chunk_keeps_strict_verification(self) -> None: + marker = COMPOSER_PROBE_MARKER("ping") + typed = "Chat from Codex: ping" + replacements = { + "type_text": Mock(), + "clear_composer": Mock(return_value=True), + "wait_for_composer_change": Mock( + return_value=(f"Chat from C\u2801dex: ping{marker}", 40) + ), + } + + with patch.dict(TYPE_BODY.__globals__, replacements), self.assertRaisesRegex( + COMPOSER_DIRTY, "incomplete" + ): + TYPE_BODY("session-id", CLAUDE_PROFILE, typed, ("", 2)) + + +class IdleAnimationStabilityTests(unittest.TestCase): + OWNED = "Chat from Claude: ping" + FRAMES = ( + ("Chat from Cl\u2801ude: ping", 23), + ("Chat from Claude: p\u2802ng", 23), + ("Chat fr\u2804m Claude: ping", 23), + ("Chat from Claude: ping\u2808", 23), + ) + + def test_matching_change_settles_through_moving_particles(self) -> None: + composer_state = Mock(side_effect=self.FRAMES) + replacements = {"composer_state": composer_state, "time": stepping_time()} + + with patch.dict(WAIT_FOR_COMPOSER_CHANGE.__globals__, replacements): + settled = WAIT_FOR_COMPOSER_CHANGE( + "session-id", + PROFILES["codex"], + ("", 2), + CHUNK_SETTLE_DELAY, + matches=lambda content: COMPOSER_HAS_EXPECTED_TAIL( + content, self.OWNED, self.OWNED, tolerant=True + ), + ) + + self.assertEqual(settled, self.FRAMES[1]) + + def test_matching_change_does_not_settle_on_a_wrong_body(self) -> None: + composer_state = Mock(return_value=("Chat from Claude: pinf", 23)) + clock = Mock(monotonic=Mock(side_effect=[0.0, 0.1, 3.1]), sleep=Mock()) + replacements = {"composer_state": composer_state, "time": clock} + + with patch.dict(WAIT_FOR_COMPOSER_CHANGE.__globals__, replacements): + self.assertIsNone( + WAIT_FOR_COMPOSER_CHANGE( + "session-id", + PROFILES["codex"], + ("", 2), + CHUNK_SETTLE_DELAY, + matches=lambda content: COMPOSER_HAS_EXPECTED_TAIL( + content, self.OWNED, self.OWNED, tolerant=True + ), + ) + ) + + def test_unchanged_text_under_moving_particles_is_no_change(self) -> None: + composer_state = Mock(side_effect=self.FRAMES) + clock = Mock( + monotonic=Mock(side_effect=[0.0, 0.1, 0.2, 0.3, 3.1]), sleep=Mock() + ) + replacements = {"composer_state": composer_state, "time": clock} + + with patch.dict(WAIT_FOR_COMPOSER_CHANGE.__globals__, replacements): + self.assertIsNone( + WAIT_FOR_COMPOSER_CHANGE( + "session-id", PROFILES["codex"], self.FRAMES[0], CHUNK_SETTLE_DELAY + ) + ) + + def test_acceptance_settles_through_particles_over_the_placeholder(self) -> None: + composer_state = Mock( + side_effect=[ + ("Ask Codex to do\u2801anything", 2), + ("Ask\u2802Codex to do anything", 2), + ("Ask Codex to do anything \u2804", 2), + ] + ) + replacements = {"composer_state": composer_state, "time": stepping_time()} + + with patch.dict(WAIT_FOR_ACCEPTED.__globals__, replacements): + self.assertTrue( + WAIT_FOR_ACCEPTED("session-id", PROFILES["codex"], (self.OWNED, 23)) + ) + + self.assertEqual(composer_state.call_count, 2) + + def test_cleanup_state_settles_on_owned_spans_through_particles(self) -> None: + composer_state = Mock( + side_effect=[ + ("Chat from Claude: p\u2802ng", 23), + ("Chat from Cl\u2801ude: p", 20), + ("Chat fr\u2804m Claude: p", 20), + ("Chat from Claude: p", 20), + ] + ) + replacements = {"composer_state": composer_state, "time": stepping_time()} + + with patch.dict(WAIT_FOR_CLEANUP_STATE.__globals__, replacements): + settled = WAIT_FOR_CLEANUP_STATE( + "session-id", + PROFILES["codex"], + self.FRAMES[0], + self.OWNED, + {19, 20, 21}, + ) + + self.assertEqual(settled, (("Chat fr\u2804m Claude: p", 20), {(0, 19)})) + self.assertEqual(composer_state.call_count, 3) + + class ComposerCleanupTests(unittest.TestCase): def test_clear_does_not_claim_an_unobserved_body_was_removed(self) -> None: type_text = Mock() @@ -1785,6 +2149,7 @@ def test_each_chunk_waits_for_an_observed_composer_transition(self) -> None: ("", 2), CHUNK_SETTLE_DELAY, None, + matches=ANY, ), call.type_text( "session-id", @@ -1798,6 +2163,7 @@ def test_each_chunk_waits_for_an_observed_composer_transition(self) -> None: first_marked, CHUNK_SETTLE_DELAY, None, + matches=ANY, ), call.type_text( "session-id", CLAUDE_PROFILE, chunks[1] + marker, None @@ -1808,6 +2174,7 @@ def test_each_chunk_waits_for_an_observed_composer_transition(self) -> None: first_unmarked, CHUNK_SETTLE_DELAY, None, + matches=ANY, ), call.type_text( "session-id", @@ -1821,6 +2188,7 @@ def test_each_chunk_waits_for_an_observed_composer_transition(self) -> None: final_marked, COMPOSER_SETTLE_DELAY, None, + matches=ANY, ), ], )