fix: complete readline prompt coverage

This commit is contained in:
Haoran
2026-08-25 21:56:24 +08:00
parent b99447946d
commit 922265145b
29 changed files with 387 additions and 191 deletions

View File

@@ -105,7 +105,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms01 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s01 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -23,6 +23,12 @@ import os
import subprocess
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -136,7 +142,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms02 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s02 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -31,6 +31,12 @@ import os
import subprocess
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -196,7 +202,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms03 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s03 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -27,6 +27,12 @@ import os
import subprocess
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -172,7 +178,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms04 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s04 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -41,6 +41,12 @@ import subprocess
import yaml
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -212,7 +218,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms05 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s05 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -40,6 +40,12 @@ import subprocess
import time
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -247,7 +253,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms06 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s06 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -27,6 +27,12 @@ import os
import subprocess
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -228,7 +234,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms07 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s07 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -31,6 +31,12 @@ import threading
import uuid
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -246,7 +252,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms08 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s08 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -50,6 +50,12 @@ import threading
import time
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -383,7 +389,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms09 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s09 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -55,6 +55,12 @@ import time
import uuid
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -464,7 +470,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms10 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s10 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -43,6 +43,12 @@ import time
import uuid
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -556,7 +562,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms11 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s11 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -37,6 +37,12 @@ import subprocess
import time
from pathlib import Path
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -767,7 +773,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms12 >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s12 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -46,6 +46,12 @@ import uuid
from pathlib import Path
from queue import Queue
try:
import readline
readline.parse_and_bind('set bind-tty-special-chars off')
except ImportError:
pass
from anthropic import Anthropic
from dotenv import load_dotenv
@@ -743,7 +749,8 @@ if __name__ == "__main__":
history = []
while True:
try:
query = input("\033[36ms_full >> \033[0m")
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s_full >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -125,6 +125,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s01 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -182,6 +182,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s02 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -232,6 +232,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s03 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -245,6 +245,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s04 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -337,6 +337,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s05 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -353,6 +353,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s06 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -368,6 +368,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s07 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -570,6 +570,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s08 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -753,6 +753,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s09 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -570,6 +570,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s10 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -488,6 +488,7 @@ if __name__ == "__main__":
history = []
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s11 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -761,6 +761,7 @@ if __name__ == "__main__":
try:
while True:
try:
# \001/\002 tell Readline the ANSI escapes have zero display width.
query = input("\001\033[36m\002s12 >> \001\033[0m\002")
except (EOFError, KeyboardInterrupt):
break

View File

@@ -72,6 +72,8 @@ KEEP_RECENT_TOOL_RESULTS = 3
PERSIST_THRESHOLD = 30000
CONTINUATION_PROMPT = "Continue from the previous response. Do not repeat completed work."
PROMPT = "\033[36ms15 >> \033[0m"
# \001/\002 tell Readline the ANSI escapes have zero display width.
READLINE_PROMPT = "\001\033[36m\002s15 >> \001\033[0m\002"
CLI_ACTIVE = False
@@ -102,10 +104,17 @@ class ConsoleBroker:
def __init__(self):
self._lock = threading.Lock()
self.reader = None
self.display_prompt = PROMPT
self.readline_prompt = READLINE_PROMPT
def ask(self, prompt: str) -> str:
def set_prompt(self, display_prompt: str, readline_prompt: str):
self.display_prompt = display_prompt
self.readline_prompt = readline_prompt
def ask(self, prompt: str | None = None) -> str:
with self._lock:
return (self.reader or input)(prompt)
active_prompt = self.readline_prompt if prompt is None else prompt
return (self.reader or input)(active_prompt)
CONSOLE = ConsoleBroker()
@@ -122,7 +131,7 @@ def terminal_print(text: str):
except Exception:
line = ""
print(f"\r\033[K{text}")
print(PROMPT + line, end="", flush=True)
print(CONSOLE.display_prompt + line, end="", flush=True)
# -- Task System --
@@ -3266,7 +3275,7 @@ if __name__ == "__main__":
args=(history, context, session_state), daemon=True).start()
while True:
try:
query = CONSOLE.ask(PROMPT)
query = CONSOLE.ask()
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -833,10 +833,16 @@ async def run_demo(argv):
f"tokens={usage['tokens']} journal=.runtime/{task['runId']}.journal.jsonl")
PROMPT = "\033[36ms16 >> \033[0m"
# \001/\002 tell Readline the ANSI escapes have zero display width.
READLINE_PROMPT = "\001\033[36m\002s16 >> \001\033[0m\002"
def run_cli():
"""Run the cumulative s15 host with Workflow added to its tool pool."""
host = load_integrated_host()
install_workflow_tool(host)
host.CONSOLE.set_prompt(PROMPT, READLINE_PROMPT)
host.CLI_ACTIVE = True
host.start_runtime_services()
print("s16: workflow runtime")
@@ -851,7 +857,7 @@ def run_cli():
).start()
while True:
try:
query = host.CONSOLE.ask("\033[36ms16 >> \033[0m")
query = host.CONSOLE.ask()
except (EOFError, KeyboardInterrupt):
break
if query.strip().lower() in ("q", "exit", ""):

View File

@@ -0,0 +1,84 @@
import ast
import re
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
SOURCE_FILES = tuple(sorted([
*ROOT.glob("s*/code.py"),
*ROOT.glob("agents/*.py"),
]))
ANSI_ESCAPE = re.compile(r"\x1b\[[0-9;]*m")
def string_assignments(tree: ast.AST) -> dict[str, str]:
values = {}
for node in ast.walk(tree):
if not isinstance(node, (ast.Assign, ast.AnnAssign)):
continue
targets = node.targets if isinstance(node, ast.Assign) else [node.target]
value = node.value
if not isinstance(value, ast.Constant) or not isinstance(value.value, str):
continue
for target in targets:
if isinstance(target, ast.Name):
values[target.id] = value.value
return values
def input_prompts(path: Path) -> list[tuple[int, str]]:
tree = ast.parse(path.read_text(encoding="utf-8"))
assignments = string_assignments(tree)
prompts = []
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
is_input = isinstance(node.func, ast.Name) and node.func.id == "input"
is_console_ask = isinstance(node.func, ast.Attribute) and node.func.attr == "ask"
if not (is_input or is_console_ask):
continue
if not node.args:
if is_console_ask and "READLINE_PROMPT" in assignments:
prompts.append((node.lineno, assignments["READLINE_PROMPT"]))
continue
argument = node.args[0]
if isinstance(argument, ast.Constant) and isinstance(argument.value, str):
prompts.append((node.lineno, argument.value))
elif isinstance(argument, ast.Name) and argument.id in assignments:
prompts.append((node.lineno, assignments[argument.id]))
return prompts
def test_colored_input_prompts_mark_ansi_as_zero_width() -> None:
checked = 0
invalid = []
for path in SOURCE_FILES:
for lineno, prompt in input_prompts(path):
escapes = list(ANSI_ESCAPE.finditer(prompt))
if not escapes:
assert "\x01" not in prompt and "\x02" not in prompt
continue
checked += 1
for escape in escapes:
marked = (
prompt[escape.start() - 1:escape.start()] == "\x01"
and prompt[escape.end():escape.end() + 1] == "\x02"
)
if not marked:
invalid.append(f"{path.relative_to(ROOT)}:{lineno}")
break
assert checked, "expected at least one colored input prompt"
assert not invalid, "ANSI escapes missing Readline markers:\n" + "\n".join(invalid)
def test_async_redraw_prompts_keep_markers_out_of_display_text() -> None:
for lesson in ("s15_integrated_harness", "s16_workflow_runtime"):
path = ROOT / lesson / "code.py"
tree = ast.parse(path.read_text(encoding="utf-8"))
assignments = string_assignments(tree)
display_prompt = assignments["PROMPT"]
readline_prompt = assignments["READLINE_PROMPT"]
assert "\x01" not in display_prompt and "\x02" not in display_prompt
assert readline_prompt.replace("\x01", "").replace("\x02", "") == display_prompt

File diff suppressed because one or more lines are too long