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)

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.

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 = 128, _values: OrderedDict[str, str]=<factory>)

Bases: object

In-memory LRU store for recently generated coalition outputs.

maxsize: int = 128

Maximum number of coalition results retained in memory.

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.InferenceScheduler(max_inflight: int, generation: GenerationConfig, store: HotResultStore, dedup: CoalitionDedupRegistry | None = None, archive: RunArchive | None = None, pending_limit: int | 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]]) → 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.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) → RunArchive

Create a run archive directory and open its SQLite database.

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)

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.

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.dedup_enabled(config: DedupConfig, generation: GenerationConfig) → bool

Return whether coalition deduplication should be active.

SQLite-backed run archive for coalition evaluation history.

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) → RunArchive

Create a run archive directory and open its SQLite database.

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 = 128, _values: OrderedDict[str, str]=<factory>)

Bases: object

In-memory LRU store for recently generated coalition outputs.

maxsize: int = 128

Maximum number of coalition results retained in memory.

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)

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.

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)

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.

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)

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]]) → 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.