Runtime archive, deduplication, and async scheduling

Business context: Production explain runs generate thousands of coalition evaluations. Persisted archives support compliance replay, deduplication avoids redundant API spend, and bounded concurrency keeps latency predictable under load.

Goal: Walk through every public runtime API in nlp-shap 0.1.4 — run archive, coalition keys, hot LRU cache, and the async inference scheduler.

Prerequisites: nlp-shap 0.1.4+, Python 3.12.

Theory: See the runtime theory page.

Expected output: 100 coalition rows round-trip through SQLite; ten unique masks collapse to ten backend calls; scheduler metrics report cache hits and bounded concurrency.

import asyncio
import json
from pathlib import Path
import tempfile

from nlp_shap import Estimand
from nlp_shap.domain.conversation import ConversationSnapshot, Message, Role, Turn
from nlp_shap.domain.estimands import estimand_to_wire
from nlp_shap.masking.codec import MaskCodec
from nlp_shap.pipeline.config import DedupConfig, GenerationConfig
from nlp_shap.pipeline.manifest import RunManifest, parse_manifest
from nlp_shap.runtime import (
    CoalitionDedupRegistry,
    CoalitionJob,
    CoalitionRecordDraft,
    HotResultStore,
    InferenceScheduler,
    RunArchive,
    build_coalition_key,
    dedup_enabled,
)

Persist coalition history

Business: Audit teams need immutable run folders that pair attribution metadata with the exact generations that produced each coalition utility.

Technical: RunArchive.open writes manifest.json, SQLite rows, and blob files. history_lazy streams records without bulk-loading blobs.

turn = Turn(messages=(Message(role=Role.USER, text="Who are you?"),))
snapshot = ConversationSnapshot.from_turns((turn,))
manifest = RunManifest(estimand=Estimand.SHAPLEY, run_id="runtime-demo")
packed = MaskCodec.pack((True, False, True))

with tempfile.TemporaryDirectory() as tmp:
    root = Path(tmp) / "runtime-demo"
    with RunArchive.open(root, manifest, flush_every=25) as archive:
        for index in range(100):
            archive.append(
                CoalitionRecordDraft(
                    snapshot_id=snapshot.snapshot_id,
                    coalition_key=f"key-{index % 10}",
                    mask=packed,
                    absence_policy="delete",
                    model_id="mock",
                    generation_text=f"generation-{index}",
                    utility=float(index),
                    elapsed_ms=float(index),
                    cache_hit=index % 2 == 0,
                )
            )

    payload = parse_manifest(json.loads((root / "manifest.json").read_text()))
    with RunArchive.open(root, manifest) as archive:
        records = list(archive.history_lazy())

print(payload.estimand, estimand_to_wire(payload.estimand), len(records))

shapley shapley 100

Coalition dedup keys

Business: Identical coalition prompts at temperature zero should not trigger duplicate model calls — a direct infrastructure cost control.

Technical: build_coalition_key hashes snapshot, players, packed mask, policy, model, and generation settings. dedup_enabled follows config auto/on/off.

generation = GenerationConfig(temperature=0.0)
key = build_coalition_key(
    snapshot_id=snapshot.snapshot_id,
    player_ids=("p0", "p1"),
    mask_present=(True, False),
    absence_policy="delete",
    model_id="mock",
    generation=generation,
)
print("dedup auto:", dedup_enabled(DedupConfig(enabled="auto"), generation))
print("key prefix:", key[:16])

registry = CoalitionDedupRegistry()
print("new:", registry.observe(key))
print("repeat:", registry.observe(key))
print("unique:", len(registry))

dedup auto: True
key prefix: 903952d03ee4df35
new: True
repeat: False
unique: 1

Hot LRU cache

Business: Recent coalitions often repeat within one explain run; an in-memory LRU avoids even hitting the dedup registry on hot paths.

Technical: HotResultStore keeps the most recent coalition generations up to maxsize.

store = HotResultStore(maxsize=2)
store.put("a", "one")
store.put("b", "two")
store.get("a")
store.put("c", "three")
print("a", store.get("a"))
print("b", store.get("b"))
print("c", store.get("c"))

a one
b None
c three

Async scheduler with bounded concurrency

Business: Unbounded parallel model calls overwhelm backends and inflate bills; max_inflight caps concurrency while metrics expose savings from cache and dedup.

Technical: InferenceScheduler.run checks the hot store, dedup registry, then a semaphore before invoking the async generate callable.

async def generate(snapshot: ConversationSnapshot) -> str:
    await asyncio.sleep(0.001)
    return snapshot.turns[0].messages[0].text

scheduler = InferenceScheduler(
    max_inflight=2,
    generation=GenerationConfig(temperature=0.0),
    store=HotResultStore(),
    dedup=CoalitionDedupRegistry(),
)
jobs = [
    CoalitionJob(
        coalition_key=f"key-{index % 10}",
        snapshot_id=snapshot.snapshot_id,
        snapshot=snapshot,
        absence_policy="delete",
        mask_words=packed.words,
        mask_n_bits=packed.n_bits,
        model_id="mock",
        utility=1.0,
    )
    for index in range(100)
]
metrics = await scheduler.run(jobs, generate)
print(metrics)

SchedulerMetrics(requested=100, executed=10, deduplicated=90, cache_hits=0)