Runtime

Runtime archive, deduplication, caching, and scheduling.

class nlp_shap.runtime.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.

class nlp_shap.runtime.CoalitionJob(coalition_key: str, snapshot_id: str, snapshot: ConversationSnapshot, absence_policy: str, mask_words: bytes, mask_n_bits: int, model_id: str, utility: float, 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.

utility: float

Utility score assigned after generation.

prefix_hash: str

Shared prompt prefix hash used for KV-cache grouping.

class nlp_shap.runtime.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 about to be scheduled.

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

Report that done of total coalition jobs have finished.

class nlp_shap.runtime.CoalitionRecord(record_id: int, snapshot_id: str, coalition_key: str, mask: PackedMask, absence_policy: str, model_id: str, generation_text: str, utility: float, elapsed_ms: float, cache_hit: bool)

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.

utility: float

Utility score assigned to the generated output.

elapsed_ms: float

Wall-clock generation time in milliseconds.

cache_hit: bool

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

class nlp_shap.runtime.CoalitionRecordDraft(snapshot_id: str, coalition_key: str, mask: PackedMask, absence_policy: str, model_id: str, generation_text: str, utility: float, elapsed_ms: float, cache_hit: bool)

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.

utility: float

Utility score assigned to the generated output.

elapsed_ms: float

Wall-clock generation time in milliseconds.

cache_hit: bool

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

class nlp_shap.runtime.HotResultStore(maxsize: int | None = 128, _values: OrderedDict[str, str]=<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.

put(coalition_key: str, generation_text: str) → None

Insert or refresh a cached generation result.

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

Bases: object

Collect span records in memory for tests and diagnostics.

spans: list[SpanRecord]

span(name: str) → Iterator[None]

Append one span record when the context exits.

class nlp_shap.runtime.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]]) → SchedulerMetrics

Execute coalition jobs and return scheduler counters.

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

Execute jobs from an iterable without creating all coroutines upfront.

async run_stream(jobs: AsyncIterator[CoalitionJob], generate: Callable[[ConversationSnapshot], Awaitable[str]]) → SchedulerMetrics

Execute jobs from an async iterator with bounded pending tasks.

class nlp_shap.runtime.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.

class nlp_shap.runtime.NullObservabilitySink

Bases: object

No-op observability sink used when telemetry is disabled.

span(name: str) → Iterator[None]

Yield without recording span metadata.

class nlp_shap.runtime.ObservabilitySink(*args, **kwargs)

Bases: Protocol

Protocol for recording explain pipeline stage spans.

span(name: str) → Iterator[None]

Record wall-clock duration for one named pipeline stage.

class nlp_shap.runtime.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.

class nlp_shap.runtime.PrefixCacheManager

Bases: object

Store and reuse transformer past_key_values for shared prefixes.

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

Return cached KV state and the longest matching prefix length.

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

Return cached KV state and update hit or miss counters.

get_at_length(token_ids: Sequence[int], length: int) → Any | None

Return cached KV state for an exact prefix length.

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

Persist KV state for an exact token-id prefix.

reset() → None

Clear cached prefixes and counters.

class nlp_shap.runtime.RunArchive(root: Path, manifest: RunManifest, flush_every: int = 50)

Bases: object

Persist coalition records to SQLite with generation text stored as blobs.

classmethod open(root: Path, manifest: RunManifest, flush_every: int = 50) → Self

Create a run archive directory and open its SQLite database.

classmethod load(root: Path) → Self

Open an existing run archive without rewriting its manifest.

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.

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.

class nlp_shap.runtime.SchedulerMetrics(requested: int, executed: int, deduplicated: int, cache_hits: int, kv_cache_hits: 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

Number of coalition generations that reused a cached prompt prefix.

class nlp_shap.runtime.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.

nlp_shap.runtime.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.

nlp_shap.runtime.build_snapshot_prefix_hash(snapshot: ConversationSnapshot) → str

Hash the shared text prefix excluding the final whitespace token.

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

Return whether coalition deduplication should be active.

nlp_shap.runtime.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.

SQLite-backed run archive for coalition evaluation history.

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

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

class nlp_shap.runtime.archive.CoalitionRecord(record_id: int, snapshot_id: str, coalition_key: str, mask: PackedMask, absence_policy: str, model_id: str, generation_text: str, utility: float, elapsed_ms: float, cache_hit: bool)

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.

utility: float

Utility score assigned to the generated output.

elapsed_ms: float

Wall-clock generation time in milliseconds.

cache_hit: bool

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

class nlp_shap.runtime.archive.CoalitionRecordDraft(snapshot_id: str, coalition_key: str, mask: PackedMask, absence_policy: str, model_id: str, generation_text: str, utility: float, elapsed_ms: float, cache_hit: bool)

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.

utility: float

Utility score assigned to the generated output.

elapsed_ms: float

Wall-clock generation time in milliseconds.

cache_hit: bool

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

class nlp_shap.runtime.archive.RunArchive(root: Path, manifest: RunManifest, flush_every: int = 50)

Bases: object

Persist coalition records to SQLite with generation text stored as blobs.

classmethod open(root: Path, manifest: RunManifest, flush_every: int = 50) → Self

Create a run archive directory and open its SQLite database.

classmethod load(root: Path) → Self

Open an existing run archive without rewriting its manifest.

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.

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.

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

Return whether coalition deduplication should be active.

nlp_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 nlp_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.

class nlp_shap.runtime.store.HotResultStore(maxsize: int | None = 128, _values: OrderedDict[str, str]=<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.

put(coalition_key: str, generation_text: str) → None

Insert or refresh a cached generation result.

Async coalition inference scheduling with deduplication.

nlp_shap.runtime.scheduler.GenerateFn

Async callable that returns generated text for one snapshot.

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

class nlp_shap.runtime.scheduler.SchedulerMetrics(requested: int, executed: int, deduplicated: int, cache_hits: int, kv_cache_hits: 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

Number of coalition generations that reused a cached prompt prefix.

class nlp_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, utility: float, 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.

utility: float

Utility score assigned after generation.

prefix_hash: str

Shared prompt prefix hash used for KV-cache grouping.

class nlp_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]]) → SchedulerMetrics

Execute coalition jobs and return scheduler counters.

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

Execute jobs from an iterable without creating all coroutines upfront.

async run_stream(jobs: AsyncIterator[CoalitionJob], generate: Callable[[ConversationSnapshot], Awaitable[str]]) → SchedulerMetrics

Execute jobs from an async iterator with bounded pending tasks.

Optional synchronous progress callbacks for coalition execution.

class nlp_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 about to be scheduled.

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

Report that done of total coalition jobs have finished.

class nlp_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.