{{ message }}
Repository navigation
Expand file tree
/
Copy pathdemo.py
More file actions
882 lines (762 loc) · 36.4 KB
/
Copy pathdemo.py
File metadata and controls
882 lines (762 loc) · 36.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
"""Record one `coder` run to a terminal cast and render that cast to a GIF.
Why not asciinema, vhs or terminalizer: none of them runs natively on Windows, and the GIF
step needs ffmpeg or a headless browser on top. This file does both halves in Python with what
the project already installs (Rich decodes the colour codes, Pillow draws the frames), so the
README hero can be regenerated on the laptop that develops the agent.
The cast is the asciinema v2 format, one JSON header line then `[time, "o" | "i", text]` events,
so the recording also plays in asciinema or agg unchanged. Input events are written by the
recorder when it answers a prompt, because a pipe does not echo what was typed.
Usage:
python scripts/demo.py prepare TASK_ID DEST # fresh copy of a suite task, prints its prompt
python scripts/demo.py record OUT.cast [--title TEXT] -- coder run DEST "prompt"
python scripts/demo.py render IN.cast OUT.gif [--rows 24] [--max-gap 1.5]
python scripts/demo.py render IN.cast OUT.svg # same frames, CSS-animated
python scripts/demo.py render IN.cast OUT.json --frames # same frames, for the web player
The chosen cast is committed as docs/figures/demo.cast, so restyling the hero is a render
away and never another paid run.
"""
from __future__ import annotations
import argparse
import codecs
import importlib.util
import json
import os
import queue
import re
import shutil
import subprocess
import sys
import threading
import time
from dataclasses import dataclass, field
from pathlib import Path
from typing import TYPE_CHECKING
from rich.ansi import AnsiDecoder
from rich.style import Style
from rich.terminal_theme import TerminalTheme
if TYPE_CHECKING: # Pillow is imported lazily, so the drawing helpers only name its types
from collections.abc import Iterator
from PIL.Image import Image as PilImage
from PIL.ImageFont import FreeTypeFont
# Everything the screen model treats as a command rather than as text: line breaks, backspace,
# bell, CSI sequences (colours and cursor movement) and OSC sequences (window titles, links).
CONTROL = re.compile(r"\r\n|\r|\n|\x08|\x07|\x1b\[[0-?]*[ -/]*[@-~]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)")
ANSI = re.compile(r"\x1b\[[0-?]*[ -/]*[@-~]|\x1b\][^\x07\x1b]*(?:\x07|\x1b\\)")
# The one prompt `coder run` asks. Extra `--answer PATTERN=TEXT` pairs extend this.
DEFAULT_ANSWERS = {r"Approve\?": "y"}
Cell = tuple[str, Style]
# --------------------------------------------------------------------------- recording
def _child_env(cols: int, rows: int) -> dict[str, str]:
"""Make the child believe it is in a colour terminal of a known size.
Rich reads COLUMNS/LINES when it cannot query the terminal, and the CLI turns FORCE_COLOR
into `force_terminal` (see `coder_agent.cli._console`). COLORTERM unlocks 24-bit colour so the
diff highlighting comes through unchanged; PYTHONIOENCODING keeps box-drawing characters
intact on a Windows pipe, which would otherwise be cp1252.
NO_COLOR has to go, not just be overridden. Rich gives it precedence over FORCE_COLOR, and it
keeps bold and dim while dropping every colour - so a shell that exports it (this one does)
yields a recording that looks plausible and is grey from end to end. The stage icons, the
Tests tick and the diff all come out the colour of the body text, and nothing in the cast
says why.
"""
env = {
**os.environ,
"FORCE_COLOR": "1",
"COLORTERM": "truecolor",
"TERM": "xterm-256color",
"COLUMNS": str(cols),
"LINES": str(rows),
"PYTHONIOENCODING": "utf-8",
"PYTHONUNBUFFERED": "1",
}
env.pop("NO_COLOR", None)
return env
def _resolve(command: list[str]) -> list[str]:
"""`coder` is a console script in the venv; a plain Popen on Windows would not find it."""
exe = shutil.which(command[0])
return [exe or command[0], *command[1:]]
def record(
command: list[str],
out: Path,
*,
answers: dict[str, str] | None = None,
cols: int = 100,
rows: int = 28,
answer_delay: float = 0.8,
timeout: float = 900.0,
title: str | None = None,
cwd: Path | None = None,
) -> int:
"""Run `command` with stdout piped, timestamp every chunk, answer prompts. Returns exit code.
A reader thread moves chunks into a queue so the main loop can also watch the clock: a child
that hangs on a rate limit is killed instead of blocking the recording forever.
"""
answers = {**DEFAULT_ANSWERS, **(answers or {})}
patterns = [(re.compile(p), text) for p, text in answers.items()]
started = time.monotonic()
proc = subprocess.Popen(
_resolve(command),
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=open(out.with_suffix(".stderr.log"), "wb"), # noqa: SIM115 - closed by the OS with the child
env=_child_env(cols, rows),
cwd=cwd,
)
assert proc.stdout is not None and proc.stdin is not None
chunks: queue.Queue[tuple[float, bytes] | None] = queue.Queue()
def reader() -> None:
while data := proc.stdout.read1(4096):
chunks.put((time.monotonic() - started, data))
chunks.put(None)
threading.Thread(target=reader, daemon=True).start()
decoder = codecs.getincrementaldecoder("utf-8")(errors="replace")
# The cast is committed, so the header goes into a public repository: record the title the
# caller chose to show rather than the real argv, whose destination is an absolute path under
# the home directory of whoever recorded the take.
shown = title or " ".join(command)
header = {
"version": 2, "width": cols, "height": rows, "timestamp": int(time.time()),
"command": shown, "title": shown,
"env": {"TERM": "xterm-256color", "SHELL": "powershell"},
}
tail = "" # output since the last answered prompt, colour codes removed, for matching
with out.open("w", encoding="utf-8") as fh:
fh.write(json.dumps(header) + "\n")
while True:
try:
item = chunks.get(timeout=0.2)
except queue.Empty:
if time.monotonic() - started > timeout:
proc.kill()
fh.write(json.dumps([time.monotonic() - started, "o",
"\r\n[demo recorder] timeout, child killed\r\n"]) + "\n")
break
continue
if item is None:
break
at, data = item
text = decoder.decode(data)
fh.write(json.dumps([round(at, 3), "o", text]) + "\n")
tail = (tail + ANSI.sub("", text))[-4000:]
for pattern, reply in patterns:
if pattern.search(tail):
time.sleep(answer_delay)
proc.stdin.write((reply + "\n").encode("utf-8"))
proc.stdin.flush()
fh.write(json.dumps([round(time.monotonic() - started, 3), "i", reply + "\n"])
+ "\n")
tail = ""
break
proc.stdin.close()
return proc.wait()
# --------------------------------------------------------------------------- screen model
class Screen:
"""A minimal terminal: a growing list of lines of styled cells, a cursor, and the SGR state.
Rich's `AnsiDecoder` keeps the current style between calls, so it is fed only the text runs
and the colour codes; line breaks, carriage returns and cursor movement are handled here,
which is what makes progress bars and re-drawn prompts come out right.
"""
def __init__(self, cols: int, rows: int) -> None:
self.cols = cols
self.rows = rows
self.lines: list[list[Cell]] = [[]]
self.row = 0
self.col = 0
self.decoder = AnsiDecoder()
# -- cursor helpers
@property
def top(self) -> int:
return max(0, len(self.lines) - self.rows)
def _line(self) -> list[Cell]:
while self.row >= len(self.lines):
self.lines.append([])
return self.lines[self.row]
def _newline(self) -> None:
self.row += 1
self.col = 0
self._line()
def _put(self, char: str, style: Style) -> None:
if self.col >= self.cols:
self._newline()
line = self._line()
while len(line) < self.col:
line.append((" ", Style.null()))
if self.col < len(line):
line[self.col] = (char, style)
else:
line.append((char, style))
self.col += 1
# -- input
def feed(self, data: str) -> None:
pos = 0
for m in CONTROL.finditer(data):
self._text(data[pos:m.start()])
self._control(m.group())
pos = m.end()
self._text(data[pos:])
def _text(self, run: str) -> None:
if not run:
return
text = self.decoder.decode_line(run)
styles = [Style.null()] * len(text.plain)
for span in text.spans:
style = span.style if isinstance(span.style, Style) else Style.parse(str(span.style))
for i in range(span.start, min(span.end, len(styles))):
styles[i] = style
for char, style in zip(text.plain, styles, strict=True):
if char == "\t":
for _ in range(8 - self.col % 8):
self._put(" ", style)
else:
self._put(char, style)
def _control(self, seq: str) -> None:
if seq in ("\n", "\r\n"):
self._newline()
elif seq == "\r":
self.col = 0
elif seq == "\x08":
self.col = max(0, self.col - 1)
elif seq == "\x07" or seq.startswith("\x1b]"):
return
elif seq.endswith("m"):
self.decoder.decode_line(seq) # updates decoder.style, returns empty text
else:
self._csi(seq)
def _csi(self, seq: str) -> None:
final = seq[-1]
params = seq[2:-1]
if params.startswith("?"): # cursor visibility, alternate screen: nothing to draw
return
n = int(params.split(";")[0]) if params.split(";")[0].isdigit() else None
if final == "K":
line = self._line()
if n in (None, 0):
del line[self.col:]
elif n == 1:
for i in range(min(self.col + 1, len(line))):
line[i] = (" ", Style.null())
else:
line.clear()
elif final == "A":
self.row = max(self.top, self.row - (n or 1))
elif final == "B":
self.row += n or 1
elif final == "C":
self.col = min(self.cols - 1, self.col + (n or 1))
elif final == "D":
self.col = max(0, self.col - (n or 1))
elif final == "G":
self.col = max(0, (n or 1) - 1)
elif final in "Hf":
parts = [int(p) if p.isdigit() else 1 for p in params.split(";")] + [1]
self.row = self.top + parts[0] - 1
self.col = parts[1] - 1
elif final == "J" and n == 2:
self.lines = [[]]
self.row = self.col = 0
# -- output
def snapshot(self) -> list[list[Cell]]:
"""The visible rows, padded to `rows` so every frame has the same shape."""
visible = [list(line) for line in self.lines[self.top:]]
return visible + [[] for _ in range(self.rows - len(visible))]
def plain(self) -> list[str]:
return ["".join(c for c, _ in line).rstrip() for line in self.lines]
# --------------------------------------------------------------------------- frames
@dataclass
class Frame:
cells: list[list[Cell]]
duration: float # seconds this frame stays on screen
@dataclass
class Cast:
header: dict
events: list[tuple[float, str, str]] = field(default_factory=list)
@classmethod
def load(cls, path: Path) -> Cast:
lines = path.read_text(encoding="utf-8").splitlines()
header = json.loads(lines[0])
events = [tuple(json.loads(line)) for line in lines[1:] if line.strip()]
return cls(header, events) # type: ignore[arg-type]
def frames_from_cast(
cast: Cast,
*,
rows: int | None = None,
max_gap: float = 1.5,
min_frame: float = 0.06,
hold: float = 3.0,
typing_delay: float = 0.06,
) -> list[Frame]:
"""Replay the cast through the screen model and emit one frame per visible change.
Time between events is capped at `max_gap`, because a model call that took twelve seconds
makes a bad GIF; identical screens extend the previous frame instead of adding one; a frame
that has not yet been shown for `min_frame` takes the newer screen instead of a new slot, so
a burst of tiny writes becomes a few visible frames rather than dozens the eye cannot
follow. The test is on the previous frame's duration, not the new one's: testing the new
one merged every 40 ms typing step into a single frame, and the whole typed command sat
still for eight seconds.
"""
cols = int(cast.header.get("width", 100))
screen = Screen(cols, rows or int(cast.header.get("height", 28)))
frames: list[Frame] = []
def emit(duration: float) -> None:
cells = screen.snapshot()
if frames and frames[-1].cells == cells:
frames[-1].duration += duration
elif frames and frames[-1].duration < min_frame:
frames[-1].cells = cells # too quick to see: show the newer state in the same slot
frames[-1].duration += duration
else:
frames.append(Frame(cells, duration))
# Prelude: the command typed at a prompt, since the pipe never showed it. A word at a time,
# not a character at a time: the task sentence is two hundred characters, and typing it out
# letter by letter spent the first five seconds of the GIF on an otherwise empty screen.
screen.feed("\x1b[1;32m$\x1b[0m ")
emit(0.6)
typed = cast.header.get("title", cast.header.get("command", ""))
for word in re.findall(r"\S+\s*", typed):
screen.feed(f"\x1b[1m{word}\x1b[0m")
emit(typing_delay)
emit(0.5)
screen.feed("\r\n")
previous = cast.events[0][0] if cast.events else 0.0
for at, kind, text in cast.events:
gap = min(max(at - previous, 0.0), max_gap)
previous = at
if kind == "o":
emit(gap)
screen.feed(text)
elif kind == "i":
emit(gap)
for char in text.rstrip("\n"):
screen.feed(f"\x1b[1m{char}\x1b[0m")
emit(typing_delay * 2)
emit(0.3)
screen.feed("\r\n")
emit(hold)
return frames
# --------------------------------------------------------------------------- drawing
# A charcoal-navy window rather than pure black, and every ANSI colour desaturated a step: the
# recording is read at a glance in a README, where a neon green on black reads as a screenshot of
# a hacker movie. Success still looks like success, it just does not shout.
BACKGROUND = (25, 26, 40)
FOREGROUND = (205, 211, 224)
TITLE_BAR = (36, 38, 56)
BORDER = (54, 57, 80)
MUTED = (124, 130, 156)
DOT = (78, 82, 108)
# The three title-bar dots, close-minimise-zoom from left to right. They are the recording's own
# red, amber and green - the normal ANSI row of `MUTED_THEME` below - rather than the saturated
# macOS traffic light, so the only colours in the chrome are colours the terminal itself can draw.
DOTS = ((214, 130, 130), (212, 182, 130), (140, 186, 146))
PADDING = 22
CORNER_RADIUS = 13
MUTED_THEME = TerminalTheme(
BACKGROUND,
FOREGROUND,
[(42, 44, 62), (214, 130, 130), (140, 186, 146), (212, 182, 130),
(138, 166, 214), (183, 152, 208), (134, 190, 197), (200, 205, 220)],
[(108, 113, 138), (226, 152, 152), (160, 203, 166), (228, 200, 150),
(158, 184, 226), (200, 172, 222), (154, 206, 212), (226, 230, 240)],
)
def find_font(bold: bool = False) -> Path:
"""A monospace face with box-drawing glyphs: matplotlib ships DejaVu Sans Mono everywhere."""
candidates: list[Path] = []
spec = importlib.util.find_spec("matplotlib")
if spec and spec.origin:
ttf = Path(spec.origin).parent / "mpl-data" / "fonts" / "ttf"
candidates.append(ttf / ("DejaVuSansMono-Bold.ttf" if bold else "DejaVuSansMono.ttf"))
windows = Path(os.environ.get("WINDIR", "C:/Windows")) / "Fonts"
candidates += [windows / ("consolab.ttf" if bold else "consola.ttf"),
Path("/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf")]
for path in candidates:
if path.exists():
return path
raise FileNotFoundError("no monospace font found; pass --font PATH")
def _rgb(style: Style, theme: TerminalTheme) -> tuple[tuple[int, int, int], tuple[int, int, int] | None]:
fg = tuple(style.color.get_truecolor(theme)) if style.color else FOREGROUND
bg = tuple(style.bgcolor.get_truecolor(theme)) if style.bgcolor else None
if style.dim:
fg = tuple(int(f * 0.55 + b * 0.45) for f, b in zip(fg, BACKGROUND, strict=True))
return fg, bg # type: ignore[return-value]
def _chrome(size: tuple[int, int], bar: int, font: FreeTypeFont, title: str,
shell: str) -> PilImage:
"""The window the frames are drawn into: title bar, dots, rounded border.
Drawn once and copied per frame; at a hundred frames, redrawing the furniture each time is
most of the render.
"""
from PIL import Image, ImageDraw
img = Image.new("RGB", size, BACKGROUND)
draw = ImageDraw.Draw(img)
width, height = size
draw.rectangle([0, 0, width, bar], fill=TITLE_BAR)
draw.line([(0, bar), (width, bar)], fill=BORDER)
for i in range(3):
cx, cy, r = 22 + i * 18, bar // 2, 5
draw.ellipse([cx - r, cy - r, cx + r, cy + r], fill=DOTS[i])
draw.text((width // 2, bar // 2), title, font=font, fill=MUTED, anchor="mm")
if shell:
draw.text((width - 20, bar // 2), shell, font=font, fill=DOT, anchor="rm")
draw.rounded_rectangle([0, 0, width - 1, height - 1], radius=CORNER_RADIUS,
outline=BORDER, width=1)
return img
def _outside_corners(size: tuple[int, int]) -> PilImage:
"""A mask of the pixels a rounded window does not cover, so the GIF can leave them clear."""
from PIL import Image, ImageDraw
mask = Image.new("L", size, 255)
ImageDraw.Draw(mask).rounded_rectangle([0, 0, size[0] - 1, size[1] - 1],
radius=CORNER_RADIUS, fill=0)
return mask.point(lambda v: 255 if v > 127 else 0)
def render_gif(
frames: list[Frame],
out: Path,
*,
cols: int,
font_size: int = 15,
font: Path | None = None,
theme: TerminalTheme = MUTED_THEME,
title: str = "coder-agent",
shell: str = "powershell",
) -> Path:
"""Draw every frame with Pillow and write an animated GIF with per-frame durations."""
from PIL import ImageDraw, ImageFont
regular = ImageFont.truetype(str(font or find_font()), font_size)
try:
strong = ImageFont.truetype(str(font or find_font(bold=True)), font_size)
except (FileNotFoundError, OSError):
strong = regular
chrome_font = ImageFont.truetype(str(font or find_font()), max(9, font_size - 2))
cw = max(1, round(regular.getlength("M")))
ascent, descent = regular.getmetrics()
lh = ascent + descent # no leading, so box-drawing glyphs on adjacent rows join up
rows = len(frames[0].cells)
bar = lh + 18
size = (cols * cw + 2 * PADDING, bar + rows * lh + 2 * PADDING)
window = _chrome(size, bar, chrome_font, title, shell)
# The corners are left transparent so the window sits on a light or a dark README alike.
corners = _outside_corners(size)
images = []
for frame in frames:
img = window.copy()
draw = ImageDraw.Draw(img)
for r, line in enumerate(frame.cells):
y = bar + PADDING + r * lh
c = 0
while c < len(line):
style = line[c][1]
end = c
while end < len(line) and line[end][1] == style:
end += 1
text = "".join(ch for ch, _ in line[c:end])
fg, bg = _rgb(style, theme)
x = PADDING + c * cw
if bg:
draw.rectangle([x, y, x + (end - c) * cw, y + lh], fill=bg)
draw.text((x, y), text, font=strong if style.bold else regular, fill=fg)
c = end
# One palette slot is reserved for "not painted": quantize to 255, then claim index 255.
paletted = img.quantize(colors=255)
paletted.putpalette(paletted.getpalette()[: 255 * 3] + list(BACKGROUND))
paletted.paste(255, mask=corners)
images.append(paletted)
durations = [max(20, int(f.duration * 1000)) for f in frames]
out.parent.mkdir(parents=True, exist_ok=True)
images[0].save(out, save_all=True, append_images=images[1:], duration=durations, loop=0,
transparency=255, disposal=1, optimize=True)
return out
# --------------------------------------------------------------------------- drawing: SVG
# The browser has no DejaVu Sans Mono to hand, so every run also carries the width it is supposed
# to occupy (`textLength`) and the glyphs are fitted to it. Without that, one reader's Consolas
# and another's Menlo put the stage grid in two different places.
SVG_FONTS = '"DejaVu Sans Mono","SFMono-Regular",Consolas,"Liberation Mono",Menlo,monospace'
_XML = {"&": "&", "<": "<", ">": ">", '"': """}
def _esc(text: str) -> str:
return "".join(_XML.get(c, c) for c in text)
def _hex(rgb: tuple[int, int, int]) -> str:
return "#{:02x}{:02x}{:02x}".format(*rgb)
def _runs(line: list[Cell]) -> Iterator[tuple[int, int, Style, str]]:
"""Split a row into maximal same-style stretches: (start column, end column, style, text)."""
c = 0
while c < len(line):
style = line[c][1]
end = c
while end < len(line) and line[end][1] == style:
end += 1
yield c, end, style, "".join(ch for ch, _ in line[c:end])
c = end
def _row_spans(frames: list[Frame]) -> list[tuple[int, int, int, list[Cell]]]:
"""Every stretch of frames over which one row holds one content: (row, first, last+1, cells).
This is what makes the SVG small. A GIF has to repaint all twenty-four rows on every frame;
a terminal transcript mostly appends, so a row drawn once usually stays until it scrolls off.
Emitting one group per *change* rather than per frame turns rows x frames repaints into a few
hundred groups, most of which live for most of the recording.
"""
spans: list[tuple[int, int, int, list[Cell]]] = []
for r in range(len(frames[0].cells)):
start, current = 0, frames[0].cells[r]
for i in range(1, len(frames) + 1):
cells = frames[i].cells[r] if i < len(frames) else None
if cells == current:
continue
if any(ch != " " or style.bgcolor for ch, style in current):
spans.append((r, start, i, current))
start, current = i, cells # type: ignore[assignment]
return spans
def render_svg(
frames: list[Frame],
out: Path,
*,
cols: int,
font_size: int = 15,
font: Path | None = None,
theme: TerminalTheme = MUTED_THEME,
title: str = "coder-agent",
shell: str = "powershell",
) -> Path:
"""Write the same recording as one animated SVG: CSS keyframes, no script, no <foreignObject>.
Scripts do not run inside an `<img>` and GitHub would strip them anyway, so the animation is
a CSS window per group: opacity 0, opacity 1 for its stretch of the timeline, opacity 0
again. Groups sharing a stretch share the rule, which is most of them - a screenful that
appears together disappears together.
"""
from PIL import ImageFont
regular = ImageFont.truetype(str(font or find_font()), font_size)
cw = max(1, round(regular.getlength("M")))
ascent, descent = regular.getmetrics()
lh = ascent + descent
rows = len(frames[0].cells)
bar = lh + 18
width = cols * cw + 2 * PADDING
height = bar + rows * lh + 2 * PADDING
small = max(9, font_size - 2)
total = sum(f.duration for f in frames) or 1.0
starts: list[float] = []
at = 0.0
for frame in frames:
starts.append(at)
at += frame.duration
starts.append(total)
# One CSS rule per distinct (appears, disappears) pair, keyed on the rounded percentages so
# that two groups a thousandth of a percent apart still share it.
windows: dict[tuple[str, str], int] = {}
body: list[str] = []
for row, first, last, cells in _row_spans(frames):
key = (f"{starts[first] / total * 100:.3f}", f"{starts[last] / total * 100:.3f}")
index = windows.setdefault(key, len(windows))
y = bar + PADDING + row * lh + ascent
parts: list[str] = []
for start, end, style, text in _runs(cells):
fg, bg = _rgb(style, theme)
x = PADDING + start * cw
run = (end - start) * cw
if bg:
parts.append(f'<rect x="{x}" y="{y - ascent}" width="{run}" height="{lh}" '
f'fill="{_hex(bg)}"/>')
# The padding a style run carries on either side is drawn by nothing, so it is
# dropped and the run starts where its first glyph does. That keeps `textLength`
# honest - it has to describe the glyphs it is given, not the column they sit in.
ink = text.strip(" ")
if ink:
lead = len(text) - len(text.lstrip(" "))
weight = ' font-weight="bold"' if style.bold else ""
parts.append(f'<text x="{x + lead * cw}" y="{y}" textLength="{len(ink) * cw}" '
f'fill="{_hex(fg)}"{weight}>{_esc(ink)}</text>')
if parts:
body.append(f'<g class="w{index}">{"".join(parts)}</g>')
rules: list[str] = []
for (appears, disappears), index in windows.items():
stops = [] if appears == "0.000" else [f"0%,{appears}%{{opacity:0}}"]
stops.append(f"{appears}%,{disappears}%{{opacity:1}}")
if disappears != "100.000":
stops.append(f"{disappears}%,100%{{opacity:0}}")
rules.append(f".w{index}{{animation-name:k{index}}}@keyframes k{index}{{{''.join(stops)}}}")
# What a still viewer should see is the finished run, not an empty terminal: a README opened
# with reduced motion on, or a preview thumbnail, gets the last frame.
final = ",".join(f".w{i}" for (_, gone), i in windows.items() if gone == "100.000")
css = (
# `white-space:pre` is not decoration: without it the browser collapses the runs of
# spaces inside a line, and `textLength` then stretches the surviving glyphs across the
# width the spaces were supposed to occupy. The whole grid comes apart.
f"text{{font-size:{font_size}px;white-space:pre}}"
f"g[class]{{opacity:0;animation-duration:{total:.2f}s;animation-iteration-count:infinite;"
f"animation-timing-function:step-end}}"
+ "".join(rules)
+ f"@media(prefers-reduced-motion:reduce){{g[class]{{animation:none}}{final}{{opacity:1}}}}"
)
dots = "".join(f'<circle cx="{22 + i * 18}" cy="{bar // 2}" r="5" fill="{_hex(dot)}"/>'
for i, dot in enumerate(DOTS))
right = (f'<text x="{width - 20}" y="{bar // 2}" fill="{_hex(DOT)}" text-anchor="end" '
f'dominant-baseline="central" font-size="{small}px">{_esc(shell)}</text>') if shell else ""
svg = (
f'<svg xmlns="http://www.w3.org/2000/svg" viewBox="0 0 {width} {height}" '
f'width="{width}" height="{height}" font-family=\'{SVG_FONTS}\' xml:space="preserve">'
f"<title>{_esc(title)}</title><style>{css}</style>"
f'<clipPath id="win"><rect width="{width}" height="{height}" rx="{CORNER_RADIUS}"/></clipPath>'
f'<g clip-path="url(#win)">'
f'<rect width="{width}" height="{height}" fill="{_hex(BACKGROUND)}"/>'
f'<rect width="{width}" height="{bar}" fill="{_hex(TITLE_BAR)}"/>'
f'<line x1="0" y1="{bar}.5" x2="{width}" y2="{bar}.5" stroke="{_hex(BORDER)}"/>'
f'{dots}'
f'<text x="{width // 2}" y="{bar // 2}" fill="{_hex(MUTED)}" text-anchor="middle" '
f'dominant-baseline="central" font-size="{small}px">{_esc(title)}</text>'
f'{right}{"".join(body)}</g>'
f'<rect x="0.5" y="0.5" width="{width - 1}" height="{height - 1}" rx="{CORNER_RADIUS}" '
f'fill="none" stroke="{_hex(BORDER)}"/></svg>'
)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(svg, encoding="utf-8")
return out
# --------------------------------------------------------------------------- drawing: JSON
# The web player is handed the screen model, not a picture of it: the same frames the GIF and the
# SVG are drawn from, as JSON, so a browser can paint them into a `<pre>`. Runs are positional and
# trailing defaults are dropped - `[text]`, `[text, fg]`, `[text, fg, bg]`, `[text, fg, bg, 1]` -
# because at four hundred rows of runs the key names would be most of the file.
def _row_runs(line: list[Cell], theme: TerminalTheme) -> list[list]:
"""One row as compact style runs, with the padding no glyph and no background needs dropped."""
runs: list[list] = []
for _, _, style, text in _runs(line):
fg, bg = _rgb(style, theme)
run: list = [text, "" if fg == FOREGROUND else _hex(fg), _hex(bg) if bg else "",
1 if style.bold else 0]
while len(run) > 1 and not run[-1]:
run.pop()
runs.append(run)
while runs and len(runs[-1]) == 1 and not runs[-1][0].strip():
runs.pop()
return runs
def render_frames(
frames: list[Frame],
out: Path,
*,
cols: int,
font_size: int = 15,
font: Path | None = None, # the browser draws in whatever monospace the reader has
theme: TerminalTheme = MUTED_THEME,
title: str = "coder-agent",
shell: str = "powershell",
) -> Path:
"""Write the frames as JSON: per frame its duration and only the rows that changed.
The same argument that makes the SVG small makes this small, and for the player it also makes
painting cheap: a terminal transcript mostly appends, so a frame touches a row or two and the
player rewrites a row or two rather than the whole screen. `fontSize` is a ceiling, not a
size - the player measures the reader's monospace font and shrinks until the columns fit.
"""
rows = len(frames[0].cells)
previous: list[list[Cell]] = [[] for _ in range(rows)]
payload: list[dict] = []
for frame in frames:
changed = {str(r): _row_runs(line, theme)
for r, line in enumerate(frame.cells) if line != previous[r]}
previous = frame.cells
payload.append({"d": round(frame.duration, 3), "rows": changed})
data = {
"version": 1,
"title": title,
"shell": shell,
"cols": cols,
"rows": rows,
"fontSize": font_size,
# The chrome is drawn by the stylesheet, but its colours come from here, so the window in
# the browser and the window in the GIF cannot drift apart without a re-render.
"theme": {
"background": _hex(BACKGROUND),
"foreground": _hex(FOREGROUND),
"titleBar": _hex(TITLE_BAR),
"border": _hex(BORDER),
"muted": _hex(MUTED),
"dot": _hex(DOT),
"dotClose": _hex(DOTS[0]),
"dotMin": _hex(DOTS[1]),
"dotMax": _hex(DOTS[2]),
"radius": f"{CORNER_RADIUS}px",
"padding": f"{PADDING}px",
},
"frames": payload,
}
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(data, separators=(",", ":")), encoding="utf-8")
return out
# --------------------------------------------------------------------------- commands
# Written into the demo copy of the task repository. The agent already runs pytest with `-q
# --no-header`, but the model reaches for a bare `pytest` through `run_command`, and that prints
# a platform banner, a plugin list and a rootdir line - three lines of nothing, in a recording
# that is twenty-four rows tall. Putting the flags in the repository is also what a real project
# does, so the demo is not a special case of the CLI.
#
# No `-q` here, deliberately. `addopts` is prepended to the command line, so a `-q` in both makes
# `-qq`, and the second one suppresses the `2 passed in 0.01s` line - which is the one line the
# `Tests` stage is built to show. `--no-header` already removes everything that was noisy.
PYTEST_INI = """[pytest]
addopts = --no-header -p no:cacheprovider
"""
def prepare(task_id: str, dest: Path) -> str:
"""Materialise one suite task into `dest` and return its prompt; the demo works on a copy."""
from coder_agent.evals.tasks import load_suite
task = next((t for t in load_suite() if t.id == task_id), None)
if task is None:
raise SystemExit(f"unknown suite task {task_id!r}")
if dest.exists():
raise SystemExit(f"{dest} exists; remove it or pick another directory")
task.materialise(dest)
ini = dest / "pytest.ini"
if not ini.exists():
ini.write_text(PYTEST_INI, encoding="utf-8")
return task.prompt.strip()
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
sub = parser.add_subparsers(dest="cmd", required=True)
p = sub.add_parser("prepare", help="copy a suite task to a directory and print its prompt")
p.add_argument("task_id")
p.add_argument("dest", type=Path)
r = sub.add_parser("record", help="run a command with stdout piped and write a .cast",
epilog="the command to record comes after --")
r.add_argument("out", type=Path)
r.add_argument("--title", help="what the GIF shows as the typed command (default: the command)")
r.add_argument("--answer", action="append", default=[], metavar="PATTERN=TEXT",
help="reply TEXT when PATTERN appears; default answers y to Approve?")
r.add_argument("--cols", type=int, default=100)
r.add_argument("--rows", type=int, default=28)
r.add_argument("--answer-delay", type=float, default=0.8)
r.add_argument("--timeout", type=float, default=900.0)
g = sub.add_parser("render",
help="draw a .cast as an animated GIF or SVG, or emit its frames as JSON")
g.add_argument("cast", type=Path)
g.add_argument("out", type=Path)
g.add_argument("--rows", type=int, help="visible rows (default: the cast height)")
g.add_argument("--max-gap", type=float, default=1.5, help="longest pause kept, seconds")
g.add_argument("--hold", type=float, default=3.0, help="how long the last frame stays")
g.add_argument("--font", type=Path)
g.add_argument("--font-size", type=int, default=15)
g.add_argument("--title", default="coder-agent", help="what the window title bar says")
g.add_argument("--shell", default="powershell", help="right-hand label in the title bar")
g.add_argument("--frames", action="store_true",
help="write the screen model as JSON for the web player (implied by .json)")
argv = sys.argv[1:] if argv is None else list(argv)
# argparse's REMAINDER would also swallow the options before it, so split on `--` by hand.
command: list[str] = []
if "--" in argv:
command = argv[argv.index("--") + 1:]
argv = argv[: argv.index("--")]
args = parser.parse_args(argv)
if args.cmd == "prepare":
sys.stdout.reconfigure(encoding="utf-8") # type: ignore[union-attr]
print(prepare(args.task_id, args.dest))
return 0
if args.cmd == "record":
if not command:
parser.error("give the command to record after --")
answers = dict(a.split("=", 1) for a in args.answer)
code = record(command, args.out, answers=answers, cols=args.cols, rows=args.rows,
answer_delay=args.answer_delay, timeout=args.timeout, title=args.title)
print(f"recorded {args.out} (exit {code})")
return code
cast = Cast.load(args.cast)
frames = frames_from_cast(cast, rows=args.rows, max_gap=args.max_gap, hold=args.hold)
suffix = args.out.suffix.lower()
if args.frames or suffix == ".json":
draw = render_frames
else:
draw = render_svg if suffix == ".svg" else render_gif
draw(frames, args.out, cols=int(cast.header.get("width", 100)),
font=args.font, font_size=args.font_size, title=args.title, shell=args.shell)
print(f"wrote {args.out}: {len(frames)} frames, {args.out.stat().st_size / 1e3:.0f} kB")
return 0
if __name__ == "__main__":
raise SystemExit(main())
You can’t perform that action at this time.
