interactive-story/backend/app/context/encoding.py

"""`cl100k_base` without a first-use download.

Upstream called `tiktoken.get_encoding("cl100k_base")`, which fetches the BPE
table from `openaipublic.blob.core.windows.net` the first time it is used and
caches it under the system temp directory. That download is invisible on a
developer machine that has already made it once, and fatal on a machine with
outbound Internet blocked: the context builder counts tokens on *every* turn,
so the first story turn died with a `ConnectionError` instead of narrating
(Phase 0B offline report; acceptance tests A01 and H11).

The table is vendored beside this module and the `Encoding` is constructed from
it directly, so no code path inside the tokenizer can reach the network — not a
cache that happens to be warm, and not an environment variable a deployment
could forget to set.

The vendored file's SHA-256 is verified against the digest `tiktoken` itself
pins for that URL. A truncated checkout or a substituted table then fails
loudly, rather than silently changing every token count the context budget is
computed from.
"""

import base64
import functools
import hashlib
from pathlib import Path

import tiktoken

ENCODING_NAME = "cl100k_base"

#: Where the table came from, recorded so the vendored copy can be re-derived.
SOURCE_URL = "https://openaipublic.blob.core.windows.net/encodings/cl100k_base.tiktoken"

#: The digest `tiktoken_ext.openai_public.cl100k_base()` pins for SOURCE_URL.
BPE_SHA256 = "223921b76ee99bde995b7ff738513eef100fb51d18c93597a113bcffe865b2a7"

BPE_PATH = Path(__file__).resolve().parent / "vendor" / "cl100k_base.tiktoken"

# Both copied from `tiktoken_ext.openai_public.cl100k_base()`. They are part of
# the encoding's identity: the same merge table with a different pattern is a
# different tokenizer, so they are pinned here rather than imported, and the
# round-trip test asserts this build agrees with tiktoken's own.
_PAT_STR = r"""'(?i:[sdmt]|ll|ve|re)|[^\r\n\p{L}\p{N}]?+\p{L}++|\p{N}{1,3}+| ?[^\s\p{L}\p{N}]++[\r\n]*+|\s++$|\s*[\r\n]|\s+(?!\S)|\s"""
_SPECIAL_TOKENS = {
    "<|endoftext|>": 100257,
    "<|fim_prefix|>": 100258,
    "<|fim_middle|>": 100259,
    "<|fim_suffix|>": 100260,
    "<|endofprompt|>": 100276,
}


def _mergeable_ranks() -> dict[bytes, int]:
    try:
        data = BPE_PATH.read_bytes()
    except OSError as exc:  # pragma: no cover - packaging fault, not a run fault
        raise RuntimeError(
            f"The vendored {ENCODING_NAME} table is missing at {BPE_PATH}. "
            f"Re-download it from {SOURCE_URL} (SHA-256 {BPE_SHA256})."
        ) from exc

    digest = hashlib.sha256(data).hexdigest()
    if digest != BPE_SHA256:
        raise RuntimeError(
            f"The vendored {ENCODING_NAME} table at {BPE_PATH} has SHA-256 "
            f"{digest}, expected {BPE_SHA256}."
        )

    # Same parse as tiktoken.load.load_tiktoken_bpe, minus its fetch/cache step.
    ranks: dict[bytes, int] = {}
    for line in data.splitlines():
        if not line:
            continue
        try:
            token, rank = line.split()
            ranks[base64.b64decode(token)] = int(rank)
        except Exception as exc:
            raise ValueError(f"Error parsing line {line!r} in {BPE_PATH}") from exc
    return ranks


@functools.lru_cache(maxsize=1)
def get_encoding() -> tiktoken.Encoding:
    """The `cl100k_base` encoding, built from the vendored table."""
    return tiktoken.Encoding(
        name=ENCODING_NAME,
        pat_str=_PAT_STR,
        mergeable_ranks=_mergeable_ranks(),
        special_tokens=_SPECIAL_TOKENS,
    )