v0.7.5: arrow keys, image request fix, deepseek-flash vision, musl CI
TUI keys: register both CSI and SS3 forms of the cursor/Home/End keys and
the xterm modifier sequences when terminfo lacks them, so an arrow sent in
the "other" form no longer turns into Esc + a typed letter. Up/Down move by
visual (soft-wrapped) row, and the draft survives a trip through history.
scripts/tui_keys drives the TUI in a pty to check all of this.
Images: text inside a multimodal content array is now a {"type":"text"}
part; DeepSeek rejected the bare string with HTTP 422. deepseek-flash
(and its deepseek-chat / deepseek-reasoner aliases) read images, so the
registry marks them vision; deepseek-v4-pro stays text only.
CI: models.c lost _DEFAULT_SOURCE in 0.7.0 and gmtime_r broke the musl
build; tests sleep with poll() instead of usleep for the same reason.
This commit is contained in:
WHC
@@ -0,0 +1,44 @@
|
||||
import os
|
||||
import sys, time
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from harness import Term
|
||||
import fakeapi
|
||||
from keys import input_text, styled, check, FAILS, base
|
||||
|
||||
for term in ("xterm-256color", "screen-256color"):
|
||||
t = Term(args=("-b", base, "-k", "sk-test"), term=term)
|
||||
t.send("some text")
|
||||
t.send("\x1b", 0.4)
|
||||
check(term + " lone esc clears input", input_text(t) == [], t)
|
||||
t.send("alpha beta")
|
||||
t.send("\x1bb") # alt+b word left
|
||||
t.send("X")
|
||||
check(term + " alt+b word left", "alpha Xbeta" in input_text(t), t)
|
||||
t.send("\x1b\r") # alt+enter newline
|
||||
t.send("Y")
|
||||
check(term + " alt+enter newline", input_text(t)[-2:] == ["alpha X", "Ybeta"], t)
|
||||
t.send("\x03", 0.2)
|
||||
# picker with CSI form
|
||||
t.send("/model\r", 0.8)
|
||||
s0 = styled(t)
|
||||
t.arrow("down", "csi")
|
||||
s1 = styled(t)
|
||||
t.arrow("up", "csi")
|
||||
s2 = styled(t)
|
||||
check(term + " picker csi down", s0 != s1, t)
|
||||
check(term + " picker csi up", s0 == s2, t)
|
||||
t.send("\x1b", 0.4)
|
||||
# home/end variants
|
||||
t.send("middle")
|
||||
for seq in ("\x1b[H", "\x1bOH", "\x1b[1~", "\x1b[7~"):
|
||||
t.send("\x1b[F" + seq)
|
||||
t.send("<")
|
||||
t.send("\x08" if False else "")
|
||||
check(term + " home variants", input_text(t)[-1:] == ["<<<<middle"], t)
|
||||
for seq in ("\x1b[F", "\x1bOF", "\x1b[4~", "\x1b[8~"):
|
||||
t.send("\x1b[H" + seq)
|
||||
t.send(">")
|
||||
check(term + " end variants", input_text(t)[-1:] == ["<<<<middle>>>>"], t)
|
||||
t.close()
|
||||
print("FAILED:", len(FAILS))
|
||||
sys.exit(min(len(FAILS), 100))
|
||||
@@ -0,0 +1,43 @@
|
||||
import json, sys, threading, time
|
||||
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
|
||||
|
||||
LOG = []
|
||||
|
||||
|
||||
class H(BaseHTTPRequestHandler):
|
||||
def log_message(self, *a):
|
||||
pass
|
||||
|
||||
def do_GET(self):
|
||||
body = json.dumps({"data": [{"id": "deepseek-flash"}, {"id": "deepseek-v4-pro"}]}).encode()
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "application/json")
|
||||
self.send_header("Content-Length", str(len(body)))
|
||||
self.end_headers()
|
||||
self.wfile.write(body)
|
||||
|
||||
def do_POST(self):
|
||||
n = int(self.headers.get("Content-Length", 0))
|
||||
raw = self.rfile.read(n)
|
||||
try:
|
||||
LOG.append(json.loads(raw))
|
||||
except Exception:
|
||||
LOG.append({"_raw": raw[:200].decode("latin1")})
|
||||
self.send_response(200)
|
||||
self.send_header("Content-Type", "text/event-stream")
|
||||
self.end_headers()
|
||||
def ev(o):
|
||||
self.wfile.write(b"data: " + json.dumps(o).encode() + b"\n\n")
|
||||
self.wfile.flush()
|
||||
base = {"id": "x", "object": "chat.completion.chunk", "model": "deepseek-flash"}
|
||||
ev({**base, "choices": [{"index": 0, "delta": {"role": "assistant", "content": "reply-%d" % len(LOG)}}]})
|
||||
ev({**base, "choices": [{"index": 0, "delta": {}, "finish_reason": "stop"}],
|
||||
"usage": {"prompt_tokens": 10, "completion_tokens": 2, "total_tokens": 12}})
|
||||
self.wfile.write(b"data: [DONE]\n\n")
|
||||
self.wfile.flush()
|
||||
|
||||
|
||||
def start():
|
||||
s = ThreadingHTTPServer(("127.0.0.1", 0), H)
|
||||
threading.Thread(target=s.serve_forever, daemon=True).start()
|
||||
return s, "http://127.0.0.1:%d/v1" % s.server_address[1]
|
||||
@@ -0,0 +1,99 @@
|
||||
"""Drive the cst-agent TUI in a pty and read the screen back (pyte).
|
||||
|
||||
Run from the repo root after `make`:
|
||||
nix-shell -p python3Packages.pyte --run "python3 scripts/tui_keys/keys.py"
|
||||
nix-shell -p python3Packages.pyte --run "python3 scripts/tui_keys/esc.py"
|
||||
(or `pip install pyte`). A local fake OpenAI-compatible server answers the
|
||||
model calls, so no API key is needed. Exit status is the failure count.
|
||||
"""
|
||||
import os, pty, sys, time, select, fcntl, termios, struct, signal, shutil, tempfile
|
||||
import pyte
|
||||
|
||||
BIN = os.environ.get("CST_BIN") or os.path.join(
|
||||
os.path.dirname(os.path.abspath(__file__)), "..", "..", "cst-agent")
|
||||
|
||||
|
||||
class Term:
|
||||
def __init__(self, cols=120, rows=36, kitty=False, env=None, args=(), home=None,
|
||||
ws=None, term="xterm-256color"):
|
||||
self.cols, self.rows, self.kitty = cols, rows, kitty
|
||||
self.screen = pyte.Screen(cols, rows)
|
||||
self.stream = pyte.ByteStream(self.screen)
|
||||
self.home = home or tempfile.mkdtemp(prefix="cstk_home_")
|
||||
self.ws = ws or tempfile.mkdtemp(prefix="cstk_ws_")
|
||||
e = dict(os.environ)
|
||||
e.update({"TERM": term, "HOME": self.home,
|
||||
"XDG_CONFIG_HOME": os.path.join(self.home, ".config"),
|
||||
"LANG": "C.UTF-8", "LC_ALL": "C.UTF-8",
|
||||
"CST_PROVIDER": "deepseek", "DEEPSEEK_API_KEY": "sk-dummy-test",
|
||||
"CST_METRICS": "0"})
|
||||
e.pop("NO_COLOR", None)
|
||||
if env:
|
||||
e.update(env)
|
||||
self.raw = b""
|
||||
pid, fd = pty.fork()
|
||||
if pid == 0:
|
||||
os.chdir(self.ws)
|
||||
os.execve(BIN, [BIN, "-N", *args], e)
|
||||
self.pid, self.fd = pid, fd
|
||||
fcntl.ioctl(fd, termios.TIOCSWINSZ, struct.pack("HHHH", rows, cols, 0, 0))
|
||||
self.pump(1.5)
|
||||
|
||||
def pump(self, t=0.25):
|
||||
end = time.time() + t
|
||||
while True:
|
||||
left = end - time.time()
|
||||
if left <= 0:
|
||||
break
|
||||
r, _, _ = select.select([self.fd], [], [], left)
|
||||
if not r:
|
||||
continue
|
||||
try:
|
||||
d = os.read(self.fd, 65536)
|
||||
except OSError:
|
||||
break
|
||||
if not d:
|
||||
break
|
||||
self.raw += d
|
||||
if self.kitty and b"\x1b[?u" in d:
|
||||
os.write(self.fd, b"\x1b[?0u")
|
||||
self.stream.feed(d)
|
||||
|
||||
def send(self, b, t=0.15):
|
||||
if isinstance(b, str):
|
||||
b = b.encode()
|
||||
os.write(self.fd, b)
|
||||
self.pump(t)
|
||||
|
||||
def appcursor(self):
|
||||
return (1 << 5) in self.screen.mode # pyte DECCKM
|
||||
|
||||
def arrow(self, d, form=None, mod=None, t=0.15):
|
||||
c = {"up": "A", "down": "B", "right": "C", "left": "D"}[d]
|
||||
if mod:
|
||||
seq = "\x1b[1;%d%s" % (mod, c)
|
||||
else:
|
||||
f = form or ("ss3" if self.appcursor() else "csi")
|
||||
seq = ("\x1bO" if f == "ss3" else "\x1b[") + c
|
||||
self.send(seq, t)
|
||||
|
||||
def lines(self):
|
||||
return [l.rstrip() for l in self.screen.display]
|
||||
|
||||
def find(self, s):
|
||||
return [(i, l) for i, l in enumerate(self.lines()) if s in l]
|
||||
|
||||
def cursor(self):
|
||||
return self.screen.cursor.y, self.screen.cursor.x
|
||||
|
||||
def dump(self, title=""):
|
||||
print("=" * 20, title, "cursor", self.cursor())
|
||||
for i, l in enumerate(self.lines()):
|
||||
print("%2d|%s" % (i, l))
|
||||
|
||||
def close(self):
|
||||
try:
|
||||
os.kill(self.pid, signal.SIGTERM)
|
||||
os.waitpid(self.pid, 0)
|
||||
except Exception:
|
||||
pass
|
||||
@@ -0,0 +1,273 @@
|
||||
import os
|
||||
import sys, time
|
||||
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
||||
from harness import Term
|
||||
import fakeapi
|
||||
|
||||
srv, base = fakeapi.start()
|
||||
FAILS = []
|
||||
VERBOSE = "-v" in sys.argv
|
||||
|
||||
|
||||
def check(name, cond, t=None):
|
||||
print(("PASS " if cond else "FAIL ") + name)
|
||||
if not cond:
|
||||
FAILS.append(name)
|
||||
if t:
|
||||
t.dump(name)
|
||||
|
||||
|
||||
def input_text(t):
|
||||
"""Input box rows: the bar rows right above the mode line."""
|
||||
L = t.lines()
|
||||
mode = [i for i, l in enumerate(L) if l.lstrip().startswith("┃ ") and
|
||||
("Build ·" in l or "Plan ·" in l)]
|
||||
if not mode:
|
||||
return []
|
||||
rows = []
|
||||
i = mode[-1] - 1
|
||||
while i >= 0:
|
||||
s = L[i].lstrip()
|
||||
if not s.startswith("┃"):
|
||||
break
|
||||
body = s[2:].split(" ")[0].rstrip() if s.startswith("┃ ") else ""
|
||||
if body == "" and rows:
|
||||
break
|
||||
if body:
|
||||
rows.append(body)
|
||||
i -= 1
|
||||
rows.reverse()
|
||||
if rows and (rows[0].startswith("Ask anything") or rows[0].startswith("Enter queues")):
|
||||
return []
|
||||
return rows
|
||||
|
||||
|
||||
def styled(t):
|
||||
"""Screen text plus the reverse/bg attributes, to see highlight moves."""
|
||||
out = []
|
||||
for y in range(t.rows):
|
||||
row = t.screen.buffer[y]
|
||||
out.append("".join("%s%s%s" % (row[x].data, int(row[x].reverse), row[x].bg[:3])
|
||||
for x in range(t.cols)))
|
||||
return out
|
||||
|
||||
|
||||
def clear(t):
|
||||
if [r for r in input_text(t) if r]:
|
||||
t.send("\x03", 0.2) # ctrl+c clears a non-empty input (twice on empty quits)
|
||||
|
||||
|
||||
def new(kitty=False, **kw):
|
||||
t = Term(kitty=kitty, args=("-b", base, "-k", "sk-test", "-m", "deepseek-flash"), **kw)
|
||||
return t
|
||||
|
||||
|
||||
def basic(t, form, tag):
|
||||
clear(t)
|
||||
t.send("hello world")
|
||||
for _ in range(5):
|
||||
t.arrow("left", form)
|
||||
t.send("X")
|
||||
check(f"{tag} left x5 insert", "hello Xworld" in input_text(t), t)
|
||||
t.arrow("right", form); t.arrow("right", form)
|
||||
t.send("Y")
|
||||
check(f"{tag} right x2 insert", "hello XwoYrld" in input_text(t), t)
|
||||
# boundaries
|
||||
for _ in range(30):
|
||||
t.arrow("left", form, t=0.02)
|
||||
t.pump(0.2)
|
||||
t.send("<")
|
||||
check(f"{tag} left past start", "<hello XwoYrld" in input_text(t), t)
|
||||
for _ in range(30):
|
||||
t.arrow("right", form, t=0.02)
|
||||
t.pump(0.2)
|
||||
t.send(">")
|
||||
check(f"{tag} right past end", "<hello XwoYrld>" in input_text(t), t)
|
||||
clear(t)
|
||||
|
||||
|
||||
def cjk(t):
|
||||
clear(t)
|
||||
t.send("你好世界")
|
||||
y, x_end = t.cursor()
|
||||
t.arrow("left"); t.arrow("left")
|
||||
y2, x2 = t.cursor()
|
||||
check("cjk cursor moves 2 cells per char", x_end - x2 == 4, t)
|
||||
t.send("A")
|
||||
check("cjk insert between wide chars", "你好A世界" in input_text(t), t)
|
||||
t.arrow("right")
|
||||
t.send("B")
|
||||
check("cjk right over wide char", "你好A世B界" in input_text(t), t)
|
||||
clear(t)
|
||||
t.send("a😀b")
|
||||
t.arrow("left"); t.arrow("left")
|
||||
t.send("Z")
|
||||
check("emoji left", "aZ😀b" in input_text(t), t)
|
||||
clear(t)
|
||||
|
||||
|
||||
def multiline(t):
|
||||
clear(t)
|
||||
t.send("line one is long")
|
||||
t.send("\x0e") # ctrl+n newline
|
||||
t.send("l2")
|
||||
t.send("\x0e")
|
||||
t.send("third line here")
|
||||
rows = input_text(t)
|
||||
check("multiline 3 rows", rows[-3:] == ["line one is long", "l2", "third line here"], t)
|
||||
for _ in range(4):
|
||||
t.arrow("left") # cursor before "here"
|
||||
t.arrow("up") # col 11 -> clamp to end of "l2"
|
||||
t.send("U")
|
||||
rows = input_text(t)
|
||||
check("up clamps column to short line", rows[-2] == "l2U", t)
|
||||
t.arrow("up") # col 3 on line one
|
||||
t.send("V")
|
||||
rows = input_text(t)
|
||||
check("up keeps column", rows[-3] == "linVe one is long", t)
|
||||
t.arrow("down"); t.arrow("down")
|
||||
t.send("W")
|
||||
rows = input_text(t)
|
||||
check("down twice to last line", "W" in rows[-1], t)
|
||||
clear(t)
|
||||
|
||||
|
||||
def history(t):
|
||||
clear(t)
|
||||
t.send("first prompt\r", 1.5)
|
||||
t.send("second prompt\r", 1.5)
|
||||
t.arrow("up")
|
||||
check("hist up -> newest", input_text(t)[-1:] == ["second prompt"], t)
|
||||
t.arrow("up")
|
||||
check("hist up -> older", input_text(t)[-1:] == ["first prompt"], t)
|
||||
t.arrow("up")
|
||||
check("hist up at oldest stays", input_text(t)[-1:] == ["first prompt"], t)
|
||||
t.arrow("down")
|
||||
check("hist down -> newer", input_text(t)[-1:] == ["second prompt"], t)
|
||||
t.arrow("down")
|
||||
check("hist down past newest -> empty", input_text(t) in ([], [""]), t)
|
||||
# draft survives a trip through history
|
||||
t.send("my draft")
|
||||
t.arrow("up")
|
||||
t.arrow("down")
|
||||
check("draft restored after up/down", input_text(t)[-1:] == ["my draft"], t)
|
||||
clear(t)
|
||||
|
||||
|
||||
def wrapped(t):
|
||||
clear(t)
|
||||
long = " ".join("w%02d" % i for i in range(40)) # ~160 chars, wraps
|
||||
t.send(long, 0.4)
|
||||
nrows = len(input_text(t))
|
||||
t.arrow("up")
|
||||
txt = "".join(input_text(t))
|
||||
check("up on wrapped line keeps the text", "w39" in txt and "w00" in txt, t)
|
||||
t.send("Q")
|
||||
rows = input_text(t)
|
||||
check("up on wrapped line moves one visual row", nrows >= 2 and "Q" in "".join(rows[:-1]), t)
|
||||
clear(t)
|
||||
|
||||
|
||||
def completion(t):
|
||||
clear(t)
|
||||
t.send("/", 0.3)
|
||||
before = styled(t)
|
||||
t.arrow("down"); t.arrow("down")
|
||||
mid = styled(t)
|
||||
t.arrow("up")
|
||||
after = styled(t)
|
||||
check("completion popup reacts to down", before != mid, t)
|
||||
check("completion up differs from down-down", mid != after, t)
|
||||
check("completion arrows do not touch input", input_text(t)[-1:] == ["/"], t)
|
||||
t.send("\x1b", 0.3)
|
||||
clear(t)
|
||||
|
||||
|
||||
def modifiers(t):
|
||||
clear(t)
|
||||
t.send("alpha beta gamma")
|
||||
t.arrow("left", mod=5) # ctrl+left -> word left
|
||||
t.send("X")
|
||||
check("ctrl+left word", "alpha beta Xgamma" in input_text(t), t)
|
||||
t.arrow("left", mod=3) # alt+left
|
||||
t.arrow("left", mod=3)
|
||||
t.send("Y")
|
||||
check("alt+left word", "alpha Ybeta Xgamma" in input_text(t), t)
|
||||
t.arrow("right", mod=5)
|
||||
t.send("Z")
|
||||
check("ctrl+right word", "alpha YbetaZ Xgamma" in input_text(t), t)
|
||||
t.arrow("left", mod=2) # shift+left: should at least not insert junk
|
||||
t.send("S")
|
||||
txt = "".join(input_text(t))
|
||||
check("shift+left no junk", "[" not in txt and ";" not in txt and "1;2" not in txt, t)
|
||||
clear(t)
|
||||
|
||||
|
||||
def split_and_burst(t, form):
|
||||
clear(t)
|
||||
t.send("abcdef")
|
||||
ss = "\x1bO" if form == "ss3" else "\x1b["
|
||||
# escape sequence split across writes (slow link)
|
||||
import os
|
||||
os.write(t.fd, ss[:1].encode()); time.sleep(0.008)
|
||||
os.write(t.fd, (ss[1:] + "D").encode()); t.pump(0.2)
|
||||
t.send("1")
|
||||
check(f"split {form} seq", "abcde1f" in input_text(t), t)
|
||||
# key-repeat burst in one write
|
||||
t.send((ss + "D") * 4, 0.3)
|
||||
t.send("2")
|
||||
check(f"burst {form} x4", "ab2cde1f" in input_text(t), t)
|
||||
clear(t)
|
||||
|
||||
|
||||
def picker(t):
|
||||
clear(t)
|
||||
t.send("/model\r", 0.8)
|
||||
s0 = styled(t)
|
||||
t.arrow("down")
|
||||
s1 = styled(t)
|
||||
t.arrow("up")
|
||||
s2 = styled(t)
|
||||
check("picker down changes selection", s0 != s1, t)
|
||||
check("picker up returns", s0 == s2, t)
|
||||
t.send("\x1b", 0.4)
|
||||
clear(t)
|
||||
|
||||
|
||||
def run(kitty, tag):
|
||||
print("#####", tag)
|
||||
t = new(kitty=kitty)
|
||||
try:
|
||||
basic(t, None, tag + " native")
|
||||
basic(t, "csi" if t.appcursor() else "ss3", tag + " other-form")
|
||||
cjk(t)
|
||||
multiline(t)
|
||||
modifiers(t)
|
||||
split_and_burst(t, "ss3")
|
||||
split_and_burst(t, "csi")
|
||||
completion(t)
|
||||
wrapped(t)
|
||||
history(t)
|
||||
picker(t)
|
||||
finally:
|
||||
t.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
run(False, "xterm")
|
||||
run(True, "kitty")
|
||||
for term in ("screen-256color", "tmux-256color", "linux", "vt220", "xterm"):
|
||||
print("#####", term)
|
||||
t = Term(args=("-b", base, "-k", "sk-test"), term=term)
|
||||
try:
|
||||
basic(t, "ss3", term + " ss3")
|
||||
basic(t, "csi", term + " csi")
|
||||
modifiers(t)
|
||||
multiline(t)
|
||||
cjk(t)
|
||||
finally:
|
||||
t.close()
|
||||
print("FAILED:", len(FAILS))
|
||||
for f in FAILS:
|
||||
print(" -", f)
|
||||
sys.exit(min(len(FAILS), 100))
|
||||
Reference in New Issue
Block a user