Skip to content

Compiled snapshot

smrititantra.core.engine.snapshot

The compiled snapshot: the store folded into immutable arrays for recall.

compile(store, tenant, …) reads everything the activation pass touches and interns it into numpy arrays and small Python tables — refs become integer indices, edges become a degree-normalized sparse matrix, aliases become an n-gram index, sensitivities become per-clearance bitmasks. The pass then runs pure numpy over this object with no further I/O.

A snapshot serializes to snapshot.npz (the numeric arrays) plus manifest.json (vocabularies, params, and the ragged Python structures), uploaded to the artifact store under <tenant>/snapshots/<version>/ with a row in the snapshots registry. load prefers the published artifact and falls back to compiling from the store. :class:SnapshotHolder swaps the live object atomically after a health check (hot reload); pin is the kill switch.

Scale kit, designed in from day one: compile(scope=…) builds a domain-subtree shard (whole-tenant is the default); above SMRITITANTRA_SNAPSHOT_EMB_MAX vectors the embedding matrix leaves the snapshot and the dense channel queries the :class:~smrititantra.core.vector_index.VectorIndex instead — :meth:Snapshot.dense_candidates hides which mode is live.

Snapshot(tenant: str, version: str, built_from_seq: int, scope: str | None, params: dict[str, float], refs: list[str], ref_to_idx: dict[str, int], kinds: Any, kind_vocab: list[str], statuses: Any, status_vocab: list[str], sensitivities: Any, sens_vocab: list[str], titles: list[str], content_refs: list[str | None], valid_from: Any, valid_to: Any, alias_index: dict[str, list[int]], max_alias_words: int, spread_indptr: Any, spread_indices: Any, spread_data: Any, requires_adj: list[list[tuple[int, str]]], payload_edges: dict[str, dict[str, Any]], bindings_by_node: list[list[dict[str, Any]]], emb: Any | None, emb_mask: Any, emb_dim: int, emb_externalized: bool, prior: Any, label_masks: dict[str, Any], event_nodes: list[tuple[int, float, list[int]]], mask_cache: dict[str, Any] = dict(), route_sig: list[dict[int, float]] = list(), route_ground: list[list[tuple[int, float]]] = list(), route_inverted: dict[int, list[int]] = dict(), route_rel: Any = None, route_pinned: Any = None, route_key: Any | None = None, route_key_mask: Any = None) dataclass

Immutable, self-contained input to the activation pass.

dense_candidates(q_emb: list[float] | None, index: VectorIndex | None, k: int) -> list[tuple[int, float]] async

Top-k (node idx, cosine) for the dense channel, mode-agnostic.

In-RAM mode multiplies against the resident matrix; externalized mode queries the vector index over the concept and asset spaces. Returns [] when there is no query embedding (embedder down).

SnapshotHolder(snapshot: Snapshot | None = None)

Holds one live snapshot; swaps it atomically after a health check.

swap(snapshot: Snapshot) -> None

Health-check the candidate, then swap the single reference.

compile(store: MemoryStore, tenant: str, config: TenantConfig, *, embedder: Embedder | None = None, index: VectorIndex | None = None, emb_max: int = 200000, scope: str | None = None, drop_rel_classes: set[str] | None = None) -> Snapshot async

Read the store and fold it into an immutable :class:Snapshot.

drop_rel_classes compiles without edges of the given relation classes — the eval harness's per-channel ablation hook (e.g. {"assoc"} answers "what would recall lose without the derived associations?").

compile_incremental(store: MemoryStore, tenant: str, config: TenantConfig, current: Snapshot, *, embedder: Embedder | None = None, index: VectorIndex | None = None, emb_max: int = 200000) -> tuple[Snapshot, str] async

Delta compile from the ledger position (the scale kit).

Classifies everything since current.built_from_seq:

  • nothing that touches the snapshot -> ("unchanged", ...) — the same arrays, cursor advanced;
  • only usage events -> ("prior_patch", ...) — the prior vector is recomputed and patched onto the existing arrays (no interning, no CSR rebuild, no re-embedding);
  • any structural change (P1/P2/S1 rows, mined episodes, vocabulary) -> ("full", ...) — the fallback is always the full rebuild, which also remains the consolidation path.

load(store: MemoryStore, artifacts: ArtifactStore, tenant: str, config: TenantConfig, *, embedder: Embedder | None = None, index: VectorIndex | None = None, emb_max: int = 200000) -> Snapshot async

Prefer the published artifact; fall back to compiling from the store.

load_version(store: MemoryStore, artifacts: ArtifactStore, tenant: str, version: str) -> Snapshot async

Deserialize one registered snapshot's artifact (gated publish, eval).

pin(store: MemoryStore, tenant: str, version: str) -> None async

Kill switch: force a specific version live (must already be registered).

publish(store: MemoryStore, artifacts: ArtifactStore, snapshot: Snapshot, *, set_live: bool = True) -> None async

Serialize, upload, register, and (default) flip the published flag.

serialize(snapshot: Snapshot) -> tuple[bytes, dict[str, Any]]

(npz bytes, manifest dict). The npz holds arrays; the manifest the rest.