ai-app/app/bench-fixture/generate.py

#!/usr/bin/env python3
"""Generates transcript.jsonl -- the synthetic fixture P0's benchmark opens in both apps.

Deterministic (fixed seed), so a Compose bench APK and an iris bench APK draw byte-identical
content: the point of the fixture is a like-for-like comparison, not a realistic one.

Never a real transcript -- see AGENTS.md's ui-sandbox.sh, which this borrows its vocabulary
style from (headings, code fences, a table, a link) rather than reusing its Claude-Code JSONL
shape. This file's shape is the *app's own event model* instead: one JSON object per line,
matching what GET /sessions/{id}/transcript returns and what Events.kt's parseSeqEvent reads
(server/src/session/driver.rs is the source of truth for the field names).

    ./generate.py                      writes transcript.jsonl and bench1.png/bench2.png here

BACKLOG_COUNT events (seq 1..BACKLOG_COUNT) are the scrolled-back history the benchmark opens
with. A further STREAM_COUNT events (seq BACKLOG_COUNT+1..) are not part of the opening window;
both bench harnesses replay them at a fixed rate as the "streaming reply" phase, appended through
the same live path a real SSE reply arrives on. Keeping both halves in one file means one
generator and one seed to keep in sync, rather than two fixtures that can drift apart.
"""
import base64
import json
import random
import struct
import zlib
from pathlib import Path

SEED = 20260905
BACKLOG_COUNT = 3200
STREAM_COUNT = 400
HERE = Path(__file__).resolve().parent / "assets"

random.seed(SEED)

LANGUAGES = ["rust", "kotlin", "python", "sh", "json", "toml"]

CODE_SNIPPETS = {
    "rust": '''fn fold_event(items: Vec<Item>, seq: u64) -> Vec<Item> {
    // a comment worth keeping: this is the fold the app's own screen runs
    let mut out = items;
    out.push(Item::new(seq));
    out
}''',
    "kotlin": '''fun foldEvent(items: List<TranscriptItem>, entry: SeqEvent): List<TranscriptItem> {
    // mirrors the server's own event model, one item per line
    return items + TranscriptItem.from(entry)
}''',
    "python": '''def render_report(frames, cpu_ms, rss_kb):
    # printed for a human to paste back, so every number carries its unit
    return f"{frames} frames, {cpu_ms}ms cpu, {rss_kb}kb peak rss"''',
    "sh": '''#!/bin/sh
# scripted scroll loop, the shape transcript-bench.sh drives on a phone
for i in $(seq 1 24); do
    ui-trace record --do "swipe 540 700 540 1600 200"
done''',
    "json": '{"seq": 1, "type": "status", "state": "running"}',
    "toml": '''[package]
name = "bench-fixture"
version = "0.1.0"''',
}

HEADINGS = [
    "## Plan",
    "## What changed",
    "## Why this approach",
    "### Open questions",
    "## Results",
]

WORDS = (
    "session render report frame budget scroll transcript fold event cache "
    "cursor probe stream backlog swipe fixture bench compose iris widget layout "
    "measure place draw tool call token context window anchor"
).split()


def paragraph(n=24):
    words = [random.choice(WORDS) for _ in range(n)]
    words[0] = words[0].capitalize()
    text = " ".join(words) + "."
    # Sprinkle markdown inline spans so the syntax highlighter/markdown parser sees a real mix.
    text = text.replace(" fold ", " **fold** ", 1)
    text = text.replace(" cursor ", " *cursor* ", 1)
    text = text.replace(" cache ", " `cache` ", 1)
    if "bench" in text:
        text = text.replace(
            " bench ", " [bench](https://example.com/bench) ", 1
        )
    return text


def make_png(rgb, size=8):
    """A tiny, valid PNG -- flat colour, no external dependency."""

    def chunk(tag, data):
        c = tag + data
        return struct.pack(">I", len(data)) + c + struct.pack(">I", zlib.crc32(c))

    sig = b"\x89PNG\r\n\x1a\n"
    ihdr = struct.pack(">IIBBBBB", size, size, 8, 2, 0, 0, 0)
    raw = b""
    for _ in range(size):
        raw += b"\x00" + bytes(rgb) * size
    idat = zlib.compress(raw)
    return sig + chunk(b"IHDR", ihdr) + chunk(b"IDAT", idat) + chunk(b"IEND", b"")


def main():
    HERE.mkdir(exist_ok=True)
    lines = []
    seq = 1
    ts = 1_788_000_000.0

    def emit(type_, **fields):
        nonlocal seq, ts
        obj = {"seq": seq, "ts": round(ts, 3), "type": type_}
        obj.update(fields)
        lines.append(json.dumps(obj, separators=(",", ":")))
        seq += 1
        ts += random.uniform(0.05, 2.0)

    emit("status", state="running")
    emit("settings", model="bench-model", permissionMode="auto")

    image_refs = []
    turn = 0
    while seq <= BACKLOG_COUNT:
        turn += 1
        emit("userMessage", text=f"Turn {turn}: {paragraph(12)}", id=None, attachments=[])

        # A tool call with kilobyte-scale input/output every few turns.
        if turn % 3 == 0:
            tool_id = f"tool-{turn}"
            big_input = json.dumps({"path": f"/repo/file_{turn}.rs", "content": paragraph(400)})
            emit("toolStart", id=tool_id, tool="Edit", input=big_input)
            big_output = "\n".join(paragraph(60) for _ in range(20))
            emit("toolUpdate", id=tool_id, output=big_output[: len(big_output) // 2])
            emit("toolEnd", id=tool_id, output=big_output)

        # A reply: a heading, prose, a fenced block in a rotating language, a table, then deltas.
        emit("assistantText", delta=f"{random.choice(HEADINGS)}\n\n")
        emit("assistantText", delta=paragraph(30) + "\n\n")
        lang = LANGUAGES[turn % len(LANGUAGES)]
        emit("assistantText", delta=f"```{lang}\n{CODE_SNIPPETS[lang]}\n```\n\n")
        if turn % 5 == 0:
            emit(
                "assistantText",
                delta="| column | value |\n|---|---|\n| a | " + paragraph(3) + " |\n\n",
            )
        # A run of small deltas -- the shape a live reply actually streams in.
        for _ in range(random.randint(3, 8)):
            emit("assistantText", delta=paragraph(6) + " ")

        # A couple of images, base64 PNGs, the way a real transcript embeds a screenshot.
        if turn in (10, 40):
            ref = f"bench{len(image_refs) + 1}.png"
            image_refs.append(ref)
            emit("image", ref=ref, about=None)

        emit("usageDelta", tokens=random.randint(200, 4000), context=random.randint(2000, 180000))

        if turn % 15 == 0:
            emit(
                "compacted",
                preTokens=180000,
                postTokens=20000,
                trigger="auto",
            )

    # The streaming-phase tail: one long reply, built entirely from text deltas, the shape a
    # bench harness replays at a fixed events/sec through the live fold path.
    emit("userMessage", text="One more, streamed live for the benchmark's timing phase.", id=None, attachments=[])
    while seq <= BACKLOG_COUNT + STREAM_COUNT:
        emit("assistantText", delta=paragraph(5) + " ")
    emit("status", state="idle")

    (HERE / "transcript.jsonl").write_text("\n".join(lines) + "\n")

    (HERE / "bench1.png").write_bytes(make_png((220, 90, 90)))
    (HERE / "bench2.png").write_bytes(make_png((90, 150, 220)))

    print(f"wrote {len(lines)} events ({BACKLOG_COUNT} backlog + {STREAM_COUNT} stream) to transcript.jsonl")


if __name__ == "__main__":
    main()