Privacy-preserving, Showboat-recorded local Gemma 4 comparison protocol

README.md

Privacy-preserving Gemma 4 comparison notebook

This Gist contains a reproducible local protocol for comparing stock Ollama gemma4:26b and TurboFieldfare on confidential instructional transcripts. It intentionally contains no confidential source path, source filename, title, topic, transcript, summary, raw backend log, model directory, API key, or private configuration.

private_benchmark.py executes exactly one requested action. It does not loop over the twelve chunks. It rejects result directories outside .tmp, captures only public-safe status to stdout, applies a finite timeout to the actual backend command, and stores all source-derived artifacts privately.

Install Showboat v0.6.1 on Darwin arm64 with checksum verification:

bash install_showboat.sh
export PATH="$PWD/.bin:$PATH"

Copy private-config.template.json to .tmp/showboat-private-config.json, supply only local confidential paths in that ignored file, and run the SHOWBOAT_PROTOCOL.md commands one at a time. The published notebook is built by Showboat, so it mixes explanatory notes, the exact path-safe commands, and captured public-safe output. If an entry is wrong or fails, inspect only local private artifacts then remove the entry with showboat pop notebook.md before retrying that one command.

The private wrapper imports the sibling corpus and backend scripts in this Gist. TurboFieldfare uses the Gemma generation configuration (temperature 1.0, Top-K 64, Top-P 0.95; repetition penalty 1.0). Ollama intentionally retains its native local defaults, so results are reported as a real-user-path comparison, not sampler-normalized throughput or quality evidence.

The final notebook provides an anonymized Video A/Video B table with duration rounded to the nearest minute and independently one-time ±3% jittered sizes. It also reports elapsed time and sampled apparent RSS. The private table must never be joined with public source metadata.

Remote pairwise judging is forbidden for these confidential materials. If quality is assessed, use an authorized human locally or a separately approved local-only judge with blind labels; report it as a small local signal, not a general quality claim.

local_pairwise.py prepares and scores those blind pairs strictly inside .tmp/. It prints only pair counts and aggregate wins/ties, never summary text. Use it only after the context sweep, comparing each context candidate to the 4K baseline on the largest block.

install_showboat.sh

#!/usr/bin/env bash
# Install only the pinned Showboat release used by the public experiment note.
set -euo pipefail

showboat_script_dir="$(cd -- "$(dirname -- "${BASH_SOURCE[0]}")" && pwd)"
if [[ -f "$showboat_script_dir/../../Package.swift" ]]; then
  showboat_root="$(cd -- "$showboat_script_dir/../.." && pwd)"
else
  showboat_root="$showboat_script_dir"
fi
showboat_bin_dir="${SHOWBOAT_BIN_DIR:-$showboat_root/.bin}"
showboat_version="v0.6.1"
showboat_asset="showboat-darwin-arm64.tar.gz"
showboat_sha256="92005313da37b534fa70845f32b544f40842f0cc909903b395bd571ccf296f74"
showboat_url="https://github.com/simonw/showboat/releases/download/${showboat_version}/${showboat_asset}"

case "$(uname -s)-$(uname -m)" in
  Darwin-arm64) ;;
  *)
    echo "error: this installer is pinned to Darwin arm64, found $(uname -s)-$(uname -m)" >&2
    exit 2
    ;;
esac

mkdir -p "$showboat_bin_dir"
temporary_dir="$(mktemp -d)"
trap 'rm -rf "$temporary_dir"' EXIT
archive="$temporary_dir/$showboat_asset"

curl --fail --location --silent --show-error --retry 3 \
  --output "$archive" "$showboat_url"
actual_sha256="$(shasum -a 256 "$archive" | awk '{print $1}')"
if [[ "$actual_sha256" != "$showboat_sha256" ]]; then
  echo "error: Showboat archive SHA-256 mismatch" >&2
  exit 1
fi

tar -xzf "$archive" -C "$temporary_dir"
install -m 0755 "$temporary_dir/showboat" "$showboat_bin_dir/showboat"
"$showboat_bin_dir/showboat" --version

local_pairwise.py

#!/usr/bin/env python3
"""Prepare and score local-only blinded pairs without publishing text."""
from __future__ import annotations

import argparse
import json
from pathlib import Path
import secrets
import sys


def private_path(path: Path) -> Path:
    resolved = path.resolve()
    root = (Path(__file__).resolve().parents[2] / ".tmp").resolve()
    try:
        resolved.relative_to(root)
    except ValueError as error:
        raise ValueError("all blind-pair artifacts must stay inside .tmp") from error
    return resolved


def prepare(args: argparse.Namespace) -> int:
    output = private_path(args.output)
    if output.exists():
        raise ValueError("blind-pair output already exists")
    candidate_a = private_path(args.candidate_a)
    candidate_b = private_path(args.candidate_b)
    pairs = []
    for index in args.chunks:
        filename = f"chunk-{index:02d}.json"
        a = (candidate_a / f"chunk-{index:02d}" / "summary.md").read_text(encoding="utf-8")
        b = (candidate_b / f"chunk-{index:02d}" / "summary.md").read_text(encoding="utf-8")
        if secrets.randbits(1):
            left, right, answer = a, b, "A"
        else:
            left, right, answer = b, a, "B"
        pairs.append({"pair": index, "left": left, "right": right, "answer": answer})
        output.mkdir(parents=True, exist_ok=False) if not output.exists() else None
        (output / filename).write_text(json.dumps(pairs[-1]), encoding="utf-8")
    (output / "instructions.md").write_text("For each pair, choose LEFT, RIGHT, or TIE for faithfulness and usefulness. Record only the choice in ratings.json. Do not copy text outside this private directory.\n", encoding="utf-8")
    print(f"blind pairs prepared={len(pairs)}")
    return 0


def score(args: argparse.Namespace) -> int:
    pairs = private_path(args.pairs)
    ratings = json.loads(private_path(args.ratings).read_text(encoding="utf-8"))
    if not isinstance(ratings, dict):
        raise ValueError("ratings must be an object mapping pair numbers to LEFT, RIGHT, or TIE")
    aggregate = {"baseline_wins": 0, "candidate_wins": 0, "ties": 0, "invalid": 0}
    for path in sorted(pairs.glob("chunk-*.json")):
        pair = json.loads(path.read_text(encoding="utf-8"))
        rating = ratings.get(str(pair["pair"]), "")
        if rating == "TIE":
            aggregate["ties"] += 1
        elif rating in {"LEFT", "RIGHT"}:
            winner = pair["answer"] if rating == "LEFT" else ("B" if pair["answer"] == "A" else "A")
            aggregate["candidate_wins" if winner == "A" else "baseline_wins"] += 1
        else:
            aggregate["invalid"] += 1
    output = private_path(args.output)
    output.write_text(json.dumps(aggregate, indent=2) + "\n", encoding="utf-8")
    print("pairwise score baseline_wins={baseline_wins} candidate_wins={candidate_wins} ties={ties} invalid={invalid}".format(**aggregate))
    return 0


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="action", required=True)
    prep = sub.add_parser("prepare")
    prep.add_argument("--candidate-a", type=Path, required=True)
    prep.add_argument("--candidate-b", type=Path, required=True)
    prep.add_argument("--output", type=Path, required=True)
    prep.add_argument("--chunks", type=int, nargs="+", required=True)
    scored = sub.add_parser("score")
    scored.add_argument("--pairs", type=Path, required=True)
    scored.add_argument("--ratings", type=Path, required=True)
    scored.add_argument("--output", type=Path, required=True)
    args = parser.parse_args(argv)
    return prepare(args) if args.action == "prepare" else score(args)


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (OSError, ValueError, json.JSONDecodeError) as error:
        print(f"error: {error}", file=sys.stderr)
        raise SystemExit(2)

prepare_benchmark_corpus.py

#!/usr/bin/env python3
"""Prepare an immutable, local benchmark corpus from an external course tree."""
from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import sys

TOOL_VERSION = "1"
CHUNK_SIZE = 11_000
CHUNK_STEP = 10_000
BENCHMARK_COUNT = 12


def digest_bytes(value: bytes) -> str:
    return hashlib.sha256(value).hexdigest()


def is_within(child: Path, parent: Path) -> bool:
    try:
        child.relative_to(parent)
        return True
    except ValueError:
        return False


def select_evenly(count: int) -> list[int]:
    if count < BENCHMARK_COUNT:
        raise ValueError("the course corpus must produce at least 12 chunks")
    # Integer arithmetic makes selection stable across Python versions.
    return [number * (count - 1) // (BENCHMARK_COUNT - 1) for number in range(BENCHMARK_COUNT)]


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("course_root", type=Path)
    parser.add_argument("output_root", type=Path)
    args = parser.parse_args(argv)
    course = args.course_root.resolve()
    output = args.output_root.resolve()
    if not course.is_dir():
        parser.error("course root must be an existing directory")
    if is_within(output, course):
        parser.error("output root must not be the course root or beneath the course tree")

    sources = sorted(path for path in course.rglob("*_video_*.md") if path.is_file())
    if not sources:
        parser.error("course contains no *_video_*.md transcripts")
    source_bytes = [path.read_bytes() for path in sources]
    try:
        source_text = [item.decode("utf-8") for item in source_bytes]
    except UnicodeDecodeError as error:
        parser.error("course transcript is not UTF-8: %s" % error)
    combined = "\n\n".join(source_text)
    all_chunks = [combined[offset : offset + CHUNK_SIZE] for offset in range(0, len(combined) - CHUNK_SIZE + 1, CHUNK_STEP)]
    try:
        selected = select_evenly(len(all_chunks))
    except ValueError as error:
        parser.error(str(error))

    # Validate every condition before creating any output directory.
    output.mkdir(parents=True, exist_ok=True)
    chunks = []
    for benchmark_number, source_index in enumerate(selected):
        name = "benchmark%02d.md" % benchmark_number
        text = all_chunks[source_index]
        encoded = text.encode("utf-8")
        (output / name).write_bytes(encoded)
        chunks.append({
            "path": name,
            "source_chunk_index": source_index,
            "offset_chars": source_index * CHUNK_STEP,
            "sha256": digest_bytes(encoded),
            "bytes": len(encoded),
        })
    manifest = {
        "format": "turbofieldfare-summarisation-benchmark-v1",
        "tool_version": TOOL_VERSION,
        "chunk_size_chars": CHUNK_SIZE,
        "chunk_step_chars": CHUNK_STEP,
        "sources": [
            {"path": str(path.relative_to(course)), "sha256": digest_bytes(data), "bytes": len(data)}
            for path, data in zip(sources, source_bytes)
        ],
        "selected_indices": selected,
        "chunks": chunks,
    }
    (output / "manifest.json").write_text(json.dumps(manifest, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except BrokenPipeError:
        raise SystemExit(1)

private-config.template.json

{
  "course_root": "/replace/with/confidential/course/root",
  "model_dir": "scratch/gemma4.gturbo",
  "results_root": ".tmp/confidential-gemma4-comparison",
  "summary_prompt": "Summarise the following instructional transcript faithfully and concisely. Preserve substantive facts, figures, and recommendations.",
  "timeout_seconds": 1800
}

private_benchmark.py

#!/usr/bin/env python3
"""Run one privacy-preserving benchmark action at a time.

The private config and every source-derived artifact remain under .tmp. Stdout
is intentionally public-safe: it never contains source text, paths, names,
topics, summaries, raw errors, or commands with confidential values.
"""
from __future__ import annotations

import argparse
import hashlib
import json
from pathlib import Path
import platform
import re
import secrets
import shutil
import subprocess
import sys
import time


HERE = Path(__file__).resolve().parent
ROOT = HERE if (HERE / "prepare_benchmark_corpus.py").is_file() else HERE.parents[1]
PREPARE = ROOT / "summarisation" / "prepare_benchmark_corpus.py"
TF = ROOT / "summarisation" / "process_chunks_turbofieldfare.py"
OLLAMA = ROOT / "summarisation" / "process_chunks_ollama.py"
DEFAULT_CONFIG = ROOT / ".tmp" / "showboat-private-config.json"
OWNER_PATTERN = "TurboFieldfareServer|TurboFieldfareMac|TurboFieldfareDecodeService|TurboFieldfareCLI|TurboFieldfarePackageTests|swiftpm-testing-helper|mlx_lm|mlx-lm"
STAMP = re.compile(r"(?<!\d)(?:(\d{1,2}):)?(\d{1,2}):(\d{2})(?!\d)")


def load_config(path: Path) -> dict:
    try:
        config = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as error:
        raise ValueError("private configuration is unavailable or invalid") from error
    required = {"course_root", "model_dir", "results_root", "summary_prompt", "timeout_seconds"}
    if not required.issubset(config) or not isinstance(config["timeout_seconds"], int):
        raise ValueError("private configuration is missing required values")
    return config


def root_for(config: dict) -> Path:
    root = Path(config["results_root"])
    root = (ROOT / root if not root.is_absolute() else root).resolve()
    try:
        root.relative_to((ROOT / ".tmp").resolve())
    except ValueError as error:
        raise ValueError("results_root must be inside .tmp") from error
    return root


def text(argv: list[str]) -> str:
    try:
        return subprocess.run(argv, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True, check=False, timeout=30).stdout
    except subprocess.TimeoutExpired:
        return ""


def free_memory() -> int | None:
    match = re.search(r"memory free percentage:\s*(\d+)%", text(["memory_pressure", "-Q"]))
    return int(match.group(1)) if match else None


def model_owner_present() -> bool:
    return bool(text(["pgrep", "-fl", OWNER_PATTERN]).strip())


def ollama_loaded() -> bool:
    return len(text(["ollama", "ps"]).splitlines()) > 1


def timestamp_duration(text_value: str) -> int:
    maximum = 0
    for hour, minute, second in STAMP.findall(text_value):
        maximum = max(maximum, int(hour or 0) * 3600 + int(minute) * 60 + int(second))
    return maximum


def video_inventory(config: dict, destination: Path) -> int:
    course = Path(config["course_root"]).resolve()
    rows = []
    for index, source in enumerate(sorted(path for path in course.rglob("*_video_*.md") if path.is_file())):
        body = source.read_text(encoding="utf-8")
        factor = 0.97 + secrets.randbelow(60_001) / 1_000_000
        label = "Video " + chr(65 + index) if index < 26 else f"Video {index + 1}"
        rows.append({"video": label, "duration_minutes": round(timestamp_duration(body) / 60), "reported_kib": max(1, round(source.stat().st_size * factor / 1024))})
    destination.write_text(json.dumps(rows, indent=2) + "\n", encoding="utf-8")
    return len(rows)


def prepare(config: dict, root: Path) -> None:
    if model_owner_present():
        raise RuntimeError("a model-owning process is already active")
    corpus = root / "corpus"
    if corpus.exists():
        raise RuntimeError("private corpus already exists")
    try:
        completed = subprocess.run([sys.executable, str(PREPARE), config["course_root"], str(corpus)], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=120)
    except subprocess.TimeoutExpired as error:
        raise RuntimeError("corpus preparation timed out") from error
    if completed.returncode:
        raise RuntimeError("corpus preparation failed")
    videos = video_inventory(config, root / "public-video-table.json")
    manifest = json.loads((corpus / "manifest.json").read_text(encoding="utf-8"))
    print(f"prepared chunks={len(manifest['chunks'])} videos={videos}")


def child_pids(pid: int) -> set[int]:
    result, pending = {pid}, [pid]
    while pending:
        for token in text(["pgrep", "-P", str(pending.pop())]).split():
            child = int(token)
            if child not in result:
                result.add(child)
                pending.append(child)
    return result


def rss_mib(pids: set[int]) -> float:
    if not pids:
        return 0.0
    values = text(["ps", "-o", "rss=", "-p", ",".join(map(str, sorted(pids)))]).split()
    return sum(int(value) for value in values if value.isdigit()) / 1024


def launch(command: list[str], stdout: Path, stderr: Path, timeout_seconds: int, sample_ollama: bool) -> tuple[int, float, float]:
    with stdout.open("wb") as out, stderr.open("wb") as err:
        process = subprocess.Popen(command, stdout=out, stderr=err)
        started, peak = time.monotonic(), 0.0
        while process.poll() is None:
            if time.monotonic() - started > timeout_seconds:
                process.terminate()
                process.wait(timeout=15)
                raise RuntimeError("command timed out")
            pids = child_pids(process.pid)
            if sample_ollama:
                pids.update(int(value) for value in text(["pgrep", "-f", "Ollama.app|ollama serve"]).split() if value.isdigit())
            peak = max(peak, rss_mib(pids))
            time.sleep(0.25)
        return process.returncode, time.monotonic() - started, peak


def run_one(config: dict, root: Path, backend: str, chunk: int, context: int, warmup: bool) -> None:
    if context not in {4096, 8192, 16384, 32768, 65536}:
        raise ValueError("unsupported context")
    if model_owner_present():
        raise RuntimeError("a TurboFieldfare model-owning process is already active")
    manifest = json.loads((root / "corpus" / "manifest.json").read_text(encoding="utf-8"))
    if not 0 <= chunk < len(manifest["chunks"]):
        raise ValueError("invalid chunk index")
    run_root = root / ("warmup" if warmup else "measured") / backend / f"context-{context}" / f"chunk-{chunk:02d}"
    if run_root.exists():
        raise RuntimeError("result already exists")
    run_root.mkdir(parents=True)
    source, output = root / "corpus" / f"benchmark{chunk:02d}.md", run_root / "summary.md"
    before = free_memory()
    if backend == "turbofieldfare":
        command = [sys.executable, str(TF), "--input", str(source), "--output", str(output), "--cli", str(ROOT / ".build" / "release" / "TurboFieldfareCLI"), "--model", config["model_dir"], "--prompt", config["summary_prompt"], "--max-context", str(context), "--temperature", "1.0", "--top-k", "64", "--top-p", "0.95", "--repetition-penalty", "1.0"]
        sample_ollama = False
    else:
        command = [sys.executable, str(OLLAMA), "--input", str(source), "--output", str(output)]
        sample_ollama = True
    exit_code, wall, peak = launch(command, run_root / "stdout.bin", run_root / "stderr.bin", config["timeout_seconds"], sample_ollama)
    item = {"backend": backend, "chunk": chunk, "context": context, "warmup": warmup, "exit_code": exit_code, "wall_seconds": wall, "peak_rss_mib": peak, "memory_free_before_percent": before, "memory_free_after_percent": free_memory(), "output_bytes": output.stat().st_size if output.exists() else 0, "output_sha256": hashlib.sha256(output.read_bytes()).hexdigest() if output.exists() else None}
    (run_root / "measurement.json").write_text(json.dumps(item, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    ok = exit_code == 0 and output.exists()
    print(f"{backend} chunk={chunk:02d} context={context} status={'ok' if ok else 'failed'} wall_seconds={wall:.3f} peak_rss_mib={peak:.1f} memory_free_before={before} memory_free_after={item['memory_free_after_percent']}")
    if not ok:
        raise RuntimeError("backend failed; inspect private artifacts")


def report(root: Path, destination: Path) -> None:
    videos = json.loads((root / "public-video-table.json").read_text(encoding="utf-8"))
    records = [json.loads(path.read_text(encoding="utf-8")) for path in sorted((root / "measured").rglob("measurement.json"))]
    lines = ["# Confidential-material Gemma 4 comparison", "", "This Showboat notebook excludes source text, summaries, source paths, filenames, topics, commands containing confidential values, and raw backend logs.", "", "## Anonymized input inventory", "", "Reported sizes are independently jittered once by a uniform random factor in [-3%, +3%] and rounded to KiB. This is a publication redaction, not an exact measurement.", "", "| Input | Duration (nearest minute) | Reported size (KiB, jittered) |", "| --- | ---: | ---: |"]
    lines += [f"| {row['video']} | {row['duration_minutes']} | {row['reported_kib']} |" for row in videos]
    lines += ["", "## Per-chunk measurements", "", "| Backend | Chunk | Context | Wall seconds | Peak apparent RSS (MiB) | Exit |", "| --- | ---: | ---: | ---: | ---: | ---: |"]
    lines += [f"| {row['backend']} | {row['chunk']:02d} | {row['context']} | {row['wall_seconds']:.3f} | {row['peak_rss_mib']:.1f} | {row['exit_code']} |" for row in records]
    lines += ["", "Peak RSS is sampled every 250 ms from the launched backend and descendants; Ollama additionally samples its local server. It is an apparent process-level maximum, not a device-memory profiler.", "", "TurboFieldfare uses temperature 1.0, top-k 64, top-p 0.95, and repetition penalty 1.0. Ollama uses its local native defaults. This is a real-user-path comparison, not sampler-normalized evidence."]
    destination.write_text("\n".join(lines) + "\n", encoding="utf-8")
    print(f"public report rows={len(records)} videos={len(videos)}")


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--config", type=Path, default=DEFAULT_CONFIG)
    sub = parser.add_subparsers(dest="action", required=True)
    sub.add_parser("preflight")
    sub.add_parser("prepare")
    for name in ("warmup", "measure"):
        action = sub.add_parser(name)
        action.add_argument("backend", choices=["turbofieldfare", "ollama"])
        action.add_argument("chunk", type=int)
        action.add_argument("--context", type=int, default=4096)
    public = sub.add_parser("public-report")
    public.add_argument("--output", type=Path, required=True)
    args = parser.parse_args(argv)
    config, root = load_config(args.config), None
    root = root_for(config)
    if args.action == "preflight":
        free = free_memory()
        course = Path(config["course_root"])
        has_transcripts = course.is_dir() and any(course.rglob("*_video_*.md"))
        ollama_model = subprocess.run(["ollama", "show", "gemma4:26b"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False, timeout=30).returncode == 0 if shutil.which("ollama") else False
        good = platform.mac_ver()[0].startswith("26.") and has_transcripts and (Path(config["model_dir"]) / "manifest.json").is_file() and (ROOT / ".build" / "release" / "TurboFieldfareCLI").is_file() and ollama_model and free is not None and free >= 20 and not model_owner_present() and not ollama_loaded()
        if not good:
            raise RuntimeError("preflight checks failed")
        print(f"preflight ok memory_free_percent={free}")
    elif args.action == "prepare":
        prepare(config, root)
    elif args.action in {"warmup", "measure"}:
        run_one(config, root, args.backend, args.chunk, args.context, args.action == "warmup")
    else:
        report(root, args.output)
    return 0


if __name__ == "__main__":
    try:
        raise SystemExit(main())
    except (RuntimeError, ValueError) as error:
        print(f"error: {error}", file=sys.stderr)
        raise SystemExit(2)

process_chunks_ollama.py

#!/usr/bin/env python3
"""Opt-in Ollama chunk processor using Ollama's native default settings."""
from __future__ import annotations

import argparse
import json
from pathlib import Path
import urllib.request


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--base-url", default="http://127.0.0.1:11434")
    parser.add_argument("--model", default="gemma4:26b")
    args = parser.parse_args(argv)
    if args.input.is_dir():
        inputs = sorted(path for path in args.input.glob("benchmark*.md") if path.is_file())
        output_paths = [args.output / source.name for source in inputs]
    elif args.input.is_file() and args.input.suffix == ".md":
        inputs = [args.input]
        # A single chunk can be used by benchmark_run's per-chunk command
        # template.  Treat an existing directory, or a path without a Markdown
        # filename, as an output directory; otherwise it is the exact result
        # file requested by the caller.
        if args.output.is_dir() or args.output.suffix != ".md":
            output_paths = [args.output / args.input.name]
        else:
            output_paths = [args.output]
    else:
        parser.error("--input must be a Markdown file or a directory containing benchmark*.md files")

    endpoint = args.base_url.rstrip("/") + "/api/generate"
    for source, destination in zip(inputs, output_paths):
        payload = json.dumps({"model": args.model, "prompt": source.read_text(encoding="utf-8"), "stream": True}).encode()
        request = urllib.request.Request(endpoint, data=payload, headers={"Content-Type": "application/json"}, method="POST")
        pieces: list[str] = []
        with urllib.request.urlopen(request) as response:
            for line in response:
                if line.strip():
                    item = json.loads(line)
                    pieces.append(item.get("response", ""))
        destination.parent.mkdir(parents=True, exist_ok=True)
        destination.write_text("".join(pieces), encoding="utf-8")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

process_chunks_turbofieldfare.py

#!/usr/bin/env python3
"""Run one TurboFieldfare CLI summary request for a benchmark chunk.

The adapter deliberately relies on the CLI's Gemma summary-profile defaults:
4,096-token context, temperature 1.0, top-p 0.95, top-k 64, repetition
penalty 1.1, and the remaining-context output limit.  Supplying only the
model and messages file keeps that profile in one public place: the CLI.
"""
from __future__ import annotations

import argparse
import json
from pathlib import Path
import subprocess
import sys
import tempfile


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--input", required=True, type=Path)
    parser.add_argument("--output", required=True, type=Path)
    parser.add_argument("--cli", required=True, type=Path)
    parser.add_argument("--model", required=True)
    parser.add_argument("--prompt", required=True)
    parser.add_argument("--max-context", type=int)
    parser.add_argument("--temperature", type=float)
    parser.add_argument("--top-k", type=int)
    parser.add_argument("--top-p", type=float)
    parser.add_argument("--repetition-penalty", type=float)
    args = parser.parse_args(argv)

    source = args.input.read_text(encoding="utf-8")
    messages = [{"role": "user", "content": args.prompt + "\n\n" + source}]

    # The message file contains source text, so never place it in the output
    # tree where benchmark artifacts are retained.  The context manager removes
    # it on both successful and failed CLI invocations.
    with tempfile.NamedTemporaryFile(
        mode="w", encoding="utf-8", suffix=".json", delete=True
    ) as message_file:
        json.dump(messages, message_file)
        message_file.flush()
        command = [str(args.cli), "--model", args.model, "--messages-file", message_file.name]
        for flag, value in [
            ("--max-context", args.max_context),
            ("--temperature", args.temperature),
            ("--top-k", args.top_k),
            ("--top-p", args.top_p),
            ("--repetition-penalty", args.repetition_penalty),
        ]:
            if value is not None:
                command.extend([flag, str(value)])
        completed = subprocess.run(
            command,
            text=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            check=False,
        )

    args.output.parent.mkdir(parents=True, exist_ok=True)
    args.output.write_text(completed.stdout, encoding="utf-8")
    if completed.stderr:
        sys.stderr.write(completed.stderr)
    return completed.returncode


if __name__ == "__main__":
    raise SystemExit(main())