OpenAI API "metadata" - not just keys, now your 30k blob storage (Python code)

This weekend with some free time to see a silly idea through to its ultimate realization and implementation, I explore a novel method to leverage the OpenAI Responses API’s metadata field as a compact, reliable, and performant server-side storage mechanism, despite its inherent limitations (16 keys of up to 512 Unicode characters each). Aimed at developers seeking efficient and safe methods for storing and retrieving small but critical data, the solution described introduces two optimized encoding approaches to go beyond simply keys and no API-based queries to be run.

This text is also intended to be a standalone test payload: pasted as a single demo TEST string into the code that measure storage and round-trip fidelity of the encoder itself - just in case you need a body of text to evaluate. So I let AI expand and expand on what’s going on here.

The code provides heuristics to select the optimal encoding strategy automatically, ensuring maximum efficiency. The implementation includes robust safeguards, clear rejection semantics, and straightforward APIs to integrate easily with OpenAI’s endpoints—such as conversations, vector stores, fine-tuning jobs, and evaluation runs. If your workflow requires reliable, deterministic metadata encoding with exact round-trip fidelity, minimal latency, and compatibility across OpenAI APIs, and are chagrined by the small storage and lack of a query endpoint for this metadata, this document and its accompanying Python solution are directly applicable.

Easily realized: 20k-40k text storage, 9900 tokens, more when paginated across on-demand entities. This entire text and encoder/decoder code can be stored in one metadata.

Python 3.11+, employs tiktoken: pip install tiktoken for best results
Included: a demo storage of long string to a temporary “conversations” item.
Not included: a read/write update cycle for database as the stringified blob in metadata.


This document is a practical exploration of the OpenAI Responses API metadata field and its unintended potential as a permanent, server-side, low-latency storage substrate. It is written for product developers who want to stretch the limits of the platform safely and predictably, while keeping a clear lens on operational constraints, ergonomics, and tradeoffs.

The core challenge starts with a deliberately small capability: metadata is a key-value map attached to a server-side object (a conversation, a vector store file, a fine-tuning job, an eval run, or a batch). The map supports at most 16 user-defined keys, with a 64-character limit on each key name and a 512- character limit on each value string. The API does not provide an endpoint for listing and filtering these objects by metadata; any discovery must be done by cross-referencing IDs the application already possesses. The surface appears modest: it is obviously not a general purpose data store, and it is not meant to be queried across the fleet.

Yet the constraint hides a surprising amount of possibility. The first step is recognizing that the value limit is a character count, not a byte count. This means the server accepts and returns UTF-8 strings up to 512 characters long per key. Because Unicode is richer than a simple byte channel, and because the API preserves exact content reliably, those 512 characters can carry more than 512 bytes of information. If one chooses code points carefully, the storage can map to a much wider alphabet of symbols than a legacy byte-encoding would suggest, while remaining fully compatible with JSON transports and HTTP. The size of the usable UTF-8 code point alphabet (here, N=1,112,032): all Unicode code points except surrogates U+D800–U+DFFF and U+FDD0–U+FDEF)

Two distinct approaches naturally follow from that insight. The first approach one might consider is a tokenstream encoder that maps model token IDs directly into Unicode code points, using a simple reversible offset and skipping invalid surrogates. In this scheme, a single character encodes exactly one token. For many natural- language strings, this one-to-one mapping is exceptionally efficient. It avoids bit packing, avoids framing, and only requires a start marker and an end marker to safely delimit the stream. Because the server preserves code points as characters, the round trip is exact. This approach shines on plain text, JSON, and semi-structured prompts where the tokenizer performs well. But is not optimal.

The second approach, a general-purpose compressor, shares a common final stage with the tokenstream method. Specifically, we compress with LZMA (no container, no checksum to keep the container small) with filter parameters optimized with trials of large set of text lengths and iterative evaluations of pre-encoders. In application use, both UTF-8 and UTF-16 are trialed. But then the bytes that are output might naively be placed with direct character mapping, or a translation of for example 5 bytes to 40 bits across two unicode characters. Close, but not optimal.

Both pipelines produce a simple stream of integers as their intermediate output. For LZMA compression, this is a stream of bytes (integers 0-255); for a tokenizer, it is a stream of token IDs (integers 0 to ~200,000). This common format allows for a single, highly efficient packing algorithm for all data types of varying base.

The key is to treat the entire stream of integers as the digits of a single, gigantic number. The problem then becomes a straightforward change of base: converting this large number from its original base (e.g., base 256 for bytes) into our target base of nearly 1.1 million valid Unicode characters. Mathematically, if we have k integers with a maximum value of M, we solve for the minimum number of characters D such that N^D >= (M+1)^k, where N is the size of our Unicode character set. This transformation is provably optimal, ensuring that the sequence of integers is stored in the absolute minimum number of characters possible.

Concretely: we convert the original stream from its native base (e.g., base 256 for bytes, or the tokenizer’s base for token IDs) into a target base of N = 1,112,032 valid Unicode scalar values (all code points except UTF-16 surrogates and U+FDD0–U+FDEF). Each stored character therefore carries log₂(N) ≈ 20.0848 bits of payload, approaching the theoretical maximum for a JSON-safe, Unicode-preserving wire.

Two upstream integer sources are supported. First, LZMA (RAW) compression (no container, no header, no checksum) over the input text, after encoding either as UTF-16LE or UTF-8 (filters fine-tuned for UTF-16 as an ultimate pattern for redundancy removal). Second, tokenizer integer streams via tiktoken (e.g., o200k_base, cl100k_base). Both produce a finite list of integers and a known maximum value per integer; the unified base-N packer then guarantees near-optimal character usage across both cases.

While a tokenizer with a larger vocabulary might seem an advantage, its storage space for symbols also increases:

>>> encoding = tiktoken.get_encoding("o200k_base")
>>> math.log(encoding.max_token_value, 2) = 17.609770311147926

>>> math.log(100276, 2) = 16.613616827902597 (cl100k_base)
>>> math.log(50280, 2) = 15.61769702930787 (50k encoders, eventually ruled-out)

Thus both are trialed automatically, with cl100k being the winner most often because of its storage advantage - 9,340 tokens of o200k grows to 9900.

A single-character signature prefixes every stored payload. This signature encodes: (1) the stream type (2 bits for {lzma, lzma8, o200k_base, cl100k_base}) and (2) the exact integer-count of the original stream (18 bits, length-inverted to avoid zero payload ambiguity). The signature occupies a reserved high-rank window of the Unicode carrier, and likely encoders are ranked so the least used is in the lowest quartile of the space, so that accidental collisions with ordinary text are vanishingly unlikely. On decode, this yields strong, constant-time rejection if a payload isn’t ours.

The encoder trials multiple candidates and automatically picks the smallest result. Concretely, it tries: lzma (UTF-16LE→LZMA-RAW), lzma8 (UTF-8→LZMA-RAW), o200k_base, and cl100k_base. Each trial is converted into base N, prefixed with a signature, and measured. The winner is the wire. This heuristic avoids premature optimization while consistently selecting the best option for the specific input distribution.

The implication is that the API’s 16×512-char limit is not an absolute barrier. On natural text workloads where the tokenizer is competent, a pure tokenstream can be smaller than compression, because there is no framing and the vocabulary is purpose-built for common morphemes. On highly redundant or binary-like data, lzma tends to win, because it exploits repeating structures that tokenization cannot compress without more characters. The practical result is that a developer can implement a simple heuristic that picks the better of two encodings for each payload, producing near-optimal size in both extremes. That is exactly what is done here.

A complicating factor is discoverability. The API does not provide a server-side method to search across many conversations or files by metadata filter. An application must already know which object to fetch. This pushes the system toward per-object stateful workflows, where metadata is appended or updated on a specific conversation or file as part of a larger transactional path. It also suggests segmenting stored data across multiple logical stores by prefixing keys (for example, audit01..auditNN, cache01..cacheNN, and notes01.. notesNN), so that independent application subsystems can use metadata safely in parallel without clobbering each other. The methods in code are parameterized to allow segmenting particular maximum key spaces to one data store, also not an absolute reservation, as unused named keys are actively freed.

Because each key name is capped at 64 characters, careful naming helps. A good pattern is a short datastore prefix and two-digit suffixes, such as datastorage01 through datastorage16. This keeps the wire predictable and allows decoders to find only the relevant keys, ignoring ancillary metadata like titles and labels. The encoders in this package follow that convention, so the layout remains readable and self-cleaning: values beyond the last used key are set to null to orphan-delete any prior content when data shrinks. The opportunity for more storage in the keys themselves is not exploited, to ensure query quality.

The technique for achieving this relies on strategic placement within the vast Unicode character space. A small, reserved block of the highest-value code points is set aside exclusively for signatures. The type and length information is mathematically combined into a single number, which is then mapped to one of these reserved characters. This placement is deliberate and crucial for robustness. Common text, especially ASCII, occupies the lowest end of the Unicode range.

In a UTF-8 context, these high-value signature characters will occupy three or four bytes, unlike the single byte used for ASCII. This is a conscious design trade-off: a slightly larger signature character buys us an unambiguous, data-rich header that guarantees the integrity of the decoding process for the entire payload that follows.

The signature has been condensed into a single, information-rich character at the start of the stream. This character serves a dual purpose. First, it acts as a unique identifier, telling the decoder precisely which method was used: LZMA with UTF-16, LZMA with UTF-8, or one of the tokenizers. Second, it encodes the exact length of the original integer stream—the number of bytes for LZMA or the number of tokens for a tokenizer. This gives the decoder a powerful strong-reject capability: if one of these specific signature characters is not present at the very beginning of the first value, it bails immediately. This allows downstream logic to safely attempt another decoder without guesswork or ambiguous states.

The decision function that chooses how to write a given payload compares the total characters that each encoding would consume. The winning encoder writes the metadata in one POST. The storage report consumable includes a count of characters, keys used, and lightweight codec-specific data. This model lets developers keep a single entry point for writing: most callers do not need to care which codec was used. Reads are also a single call: the decoder attempts XZ first for fast signature rejection, then tries tokenstream. This reduces code complexity for application authors.

Applications of a string-based blob are flexible. JSON can be parsed within the round-tripped text to yield structured state: transient caches, evaluated prompts and chain-of-thoughts (if permitted by your governance model), human readable traces, audit invariants, or lightweight configuration knobs bound to a conversation. Because many endpoints support metadata, the same string can accompany vector store files (for lineage annotations), fine-tune jobs (for experiment labels and hyperparameters), eval runs (for test corpus identity), and batch jobs (for submission controls and analytics). The lack of a global metadata filter is mitigated by storing backpointers (IDs) within these blobs, so related objects can be fetched by ID and then cross-validated locally.

Parallel logical stores can be created by choosing multiple datastore prefixes. Each prefix has up to 16 keys, but in aggregate you must still honor the endpoint’s total cap of 16 user-defined metadata keys. This means multiplexing is primarily useful when the caller owns the whole 16-key budget. Many applications will dedicate all 16 to one store. In that case, the internal string can contain multiple virtual sections in JSON, such as a map of tables, and the upstream code can sustain partial updates by reading, modifying the JSON, and writing it back through the same automatic encoder selection.

Updating such a blob needs a strategy. Since the values are stored across sequential keys, updates overwrite the same key names idempotently; there are no orphan anomalies if the writer always nulls out unused higher-index keys. When the content is JSON, the pattern is: read metadata, decode to a string, json.loads to a dict, apply your changes, dump to a minified string, then write. For binary data wrapped as base64 or similar, compressing first through lzma is generally superior. Concurrency should be handled at the application level, with optimistic retries: fetch, decode, merge, write, and compare lengths and timestamps as needed.

Integrity and durability considerations are pragmatic. The platform’s metadata is stable and survives across object lifecycle operations. The auto-decoder’s strong rejection plus LZMA integrity gives high confidence in reads. If the application warrants it, you can embed a hash in the JSON content, or prepend a content signature to the data before compression. Encryption can be layered on top, as long as the ciphertext remains byte-oriented; lzma will handle it, though compression becomes ineffective, so in that scenario tokenstream may not win either. But the robust space use of the encoder still delivers 20566 bytes of storage layer for even uncompressible loads.

Size planning depends on workload. Tokenstream handles roughly one token per semantic unit and compresses them to under that as character count. For English prose and JSON, this often gives 3–4 characters per word. In contrast, lzma supports about 20 KB of compressed bytes, which on redundant inputs translates into very large original strings. For pathological random Unicode (e.g., CJK sequences with low redundancy), tokenization inflates while compression remains bounded; lzma will outperform tokenstream decisively. The auto writer captures this behavior empirically, avoiding premature optimization.

A final note on benchmarking: always measure with your own data. Real-world payloads vary wildly. Logs compress well; random user-generated strings do not. Templates with placeholders are very compressible; literal emojis are not. For each workload, the automatic path provides a simple, reliable choice that you can validate with end-to-end round trips. Track stored character counts, keys used, and error rates. Set conservative maximums and surface clear error messages to operators when a payload is too large or malformed. You can set hard maximums, or be optimistic, with the library here raising its own exception for going over an alloted datastore.

To conclude, the 16-key metadata limit is not a brick wall. With enough idle coding time, it is a compact, deterministic workspace, and with careful encoding it becomes a practical mini store for conversation-local state, artifact manifests, evaluation outcomes, and small domain datasets. The two encoders in this package, tokenstream and lzma cover the main spectrum, from text that tokenizes well to data that compresses well. The auto writer selects the winner transparently.

What will you attach today?


Usage signatures:

def encode_metadata(
    text: str,
    *,
    max_keys: int = 16,
    datastore_name: str = "datastorage",
) -> tuple[dict[str, str | None], dict]:
    """
    Finds the most compact representation for text data by trialing multiple
    encoding methods and handles trial failures gracefully.
def decode_metadata(
    metadata: dict[str, str],
    datastore_name: str = "datastorage",
) -> str:
    """
    Decodes a dictionary of metadata keys packed by `encode_metadata`. It automatically
    detects the encoding method and applies the correct decoding process.
    """

What you get when you run the “report” commented method in main():


-- 1. Conversations: Storing 83 characters of:
This language is 83 characters, a twenty token string, yet stores to only 17 cha..
-- 1a. Storage Report:
{
  "input_chars": 83,
  "stored_chars": 18,
  "keys_used": 1,
  "bytes_per_key": {
    "datastorage01": 18
  },
  "codec": "cl100k_base",
  "trials": [
    {
      "type": "lzma",
      "stream_len": 107,
      "chars": 44
    },
    {
      "type": "lzma8",
      "stream_len": 87,
      "chars": 36
    },
    {
      "type": "o200k_base",
      "stream_len": 20,
      "chars": 19
    },
    {
      "type": "cl100k_base",
      "stream_len": 20,
      "chars": 18
    }
  ],
  "info": {
    "stream_type": "tokens",
    "stream_len": 20,
    "payload_chars": 17
  }
}
-- 2. Retrieving
-- 3. Round-trip OK. Codec: cl100k_base
import logging

logger = logging.getLogger(__name__)
import sys
import base64
import math
import lzma
from typing import Any

try:
    import httpx  # required for demo, RESTful calls
except Exception as e:
    raise ImportError("Missing dependency: install with `pip install httpx`") from e

try:
    import tiktoken
except Exception as e:
    print("Optional dependency missing: install with `pip install tiktoken`")


# HTTP helpers and Conversations lifecycle
# - get_api_key_headers(): returns Bearer auth headers from OPENAI_API_KEY
# - create_conversation(): POST /v1/conversations with an initial developer message
# - delete_conversation(): DELETE /v1/conversations/{id}, tolerant of 2xx/400 and logs details

def get_api_key_headers(
    api_key: str | None = None,
    printing: bool = False,
    dotenv_override: bool = True,
    env_path: str | None = None,
) -> dict[str, str]:
    """
    Returns OpenAI API auth headers.

    Behavior:
      - A passed `api_key` bypasses all env variable methods
      - Attempts to load variables from a .env file (if python-dotenv is available).
      - By default, .env VALUES OVERRIDE existing os.environ entries (dotenv_override=True).
      - You can point to a specific file via env_path; otherwise we use find_dotenv(usecwd=True).
      - Each .env path is loaded at most once per process (cached).

    Notes:
      - This updates process environment variables; in multi-account, single-process scenarios,
        prefer passing explicit headers/keys per call rather than relying on global env.
      - Returns dict, thus not duplicate header keys.

    Usage: To avoid cross-call bleed, pass a *copy* before mutation is possible, e.g.:
    api_call(url, headers = {**get_api_key_headers(), "OpenAI-Beta": "responses=v99"}, ...)
    """
    if api_key:
        return {"Authorization": f"Bearer {api_key}"}

    # One-time cache per path
    if not hasattr(get_api_key_headers, "_dotenv_loaded_paths"):
        get_api_key_headers._dotenv_loaded_paths = set()

    # Best-effort .env loading (optional dependency)
    try:
        if env_path is None:
            from dotenv import load_dotenv, find_dotenv  # lazy import

            dotenv_path = find_dotenv(usecwd=True)
        else:
            from dotenv import load_dotenv  # lazy import

            dotenv_path = env_path

        if dotenv_path and dotenv_path not in get_api_key_headers._dotenv_loaded_paths:
            load_dotenv(dotenv_path=dotenv_path, override=dotenv_override)
            get_api_key_headers._dotenv_loaded_paths.add(dotenv_path)
    except Exception:
        # Missing python-dotenv or any load failure is a no-op.
        pass

    import os

    api_key = os.environ.get("OPENAI_API_KEY")
    if not api_key:
        raise ValueError(
            "ERROR: Set the OPENAI_API_KEY environment variable (or in your .env)."
        )
    org_id = os.environ.get("OPENAI_ORG_ID")
    project_id = os.environ.get("OPENAI_PROJECT_ID")

    if printing:
        print(f"Using OPENAI_API_KEY {api_key[:10]}...{api_key[-4:]}")
        print(f"Using optional OPENAI_ORG_ID {org_id}")
        print(f"Using optional OPENAI_PROJECT_ID {project_id}")

    headers: dict[str, str] = {"Authorization": f"Bearer {api_key}"}
    if org_id:
        headers["OpenAI-Organization"] = org_id
    if project_id:
        headers["OpenAI-Project"] = project_id
    return headers

def create_conversation(placeholder_count: int | None = 0) -> str:
    """
    Create a conversation with no messages, and up to 16 placeholder metadata keys.
    placeholder_count: number of placeholder fields to include (0–16) to consume from max 16.
    """
    # normalize and validate placeholder_count
    count = placeholder_count or 0
    if not isinstance(count, int) or not (0 <= count <= 16):
        raise ValueError(
            f"placeholder_count must be an int between 0 and 16; got {placeholder_count!r}"
        )

    # build payload with numbered placeholder keys
    payload: dict[str, object] = {"metadata": {}}
    for i in range(1, count + 1):
        key = f"placeholder{i:02d}"
        payload["metadata"][key] = "placeholder data"
    import httpx

    try:
        with httpx.Client(timeout=20) as client:
            response = client.post(
                "https://api.openai.com/v1/conversations",
                headers=get_api_key_headers(),
                json=payload,
            )

        # Check for HTTP error status codes
        try:
            response.raise_for_status()
        except httpx.HTTPStatusError as exc:
            # Attempt to retrieve error details from the response
            resp = exc.response
            try:
                body_content = resp.json()
            except ValueError:
                body_content = resp.text or "<no body>"

            # Format for logging
            if isinstance(body_content, (dict, list)):
                formatted_body = json.dumps(body_content, indent=2)
            else:
                formatted_body = str(body_content)

            logger.error(
                "Failed to create conversation: status %s\nResponse body:\n%s",
                resp.status_code,
                formatted_body,
            )
            # Terminal failure
            raise

        # Extract and log the new conversation ID
        data = response.json()
        conversation_id = data.get("id")
        if not conversation_id:
            logger.error(
                "Unexpected response creating conversation: missing 'id' field\nResponse JSON:\n%s",
                json.dumps(data, indent=2),
            )
            raise RuntimeError("Invalid API response: no conversation ID returned")

        logger.info("Conversation created: %s", conversation_id)
        return conversation_id

    except httpx.RequestError as exc:
        # Network or connection-level errors
        logger.error("Request error creating conversation: %s", exc)
        raise

    except Exception:
        # Catch-all for any other failures
        logger.exception("Unexpected error in create_conversation()")
        raise

def delete_conversation(conversation_id: str) -> None:
    """
    Delete the conversation so the demo doesn’t leave stray server objects.
    On 2xx or 400, it's treated as success. Errors are logged to stderr.
    """
    import sys, httpx

    try:
        with httpx.Client(timeout=20) as client:
            response = client.delete(
                f"https://api.openai.com/v1/conversations/{conversation_id}",
                headers=get_api_key_headers(),
            )
        try:
            response.raise_for_status()
        except httpx.HTTPStatusError as exc:
            import json

            resp = exc.response
            # Attempt to parse JSON, fall back to raw text
            try:
                body_content = resp.json()
            except ValueError:
                body_content = resp.text

            # Format body for logging/printing
            if isinstance(body_content, (dict, list)):
                formatted_body = json.dumps(body_content, indent=2)
            else:
                formatted_body = str(body_content)

            if resp.status_code == 400:
                logger.warning(
                    "Conversation wasn't deleted or didn't exist: %s\nResponse body:\n%s",
                    conversation_id,
                    formatted_body,
                )
            else:
                print(
                    f"Error deleting conversation {conversation_id}: "
                    f"status {resp.status_code}\nResponse body:\n{formatted_body}"
                )
                logger.error(
                    "HTTP status error deleting conversation %s: %s",
                    conversation_id,
                    formatted_body,
                )
                raise
        else:
            logger.info("Conversation deleted: %s", conversation_id)
    except Exception as exc:
        print(f"[warn] Couldn’t delete {conversation_id}: {exc}", file=sys.stderr)
# ######################## END EXTRA API METHODS ###################### #

# -- encoder/decoder common helpers, constants

class MetadataCapacityError(Exception):
    """Raised when the encoded payload exceeds the available metadata capacity."""


# Unicode Carrier Space (for mapping integers to valid, non-surrogate code points)
SURROGATE_START = 0xD800
SURROGATE_END = 0xDFFF
MAX_CP = 0x10FFFF
# Non-character range U+FDD0..U+FDEF is also excluded
MID_LEN = (0xFDCF - 0xE000 + 1)
# Total size of the carrier space
N = (SURROGATE_START) + MID_LEN + (MAX_CP - 0xFDF0 + 1)
LOG2N = math.log2(N)

# Metadata Storage
METADATA_CHUNK_SIZE = 512
METADATA_CLEANUP_KEYS = 16

LZMA_FILTERS = [
    {
        "id": 33,           # lzma.FILTER_LZMA2
        "preset": 6,
        "dict_size": 65536, # 2**16
        "lc": 2, "lp": 1, "pb": 1,
        "mode": 2,          # lzma.MODE_NORMAL
        "nice_len": 273,
        "mf": 20,           # lzma.MF_BT4
        "depth": 65536,     # 2**16
    },
]

# --- Symmetrical Helper Functions ---


# Rank <-> Code Point (Defines the character set used for data)
def _rank_to_cp(r: int) -> int:
    if not (0 <= r < N):
        raise ValueError("rank out of range")
    if r < SURROGATE_START:
        return r
    r -= SURROGATE_START
    if r < MID_LEN:
        return 0xE000 + r
    r -= MID_LEN
    return 0xFDF0 + r


def _cp_to_rank(cp: int) -> int:
    if not (
        0 <= cp <= MAX_CP
        and not (SURROGATE_START <= cp <= SURROGATE_END)
        and not (0xFDD0 <= cp <= 0xFDEF)
    ):
        raise ValueError("invalid code point for carrier")
    if cp < SURROGATE_START:
        return cp
    if cp <= 0xFDCF:
        return SURROGATE_START + (cp - 0xE000)
    return SURROGATE_START + MID_LEN + (cp - 0xFDF0)

# --- Symmetrical Signature Codec ---

# 1. Type ID Mapping
# Defines the 2-bit integer assigned to each encoding method.
TYPE_MAP = {"cl100k_base": 3, "o200k_base": 2, "lzma": 1, "lzma8": 0}
REVERSE_TYPE_MAP = {v: k for k, v in TYPE_MAP.items()}

# 2. Bit Layout for the 20-bit Signature Payload
# The signature packs a type ID and a length/count into a single integer.
SIGNATURE_TYPE_ID_BITS = 2
SIGNATURE_LENGTH_BITS = 18
SIGNATURE_TYPE_ID_SHIFT = (
    SIGNATURE_LENGTH_BITS  # Type ID is stored in the most significant bits
)
SIGNATURE_LENGTH_MASK = (1 << SIGNATURE_LENGTH_BITS) - 1

# 3. Rank Space Allocation for Signatures
# Signatures are mapped to the highest-valued code points in our carrier space
# to ensure they are distinct from any possible data character.
SIGNATURE_PAYLOAD_BITS = SIGNATURE_TYPE_ID_BITS + SIGNATURE_LENGTH_BITS
SIGNATURE_PAYLOAD_SIZE = 1 << SIGNATURE_PAYLOAD_BITS
# The following constant 'N' represents the total size of the valid Unicode character
# space we use for encoding. It is required to calculate the signature's location.
# N = 1,112,032 (all scalars excluding surrogates and U+FDD0..U+FDEF)
N = (0xD800) + (0xFDCF - 0xE000 + 1) + (0x10FFFF - 0xFDF0 + 1)
SIGNATURE_BASE_RANK = N - SIGNATURE_PAYLOAD_SIZE

# 4. Safety Limit
# An arbitrary high number to prevent decoding impossibly large payloads.
MAX_CAPACITY_BITS = 164514  # Corresponds to (2^18 - 1), the max length value

# --- Symmetrical Signature Functions ---
# These functions depend on the lower-level Unicode carrier helpers:
# _rank_to_cp(rank: int) -> int
# _cp_to_rank(codepoint: int) -> int

def encode_signature_char(type_id: int | str, length: int) -> str:
    """Encodes a type ID and a data length/count into a single signature character."""
    # Resolve the string type_id (e.g., "lzma") to its integer representation (e.g., 1)
    type_id_num = TYPE_MAP[type_id] if isinstance(type_id, str) else int(type_id)

    # Invert the length to ensure that a length of 0 does not result in a payload of 0,
    # which could be ambiguous. This uses the full range of the 18 bits.
    inverted_length = SIGNATURE_LENGTH_MASK - length

    # Combine the type and length into a single 20-bit integer payload
    signature_payload = (type_id_num << SIGNATURE_TYPE_ID_SHIFT) | inverted_length

    # Map the payload to the reserved signature space at the top of the rank range
    rank = SIGNATURE_BASE_RANK + signature_payload
    return chr(_rank_to_cp(rank))

def decode_signature(
    s: str,
    allowed_types: int | str | list[int | str]
) -> tuple[int, int]:
    """Decodes a signature character into a type ID and data length/count."""
    if not s:
        raise ValueError("Cannot decode signature from empty string")

    # Map the character's code point back to its rank in our carrier space
    rank = _cp_to_rank(ord(s[0]))
    if rank < SIGNATURE_BASE_RANK:
        raise ValueError("Character is not a valid signature (rank is too low)")

    # Extract the 20-bit payload integer from the rank
    signature_payload = rank - SIGNATURE_BASE_RANK

    # Deconstruct the payload into its constituent parts
    type_id_num = signature_payload >> SIGNATURE_TYPE_ID_SHIFT
    inverted_length = signature_payload & SIGNATURE_LENGTH_MASK

    # Restore the original length from its inverted form
    length = SIGNATURE_LENGTH_MASK - inverted_length

    # Validate the decoded type against the list of types the caller expects
    allowed_iter = [allowed_types] if isinstance(allowed_types, (int, str)) else allowed_types
    # This comprehension robustly handles a mix of strings and ints in allowed_types
    allowed_set: set[int] = {TYPE_MAP.get(x, x) for x in allowed_iter}

    if type_id_num not in allowed_set:
        decoded_type_name = REVERSE_TYPE_MAP.get(type_id_num, "unknown")
        raise ValueError(f"Signature type '{decoded_type_name}' is not in the allowed list")
    
    if length >= MAX_CAPACITY_BITS:
        raise ValueError(f"Signature length {length} exceeds capacity limit")

    return type_id_num, length

# --- Symmetrical Integer Stream Codec ---

def encode_integer_stream(
    integers: list[int],
    max_value: int
) -> list[int]:
    """
    Converts a stream of integers from a smaller base to a stream of integers
    in the larger Unicode carrier base (N).

    This is a change of base operation. It treats the input integer list as
    digits of a single large number in base `M+1` and converts that number to
    base `N`, returning the new digits (ranks). This is highly efficient for
    packing data.

    Args:
        integers: A list of integers to encode (e.g., tokens).
        max_value: The maximum possible value for an integer in the input stream (M).

    Returns:
        A list of integers (ranks) representing the packed data in base N.
    """
    if not integers:
        return []

    source_base = max_value + 1
    token_count = len(integers)

    # Calculate the number of output digits (ranks) required in base N.
    # This is derived from the inequality: N^num_output_ranks >= source_base^token_count
    bits_per_token = math.log2(source_base)
    num_output_ranks = math.ceil(token_count * bits_per_token / LOG2N)

    # --- Symmetrical Operation Part 1: Convert input digits to one large integer ---
    large_integer_representation = 0
    for token in integers:
        if not (0 <= token < source_base):
            raise ValueError(f"Token {token} out of range for base {source_base}")
        large_integer_representation = large_integer_representation * source_base + token

    # --- Symmetrical Operation Part 2: Convert large integer to output digits ---
    output_ranks = [0] * num_output_ranks
    temp_large_int = large_integer_representation
    for i in range(num_output_ranks - 1, -1, -1):
        temp_large_int, r = divmod(temp_large_int, N)
        output_ranks[i] = r
    
    if temp_large_int != 0:
        raise ValueError("Overflow during base conversion encoding")

    return output_ranks

def decode_integer_stream(
    ranks: list[int],
    max_value: int,
    count: int
) -> list[int]:
    """
    Converts a stream of integers (ranks) from the large Unicode carrier base (N)
    back to a stream of integers in a smaller base.

    This is the exact inverse of `encode_integer_stream`.

    Args:
        ranks: A list of integers (ranks) in base N.
        max_value: The maximum possible value for an integer in the output stream (M).
        count: The expected number of integers in the final output list.

    Returns:
        The original list of integers (e.g., tokens).
    """
    if not ranks and count == 0:
        return []
    if not ranks and count > 0:
        raise ValueError("Cannot decode non-zero token count from empty rank list")

    target_base = max_value + 1

    # --- Symmetrical Operation Part 1: Convert input ranks to one large integer ---
    large_integer_representation = 0
    for rank in ranks:
        large_integer_representation = large_integer_representation * N + rank

    # --- Symmetrical Operation Part 2: Convert large integer to output digits ---
    output_integers_rev = []
    temp_large_int = large_integer_representation
    for _ in range(count):
        temp_large_int, t = divmod(temp_large_int, target_base)
        output_integers_rev.append(t)

    if temp_large_int != 0 and count > 0:
        # This can happen if num_output_ranks was slightly overestimated due to ceil()
        # but the resulting number should still be valid. A true error is if the
        # decoded token count doesn't match.
        pass

    return list(reversed(output_integers_rev))

# --- Type-Specific Integer Stream Generators (Updated Section) ---

def _get_lzma_integer_stream(text: str, encoding: str) -> list[int]:
    """Compresses text with LZMA using a specified encoding and returns bytes."""
    import lzma
    raw_bytes = text.encode(encoding)
    compressed_bytes = lzma.compress(raw_bytes, format=lzma.FORMAT_RAW, filters=LZMA_FILTERS)
    return list(compressed_bytes)

def _get_tokenizer_integer_stream(text: str, enc_name: str) -> list[int]:
    """
    Tokenizes text and returns the result as a list of integers (tokens).
    Raises ValueError if text is too long for this trial to be practical.
    """
    if len(text) > 65_536:
        raise ValueError("Input text exceeds 65,536 character limit for tokenization trial")
    
    import tiktoken
    enc = tiktoken.get_encoding(enc_name)
    return enc.encode(text, allowed_special="all")

# --- Type-Specific Final Decoders (Updated Section) ---

def _decode_from_lzma_stream(integers: list[int], encoding: str) -> str:
    """Decodes an integer stream (bytes) from LZMA using a specified encoding."""
    import lzma
    try:
        compressed_bytes = bytes(integers)
        raw_bytes = lzma.decompress(compressed_bytes, format=lzma.FORMAT_RAW, filters=LZMA_FILTERS)
        return raw_bytes.decode(encoding)
    except lzma.LZMAError as e:
        raise ValueError(f"LZMA decompression failed: {e}") from e

def _decode_from_tokenizer_stream(integers: list[int], enc_name: str) -> str:
    """Decodes an integer stream (tokens) back to text."""
    import tiktoken
    try:
        enc = tiktoken.get_encoding(enc_name)
        return enc.decode(integers)
    except Exception as e:
        raise ValueError(f"tiktoken decode failed: {e}") from e

# --- Unified Main Encoder/Decoder ---

def encode_metadata(
    text: str,
    *,
    max_keys: int = 16,
    datastore_name: str = "datastorage",
) -> tuple[dict[str, str | None], dict]:
    """
    Finds the most compact representation for text data by trialing multiple
    encoding methods and handles trial failures gracefully.

    Returns
    -------
    (metadata, report)
      metadata : dict[str, str|None] ready to POST as "metadata"
      report   : {
                   "input_chars": int,
                   "stored_chars": int,
                   "keys_used": int,
                   "bytes_per_key": dict[str,int],
                   "codec": str,
                   "info": dict
                 }
    """
    # --- 1. Trial all encoding methods robustly ---
    candidates = []
    trial_configs = [
        {"type_id": "lzma", "encoding": "utf-16le"},
        {"type_id": "lzma8", "encoding": "utf-8"},
        {"type_id": "o200k_base"},
        {"type_id": "cl100k_base"},
    ]
    trials_list=[]
    for config in trial_configs:
        type_id = config["type_id"]
        try:
            if type_id.startswith("lzma"):
                max_value = 255
                integer_stream = _get_lzma_integer_stream(text, encoding=config["encoding"])
            else: # Tokenizer
                import tiktoken
                enc = tiktoken.get_encoding(type_id)
                max_value = enc.max_token_value
                integer_stream = _get_tokenizer_integer_stream(text, type_id)
            
            ranks = encode_integer_stream(integer_stream, max_value)
            total_chars = 1 + len(ranks)
            trials_list+=[{"type": type_id, "stream_len": len(integer_stream), "chars": total_chars}]
            
            candidates.append((total_chars, type_id, integer_stream, ranks))

        except Exception as e:
            logger.warning(f"Trial for '{type_id}' failed and will be skipped: {e}")

    # --- 2. Select the best candidate ---
    if not candidates:
        raise MetadataCapacityError("All encoding trials failed. Cannot process metadata.")

    best_chars, winner_type, winner_stream, winner_ranks = min(candidates)

    # --- 3. Capacity Check and Stream Construction ---
    capacity = max_keys * METADATA_CHUNK_SIZE
    if best_chars > capacity:
        raise MetadataCapacityError(
            f"Best encoding ({winner_type}) produced {best_chars} chars, exceeding capacity of {capacity}."
        )
    signature = encode_signature_char(type_id=winner_type, length=len(winner_stream))
    payload_chars = "".join(chr(_rank_to_cp(r)) for r in winner_ranks)
    wire_string = signature + payload_chars

    # --- 4. Chunk into the metadata dictionary ---
    def _key(i: int) -> str:
        return f"{datastore_name}{i:02d}"

    chunks = [wire_string[i:i + METADATA_CHUNK_SIZE] for i in range(0, len(wire_string), METADATA_CHUNK_SIZE)]
    meta: dict[str, str | None] = {}
    for i, chunk in enumerate(chunks, start=1):
        meta[_key(i)] = chunk
    for i in range(len(chunks) + 1, METADATA_CLEANUP_KEYS + 1):
        meta[_key(i)] = None

    # --- 5. Assemble Detailed Report and Perform Final Checks ---
    report = {
        "input_chars": len(text),
        "stored_chars": best_chars,
        "keys_used": len(chunks),
        "bytes_per_key": {_key(i): len(chunk) for i, chunk in enumerate(chunks, start=1)},
        "codec": winner_type,
        "trials": trials_list,
        "info": {
            "stream_type": "bytes" if winner_type.startswith("lzma") else "tokens",
            "stream_len": len(winner_stream),
            "payload_chars": len(winner_ranks),
        }
    }
    if winner_type.startswith("lzma"):
        report["info"]["pre_encoding"] = "utf-8" if winner_type == "lzma8" else "utf-16le"

    decoded_text = decode_metadata(meta, datastore_name)
    assert decoded_text == text, "Unified metadata ERROR: Round-trip assurance mismatch"

    return meta, report

def decode_metadata(
    metadata: dict[str, str],
    datastore_name: str = "datastorage",
) -> str:
    """
    Decodes a dictionary of metadata keys packed by `encode_metadata`. It automatically
    detects the encoding method and applies the correct decoding process.
    """
    import tiktoken

    # --- 1. Aggregate the dictionary into a character stream ---
    if not isinstance(metadata, dict):
        raise ValueError("Metadata must be a dictionary")
        
    chunk_list: list[tuple[int, str]] = []
    prefix_len = len(datastore_name)
    for k, v in metadata.items():
        if isinstance(v, str) and k.startswith(datastore_name):
            suffix = k[prefix_len:]
            if suffix.isdigit():
                chunk_list.append((int(suffix), v))

    if not chunk_list:
        raise ValueError(f"No data found for datastore '{datastore_name}'")

    chunk_list.sort(key=lambda kv: kv[0])
    wire_string = "".join(v for _, v in chunk_list)
    if not wire_string: return ""

    # --- 2. Unpack the character stream and decode signature ---
    signature_char, *payload_chars_list = wire_string
    
    allowed_types = list(TYPE_MAP.keys())
    type_id_num, integer_count = decode_signature(signature_char, allowed_types=allowed_types)
    type_id_str = REVERSE_TYPE_MAP[type_id_num]

    ranks = [_cp_to_rank(ord(ch)) for ch in payload_chars_list]

    # --- 3. Dispatch to the correct decoder based on type ---
    if type_id_str.startswith("lzma"):
        max_value = 255
        # Select encoding based on the specific type ID
        encoding = "utf-8" if type_id_str == "lzma8" else "utf-16le"
        integer_stream = decode_integer_stream(ranks, max_value, integer_count)
        return _decode_from_lzma_stream(integer_stream, encoding)
    
    elif type_id_str in ("o200k_base", "cl100k_base"):
        enc = tiktoken.get_encoding(type_id_str)
        max_value = enc.max_token_value
        integer_stream = decode_integer_stream(ranks, max_value, integer_count)
        return _decode_from_tokenizer_stream(integer_stream, type_id_str)
        
    else:
        raise ValueError(f"Unsupported signature type ID: {type_id_str}")

# usage idea for Conversations API
def read_conversation_metadata(
    conversation_id: str,
    datastore_name: str = "datastorage",
) -> str:
    """
    Conversations: Read and decode string data from `conversation_id` metadata using
    automatic detection via strong-reject decoders (xz-u20 first, then tokenstream).
    Orphan keys and unrelated metadata are ignored.
    """
    import httpx  # lazy import

    with httpx.Client(timeout=20) as client:
        resp = client.get(
            f"https://api.openai.com/v1/conversations/{conversation_id}",
            headers=get_api_key_headers(),
        )
        resp.raise_for_status()
        md = resp.json().get("metadata", {}) or {}
        if not isinstance(md, dict):
            raise ValueError("Conversation metadata is not a dict.")

    return decode_metadata(md, datastore_name=datastore_name)

# Conversations API - store string 'data' to metadata datastore_name
def replace_conversation_metadata(
    id: str,
    data: str,
    *,
    max_keys: int = 16,
    datastore_name: str = "datastorage",
) -> dict:
    """
    Conversation metadata updater that uses encode_metadata() and writes the
    chosen metadata to the Conversations API.

    Returns
    -------
    dict report with fields from encode_metadata(), augmented with any HTTP details.
    """
    import httpx  # lazy

    metadata, report = encode_metadata(data, max_keys=max_keys, datastore_name=datastore_name)

    with httpx.Client(timeout=20) as client:
        existing = client.get(
            f"https://api.openai.com/v1/conversations/{id}",
            headers=get_api_key_headers(),  # noqa: F821
        ).json().get("metadata", {}) or {}

        predicted = {**existing, **metadata}
        if sum(v is not None for v in predicted.values()) > 16:
            raise ValueError("New metadata plus existing keys would exceed maximum 16!")

        resp = client.post(
            f"https://api.openai.com/v1/conversations/{id}",
            headers=get_api_key_headers(),  # noqa: F821
            json={"metadata": metadata},
        )
        resp.raise_for_status()

    return report




# Script entry point
# - test strings
# - __main__: round-trip tests to the Conversations API


TEST = r"""
Need structured storage?

{
  "type": "json_schema",
  "json_schema": {
    "name": "output",
    "strict": true,
    "schema": {
      "type": "object",
      "properties": {
        "output_to_user": {
          "type": "string",
          "description": "The main message or output intended for the user."
        },
        "title_for_chat": {
          "type": "string",
          "description": "A brief, descriptive title suitable for chat summaries."
        }
      },
      "required": [
        "output_to_user",
        "title_for_chat"
      ],
      "additionalProperties": false
    }
  }
}

We can do that too!

---

These special tokens in strings could trigger a tokenizer if not configured correctly.
We throw them in to ensure the tokenizer doesn't flag or raise on token signals.
It instead encodes as their native tokens.

special_tokens = {
    **base_enc["special_tokens"],
    "<|startoftext|>": 199998,
    "<|endoftext|>": 199999,
    "<|reserved_200000|>": 200000,
    "<|reserved_200001|>": 200001,
    "<|return|>": 200002,
    "<|constrain|>": 200003,
    "<|reserved_200004|>": 200004,
    "<|channel|>": 200005,
    "<|start|>": 200006,
    "<|end|>": 200007,
    "<|message|>": 200008,
    "<|reserved_200009|>": 200009,
    "<|reserved_200010|>": 200010,
    "<|reserved_200011|>": 200011,
    "<|call|>": 200012,
}

A sampling of different languages tests the input.

ASCII/punct: !"#$%&'()*+,-./:;<=>?@[\]^_`{|}~
Latin: café naïve coöperate soufflé señor piñata
Decomposed accents: é ñ å (e+U+0301, n+U+0303, a+U+030A)
Greek: Ελληνικά — Αθήνα
Cyrillic: Привет мир
Hebrew (RTL): שלום עולם
Arabic (RTL, with diacritics): السَّلامُ عَلَيْكُمْ
Mixed bidi: ABC ‏עברית‏ DEF
Devanagari: नमस्ते दुनिया
Thai: สวัสดีโลก
CJK: 你好,世界;こんにちは世界;안녕하세요 세계
Rare CJK: 𠮷 野家
Math letters: 𝒜𝓑𝔠 𝔸𝕓𝕔 𝖆𝖇𝖈
Music symbol: 𝄞
Emoji: 😀 😁 😂 🤣 😜 🥳
Emoji ZWJ: 👨‍👩‍👧‍👦 🧑‍💻 👩🏽‍🚀
Skin tones: 👍 👍🏻 👍🏽 👍🏿
Keycaps: 1️⃣ 2️⃣ #️⃣ *️⃣
Flags: 🇺🇸 🇯🇵 🇪🇺
Variation selector: ☺️
Zero‑width joiner: A‍B (U+200D between)
Zero‑width non‑joiner: क्‌ष (U+200C)
Zero‑width space: X​Y (U+200B)
Directional marks: LRM‎ RLM‏ (U+200E U+200F)
FSI/RLI/PDI: a<U+2067>b<U+2069> a<U+2066>b<U+2069> (U+2067/U+2066/U+2069)
Quotes/backslashes: "double" 'single' backslash \ and JSON {} []
Combining stack: a̷̲̐ b̸̛͓ c͜͟͟͡͞
    """.strip()  # can cap this test with [:500] slice or similar

TEST="This language is 83 characters, a twenty token string, yet stores to only 17 chars."

def demo_auto_roundtrip(test_string: str = "") -> None:
    """
    Auto path:
    Create conversation → metadata codec → store → retrieve → verify → delete.

    Uses the global TEST string by default (if test_string is falsy)
    Prints a storage report with the chosen codec details.

    Raises on round-trip mismatch or any HTTP/codec errors.
    """
    import json
    sample = test_string or TEST
    conv_id = create_conversation()
    try:
        print(f"-- 1. Conversations: Storing {len(sample)} characters of:\n{sample[:80]}..")
        info = replace_conversation_metadata(conv_id, sample)
        print(f"-- 1a. Storage Report:\n{json.dumps(info,indent=2)}")
        print("-- 2. Retrieving")
        recovered = read_conversation_metadata(conv_id)
        assert recovered == sample, "3. BAD/ERROR: auto Round-trip mismatch"
        codec = None
        if isinstance(info, dict):
            codec = info.get("codec") or (info.get("info") or {}).get("codec")
        print(f"-- 3. Round-trip OK. Codec: {codec or 'unknown'}\n{('*'*50)}")
    except Exception as e:
        print(e)
        raise
    finally:
        delete_conversation(conv_id)

def encode_report(data):
    """
    Calls the metadata encoder (which tries two encoders).
    - Disregards the metadata consumable
    - prints the report returned
    """
    import json
    metadata, report = encode_metadata(
        data,
        max_keys=16,
        datastore_name="datastorage",
    )
    print(json.dumps(report, indent=2))

if __name__ == "__main__":
    demo_auto_roundtrip()  # creates temporary conversation to demo
    #encode_report(TEST)   # just store and show metric and trials made