claude-plugin-inference-arbitrage/tests/fixtures/transcripts/generate.py

#!/usr/bin/env python3
"""Regenerates the synthetic fixture transcripts in this directory.

Every value here is invented. NOTHING in these fixtures comes from a real
session — real transcripts contain secrets and private work and must never be
committed (design/spec.md FR-3.5). The committed .jsonl files are the test
input; this generator exists so that a threshold change can be reflected in
them without hand-editing JSON.

Shape mimics ~/.claude/projects/**/*.jsonl: one JSON object per line, assistant
lines carrying message.usage + attributionPlugin/attributionSkill (main
sessions) or attributionAgent (subagent sidechain files), user lines carrying
tool_result blocks.

The numbers are chosen so every metric is hand-computable — see
tests/scan.test.sh for the arithmetic each assertion checks.
"""

import json
import os
from datetime import datetime, timedelta, timezone

HERE = os.path.dirname(os.path.abspath(__file__))
PROJECT = os.path.join(HERE, "fake-project")
MODEL = "claude-sonnet-5"
T0 = datetime(2026, 7, 1, 12, 0, 0, tzinfo=timezone.utc)

# cache_read 100k + output 100, tier 0.4 (sonnet-5 input $2/Mtok / 5)
#   weighted = 0.4 * (100000*0.1 + 100*5) = 4200 per inline turn
CTX = 100_000
SIDECHAIN_CTX = 200_000

_counter = [0]


def usage(ctx, out):
    return {
        "input_tokens": 0,
        "cache_creation_input_tokens": 0,
        "cache_read_input_tokens": ctx,
        "output_tokens": out,
    }


def assistant(session, minute, content, out, ctx=CTX, sidechain=False, **attrib):
    _counter[0] += 1
    line = {
        "type": "assistant",
        "uuid": f"u-{_counter[0]}",
        "requestId": f"req-{_counter[0]}",
        "sessionId": session,
        "isSidechain": sidechain,
        "cwd": "/fake/project",
        "timestamp": (T0 + timedelta(minutes=minute))
        .isoformat()
        .replace("+00:00", "Z"),
        "message": {
            "id": f"msg-{_counter[0]}",
            "model": MODEL,
            "role": "assistant",
            "content": content,
            "usage": usage(ctx, out),
        },
    }
    line.update(attrib)
    return line


def tool_use(tid, name, args):
    return {"type": "tool_use", "id": tid, "name": name, "input": args}


def text(body):
    return {"type": "text", "text": body}


def tool_result(session, minute, tid, body, is_error=False):
    return {
        "type": "user",
        "uuid": f"r-{tid}",
        "sessionId": session,
        "isSidechain": False,
        "timestamp": (T0 + timedelta(minutes=minute))
        .isoformat()
        .replace("+00:00", "Z"),
        "message": {
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": tid,
                    "content": body,
                    "is_error": is_error,
                }
            ],
        },
    }


# Deliberately verbose so read amplification is > 1: the model quotes back only
# one word of a long result.
HAYSTACK = " ".join(f"haystack_line_{i} filler filler filler" for i in range(200))

SKILL = {
    "attributionPlugin": "fixture-plugin",
    "attributionSkill": "fixture-plugin:mine-me",
}
THINK = {
    "attributionPlugin": "fixture-plugin",
    "attributionSkill": "fixture-plugin:think-hard",
}
AGENT = {
    "attributionPlugin": "fixture-plugin",
    "attributionAgent": "fixture-plugin:fetcher-agent",
}


def session_a():
    """3 invocations of the same 3-tool mechanical sequence, 30 min apart."""
    lines = []
    for inv in range(3):
        base = inv * 30
        lines.append(
            assistant(
                "sess-a",
                base,
                [
                    tool_use(
                        f"t{inv}a", "Grep", {"pattern": "needle", "path": "/fake/x"}
                    )
                ],
                100,
                **SKILL,
            )
        )
        lines.append(tool_result("sess-a", base, f"t{inv}a", HAYSTACK))
        lines.append(
            assistant(
                "sess-a",
                base + 1,
                [
                    text("needle"),
                    tool_use(f"t{inv}b", "Read", {"file_path": "/fake/y.md"}),
                ],
                100,
                **SKILL,
            )
        )
        lines.append(tool_result("sess-a", base + 1, f"t{inv}b", "short result"))
        lines.append(
            assistant(
                "sess-a",
                base + 2,
                [
                    tool_use(
                        f"t{inv}c", "mcp__plugin_tracker__issue_read", {"index": 41}
                    )
                ],
                100,
                **SKILL,
            )
        )
        lines.append(tool_result("sess-a", base + 2, f"t{inv}c", "ok"))
    # One unattributed turn in the same session: counts toward candidate_turns
    # (coverage denominator) but never toward the plugin's spend.
    lines.append(assistant("sess-a", 95, [text("unrelated work")], 500))
    return lines


def session_b():
    """One invocation exercising judgment, a mutating tool, and a retry."""
    lines = [
        assistant("sess-b", 0, [text("a long deliberation " * 50)], 2000, **THINK),
        assistant(
            "sess-b",
            1,
            [tool_use("tb1", "Write", {"file_path": "/fake/out.md", "content": "x"})],
            100,
            **THINK,
        ),
        tool_result("sess-b", 1, "tb1", "permission denied", is_error=True),
        assistant(
            "sess-b",
            2,
            [tool_use("tb2", "Write", {"file_path": "/fake/out.md", "content": "x"})],
            100,
            **THINK,
        ),
    ]
    return lines


def session_c():
    """Sidechain file: agent-attributed, double the inline context."""
    return [
        assistant(
            "sess-c",
            0,
            [tool_use("tc1", "Glob", {"pattern": "*.md"})],
            100,
            ctx=SIDECHAIN_CTX,
            sidechain=True,
            **AGENT,
        ),
        assistant(
            "sess-c",
            1,
            [tool_use("tc2", "Read", {"file_path": "/fake/z.md"})],
            100,
            ctx=SIDECHAIN_CTX,
            sidechain=True,
            **AGENT,
        ),
    ]


def session_d():
    """A session the target never touched: must not reach the denominator."""
    return [
        assistant("sess-d", 0, [text("someone else's work")], 300),
        assistant("sess-d", 1, [text("still not ours")], 300),
    ]


def write(path, lines):
    os.makedirs(os.path.dirname(path), exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        for line in lines:
            f.write(json.dumps(line) + "\n")


def main():
    write(os.path.join(PROJECT, "sess-a.jsonl"), session_a())
    write(os.path.join(PROJECT, "sess-b.jsonl"), session_b())
    write(
        os.path.join(PROJECT, "sess-c", "subagents", "agent-fetcher.jsonl"), session_c()
    )
    write(os.path.join(PROJECT, "sess-d.jsonl"), session_d())


if __name__ == "__main__":
    main()