Runtime

Runtime archive, deduplication, caching, and scheduling.

@domain runtime @see [[knowledge/architecture/runtime]]

SQLite-backed run archive for coalition evaluation history.

@domain runtime @see [[knowledge/architecture/runtime]] @tested-by [[tests/runtime/test_archive.py]]

mllm_shap.runtime.archive.BASE_GENERATION_FILE = 'base_generation.txt'

Filename for the grand-coalition reference generation at the archive root.

mllm_shap.runtime.archive.ARCHIVE_SCHEMA_VERSION: Final[int] = 4

SQLite user_version written by this release and required on load.

mllm_shap.runtime.archive.MANIFEST_SCHEMA_VERSION: Final[int] = 4

Manifest schema_version written by this release and required on load.

class mllm_shap.runtime.archive.ManifestPayloadSource(*args, **kwargs)

Bases: Protocol

Manifest that serializes itself to the JSON stored as manifest.json.

to_dict() → Mapping[str, object]

Return the JSON-compatible manifest payload.

mllm_shap.runtime.archive.check_manifest_schema_version(payload: Mapping[str, object]) → None

Refuse manifests written with a different schema version.

Raises:

ManifestVersionError – If schema_version is missing or differs from MANIFEST_SCHEMA_VERSION.

mllm_shap.runtime.archive.read_manifest_payload(path: Path) → Mapping[str, object]

Read manifest.json and return its top-level JSON object.

Raises:

ArchiveCorruptError – If the file is not UTF-8 JSON or its root is not an object.

class mllm_shap.runtime.archive.CoalitionRecord(record_id: int, snapshot_id: str, coalition_key: str, mask: PackedMask, absence_policy: str, model_id: str, generation_text: str, elapsed_ms: float, cache_hit: bool, logprobs: tuple[float, ...] | None = None)

Bases: object

One persisted coalition evaluation row.

record_id: int

Monotonic archive identifier for the coalition row.

snapshot_id: str

Conversation snapshot identifier evaluated for this coalition.

coalition_key: str

Stable deduplication key for the coalition evaluation.

mask: PackedMask

Packed coalition mask bytes and original bit length.

absence_policy: str

Registered absence-policy identifier used for rendering.

model_id: str

Backend model identifier used for generation.

generation_text: str

Generated model text for the coalition.

elapsed_ms: float

Wall-clock generation time in milliseconds.

cache_hit: bool

Whether the generation was served from an in-memory cache.

logprobs: tuple[float, ...] | None

Token logprobs of the generation; None when the backend gave none.

class mllm_shap.runtime.archive.CoalitionRecordDraft(snapshot_id: str, coalition_key: str, mask: PackedMask, absence_policy: str, model_id: str, generation_text: str, elapsed_ms: float, cache_hit: bool, logprobs: tuple[float, ...] | None = None)

Bases: object

Input payload used when appending a coalition record.

snapshot_id: str

Conversation snapshot identifier evaluated for this coalition.

coalition_key: str

Stable deduplication key for the coalition evaluation.

mask: PackedMask

Packed coalition mask bytes and original bit length.

absence_policy: str

Registered absence-policy identifier used for rendering.

model_id: str

Backend model identifier used for generation.

generation_text: str

Generated model text for the coalition.

elapsed_ms: float

Wall-clock generation time in milliseconds.

cache_hit: bool

Whether the generation was served from an in-memory cache.

logprobs: tuple[float, ...] | None

Token logprobs of the generation; None when the backend gave none.

class mllm_shap.runtime.archive.RunArchive(root: Path, flush_every: int = 50, read_only: bool = False)

Bases: object

Persist coalition records, generation text included, to one SQLite file.

The database uses WAL journaling with synchronous=NORMAL; rows are committed every flush_every appends and on close().

classmethod open(root: Path, manifest: ManifestPayloadSource, flush_every: int = 50, on_exists: Literal['error', 'overwrite'] = 'error') → Self

Create a run archive directory and open its SQLite database.

An existing archive whose run did not finish (no completion marker) is replaced whatever on_exists says, so a crashed run can be re-run.

Raises:

ArchiveExistsError – If root already holds the archive of a finished run and on_exists is "error".

classmethod load(root: Path) → Self

Open an existing run archive read-only, without writing any file.

Raises:

• ArchiveNotFoundError – If the manifest or the SQLite database is missing (a FileNotFoundError).

• ArchiveCorruptError – If the manifest is not a JSON object or the database cannot be read.

• ManifestVersionError – If the manifest or SQLite schema version is not the one written by this release.

write_manifest(manifest: ManifestPayloadSource) → None

Write manifest to manifest.json, replacing the previous one.

write_base_generation(text: str) → None

Persist the grand-coalition reference generation before coalition rows.

read_base_generation() → str | None

Return the archived base generation when present.

append(draft: CoalitionRecordDraft) → int

Append one coalition record and return its archive identifier.

mark_complete() → None

Record that every coalition of the run was archived.

property is_complete: bool

Return whether mark_complete() ran for this archive.

flush() → None

Commit pending archive writes to disk.

history_lazy() → Iterator[CoalitionRecord]

Iterate coalition records one row at a time without bulk preloading.

close() → None

Flush and close the archive database connection.

Coalition deduplication keys and registry.

@domain runtime @see [[knowledge/architecture/runtime]] @tested-by [[tests/runtime/test_dedup.py]]

mllm_shap.runtime.dedup.dedup_enabled(config: DedupConfig, generation: GenerationConfig) → bool

Return whether coalition deduplication should be active.

mllm_shap.runtime.dedup.build_coalition_key(snapshot_id: str, player_ids: tuple[str, ...], mask_present: tuple[bool, ...], absence_policy: str, model_id: str, generation: GenerationConfig) → str

Build a stable SHA256 coalition key for deduplication.

class mllm_shap.runtime.dedup.CoalitionDedupRegistry

Bases: object

Track coalition keys that already required backend execution.

observe(coalition_key: str) → bool

Record a coalition key and return whether it is newly observed.

contains(coalition_key: str) → bool

Return whether a coalition key was already observed.

Hot LRU cache for coalition generation results.

@domain runtime @see [[knowledge/architecture/runtime]] @tested-by [[tests/runtime/test_store.py]]

class mllm_shap.runtime.store.HotResultStore(maxsize: int | None = 128, _values: OrderedDict[str, tuple[str, tuple[float, ...] | None]]=<factory>)

Bases: object

In-memory LRU store for recently generated coalition outputs.

maxsize: int | None = 128

Maximum retained results; None disables eviction.

get(coalition_key: str) → str | None

Return a cached generation or None when absent.

logprobs(coalition_key: str) → tuple[float, ...] | None

Return the cached generation’s token logprobs, None when unknown.

put(coalition_key: str, generation_text: str, logprobs: tuple[float, ...] | None = None) → None

Insert or refresh a cached generation result.

Async coalition inference scheduling with deduplication.

@domain runtime @see [[knowledge/architecture/runtime]] @tested-by [[tests/runtime/test_scheduler.py]]

mllm_shap.runtime.scheduler.GenerateFn

Async callable that returns the generated text, or a record carrying it and its token logprobs, for one snapshot.

alias of Callable[[ConversationSnapshot], Awaitable[str | GenerationRecord]]

class mllm_shap.runtime.scheduler.SchedulerMetrics(requested: int, executed: int, deduplicated: int, cache_hits: int, kv_cache_hits: int = 0, kv_cache_reused_tokens: int = 0)

Bases: object

Counters collected while executing coalition jobs.

requested: int

Number of coalition jobs submitted to the scheduler.

executed: int

Number of jobs that invoked the backend generate callable.

deduplicated: int

Number of jobs skipped because the coalition key was already executed.

cache_hits: int

Number of jobs served from the hot result store without backend calls.

kv_cache_hits: int

Coalition generations that reused a cached prompt prefix; filled by the orchestrator from the backend, 0 from the scheduler alone.

kv_cache_reused_tokens: int

Prompt tokens served from cached prefixes; filled like kv_cache_hits.

class mllm_shap.runtime.scheduler.CoalitionJob(coalition_key: str, snapshot_id: str, snapshot: ConversationSnapshot, absence_policy: str, mask_words: bytes, mask_n_bits: int, model_id: str, prefix_hash: str = '')

Bases: object

One coalition evaluation request routed through the scheduler.

coalition_key: str

Stable deduplication key for the coalition.

snapshot_id: str

Identifier of the evaluated conversation snapshot.

snapshot: ConversationSnapshot

Conversation snapshot passed to the backend for generation.

absence_policy: str

Registered absence-policy identifier used for rendering.

mask_words: bytes

Packed coalition mask bytes.

mask_n_bits: int

Original coalition mask bit length.

model_id: str

Backend model identifier used for generation.

prefix_hash: str

Shared prompt prefix hash used for KV-cache grouping.

class mllm_shap.runtime.scheduler.InferenceScheduler(max_inflight: int, generation: GenerationConfig, store: HotResultStore, dedup: CoalitionDedupRegistry | None = None, archive: RunArchive | None = None, pending_limit: int | None = None, progress: CoalitionProgress | None = None)

Bases: object

Execute coalition jobs with bounded concurrency and optional deduplication.

async run(jobs: Sequence[CoalitionJob], generate: Callable[[ConversationSnapshot], Awaitable[str | GenerationRecord]]) → SchedulerMetrics

Execute coalition jobs and return scheduler counters.

async run_iter(jobs: Iterable[CoalitionJob], generate: Callable[[ConversationSnapshot], Awaitable[str | GenerationRecord]], total: int | None = None) → SchedulerMetrics

Execute jobs from an iterable without creating all coroutines upfront.

Jobs run in one asyncio.TaskGroup, with at most pending_limit tasks alive at a time. The first job failure cancels every other task and is re-raised as is (not wrapped in an exception group), so callers never observe partial success. With total the progress sink is told the cumulative planned count of all runs so far.

Optional synchronous progress callbacks for coalition execution.

@domain runtime @see [[knowledge/architecture/runtime]] @tested-by [[tests/runtime/test_progress.py]]

class mllm_shap.runtime.progress.CoalitionProgress(*args, **kwargs)

Bases: Protocol

Sync hooks notified as coalition jobs are planned and finished.

on_coalitions_planned(total: int) → None

Report the number of coalition jobs planned so far in the run.

Called again with a larger total when a second sampling phase (Neyman) adds jobs; finished counts keep growing across phases.

on_coalition_finished(done: int, total: int) → None

Report that done of total coalition jobs have finished.

class mllm_shap.runtime.progress.NullCoalitionProgress

Bases: object

No-op progress sink used when callers omit a progress callback.

on_coalitions_planned(total: int) → None

Ignore planned coalition count.

on_coalition_finished(done: int, total: int) → None

Ignore finished coalition count.

Performance summaries for explain pipeline runs.

@domain runtime @see [[knowledge/architecture/runtime]]

class mllm_shap.runtime.metrics.PerfSummary(total_ms: float, generation_ms: float, scoring_ms: float, aggregation_ms: float)

Bases: object

Wall-clock breakdown for one explain or reanalyze run.

total_ms: float

Total wall-clock time for the run.

generation_ms: float

Time spent in base and coalition generation.

scoring_ms: float

Time spent scoring coalition outputs with the value function.

aggregation_ms: float

Time spent aggregating coalition payoffs into attributions.

Optional observability hooks for explain pipeline stages.

@domain runtime @see [[knowledge/architecture/runtime]] @tested-by [[tests/runtime/test_telemetry.py]]

class mllm_shap.runtime.telemetry.SpanRecord(name: str, duration_ms: float)

Bases: object

One completed pipeline stage span.

name: str

Stage identifier emitted by the orchestrator or estimator.

duration_ms: float

Wall-clock duration of the stage in milliseconds.

class mllm_shap.runtime.telemetry.ObservabilitySink(*args, **kwargs)

Bases: Protocol

Protocol for recording explain pipeline stage spans.

span(name: str) → Generator[None]

Record wall-clock duration for one named pipeline stage.

class mllm_shap.runtime.telemetry.NullObservabilitySink

Bases: object

No-op observability sink used when telemetry is disabled.

span(name: str) → Generator[None]

Yield without recording span metadata.

class mllm_shap.runtime.telemetry.InMemoryObservabilitySink(spans: list[SpanRecord] = <factory>)

Bases: object

Collect span records in memory for tests and diagnostics.

spans: list[SpanRecord]

span(name: str) → Generator[None]

Append one span record when the context exits.

Prefix-cache storage for shared prompt token prefixes.

@domain runtime @see [[knowledge/architecture/runtime]] @tested-by [[tests/runtime/test_kv_cache.py]]

mllm_shap.runtime.kv_cache.build_snapshot_prefix_hash(snapshot: ConversationSnapshot, player_set: PlayerSet, mask: CoalitionMask) → str

Hash the coalition prefix: the presence of every player except the last.

Coalitions of snapshot that differ only in the last player render the same text up to that player’s span, so they share a prompt prefix and run next to each other for KV-cache reuse. Nothing is re-tokenised.

class mllm_shap.runtime.kv_cache.CroppableCache(*args, **kwargs)

Bases: Protocol

KV state that can be cut back to its first tokens (HF DynamicCache).

crop(max_length: int) → None

Keep the state of the first max_length tokens only.

class mllm_shap.runtime.kv_cache.PrefixCacheManager(max_entries: int = 8)

Bases: object

Keep at most max_entries prefix KV states keyed by token-id tuple.

hits

Number of lookups that found a cached prefix.

misses

Number of lookups that found nothing.

longest_prefix(token_ids: Sequence[int]) → tuple[Any | None, int]

Return the cached state sharing the longest token prefix with token_ids.

A stored key matches when it is a proper prefix of token_ids, or, for CroppableCache states, when it shares at least one leading token (the caller crops the state to the returned length). At least one token is always left to process. Walks every stored entry once.

store(token_ids: Sequence[int], past_key_values: Any) → None

Store the KV state of token_ids, evicting the oldest beyond the cap.

reset() → None

Clear cached prefixes and counters.

mllm_shap.runtime.kv_cache.group_jobs_for_prefix_cache(jobs: list[T], *, key: Callable[[T], str]) → list[T]

Sort jobs so coalitions with the same prefix hash run adjacently.