MemoryService
Validate and orchestrate Memory operations without exposing storage details.
Functions
constructor__init__(*, backend, candidate_pipeline=None, embedding_model=None, reranker=None, rerank_candidate_limit=30, source_resolver=None, artifact_resolver=None, id_factory=None, prompt_context=None, write_gate=None, capacity_budget=None, compaction=None, max_history_revisions=100) -> NoneparambackendMemoryBackendparamcandidate_pipelineCandidatePipeline | None= Noneparamembedding_modelEmbeddingModel | None= NoneparamrerankerMemoryReranker | None= Noneparamrerank_candidate_limitint= 30paramsource_resolver_SourceResolver | None= Noneparamartifact_resolver_ArtifactResolver | None= Noneparamid_factoryIdFactory | None= Noneparamprompt_contextScopedPrompts | None= Noneparamwrite_gateMemoryWriteGate | None= Noneparamcapacity_budgetMemoryCapacityBudget | None= NoneparamcompactionMemoryCompactionPolicy | None= Noneparammax_history_revisionsint= 100Returns
Nonefuncget(memory) -> MemoryReturn the canonical exact Memory Revision matching memory.
parammemoryMemoryReturns
powercontext.builtin.artifacts.memory.models.Memoryfunclatest(memory) -> MemoryReturn the current head of the same Memory identity.
parammemoryMemoryReturns
powercontext.builtin.artifacts.memory.models.Memoryfuncrevisions(memory, /, *, since_revision=0, through_revision=None) -> tuple[Memory, ...]Return history in ascending order within (since_revision, through_revision].
The upper bound defaults to the current head. The history limit applies to the requested interval, and oversized intervals are never truncated.
parammemoryMemoryparamsince_revisionint= 0paramthrough_revisionint | None= NoneReturns
tuple[powercontext.builtin.artifacts.memory.models.Memory, ...]funchead(artifact_id) -> MemoryReturn the current Memory head by its stable Artifact identity.
paramartifact_idstrReturns
powercontext.builtin.artifacts.memory.models.Memoryfunccapacity(memory) -> MemoryCapacityMeasure an exact Revision, including eligible tombstones even when compaction is disabled.
Eligibility can load complete manifests across the tombstone recovery window. Cost scales with their combined size; this is not a cheap counter.
parammemoryMemoryReturns
powercontext.builtin.artifacts.memory.models.MemoryCapacityfunc_capacity_limits() -> tuple[tuple[MemoryCapacityDimension, int], ...]Returns
tuple[tuple[powercontext.builtin.artifacts.memory.models.MemoryCapacityDimension, int], ...]func_capacity_values(content, content_bytes=None) -> dict[MemoryCapacityDimension, int]paramcontentMemoryContentparamcontent_bytesbytes | None= NoneReturns
dict[powercontext.builtin.artifacts.memory.models.MemoryCapacityDimension, int]func_require_capacity(base, content, *, growth, content_bytes) -> NoneparambaseMemory | NoneparamcontentMemoryContentparamgrowthfrozenset[MemoryCapacityDimension]paramcontent_bytesbytesReturns
Nonefunc_compactable_entry_ids(memory) -> tuple[str, ...]parammemoryMemoryReturns
tuple[str, ...]funccompact(memory, *, dry_run=False, limit=None, reason=None) -> MemoryCompactionResultDrop aged, untagged tombstones; retain every prior Revision and entry body.
Previews are available while compaction is disabled. Reclaimed bytes are the signed difference of complete canonical contents, including the audit changes and reason, which can outweigh a small manifest reduction.
parammemoryMemoryparamdry_runbool= Falseparamlimitint | None= Noneparamreasonstr | None= NoneReturns
powercontext.builtin.artifacts.memory.models.MemoryCompactionResultfunchead_entries(artifact_id) -> tuple[Memory, tuple[MemoryEntryVersion, ...]]Return the current Memory head together with its validated entry objects.
entries re-reads the caller's Memory to prove it matches storage, which a head
read straight from the backend already satisfies. Callers that need both the head
and its entries use this instead, so a read-only pass over many Scopes does not
re-fetch every Memory Revision it just loaded.
paramartifact_idstrReturns
tuple[powercontext.builtin.artifacts.memory.models.Memory, tuple[powercontext.builtin.artifacts.memory.models.MemoryEntryVersion, ...]]funcrevision(memory) -> MemoryReturn one exact Memory Revision by its stable reference.
parammemoryArtifactRefReturns
powercontext.builtin.artifacts.memory.models.Memoryfuncremember(*, memory, sources=(), artifacts=(), entries=(), mode='auto') -> Memory | NoneAppend or extract validated entry changes against one exact head.
parammemoryMemory | NoneparamsourcesSequence[Source]= ()paramartifactsSequence[Artifact[object]]= ()paramentriesSequence[MemoryEntryInput]= ()parammodeMemoryRememberMode= 'auto'Returns
powercontext.builtin.artifacts.memory.models.Memory | Nonefuncplan_remember(*, memory, sources=(), artifacts=(), entries=(), mode='auto') -> MemoryWritePlanValidate and prepare a write without mutating authoritative storage.
parammemoryMemory | NoneparamsourcesSequence[Source]= ()paramartifactsSequence[Artifact[object]]= ()paramentriesSequence[MemoryEntryInput]= ()parammodeMemoryRememberMode= 'auto'Returns
powercontext.builtin.artifacts.memory.protocols.MemoryWritePlanfuncapply(plan) -> Memory | NoneApply one prepared write through this service's transaction boundary.
paramplanMemoryWritePlanReturns
powercontext.builtin.artifacts.memory.models.Memory | Nonefuncforget(memory, *, entries, reason=None) -> MemoryDeactivate logical entries without deleting immutable content.
parammemoryMemoryparamentriesSequence[MemoryEntryVersion]paramreasonstr | None= NoneReturns
powercontext.builtin.artifacts.memory.models.Memoryfuncreactivate(memory, *, entries, reason=None) -> MemoryRestore inactive logical entries without creating body versions.
parammemoryMemoryparamentriesSequence[MemoryEntryVersion]paramreasonstr | None= NoneReturns
powercontext.builtin.artifacts.memory.models.Memoryfuncorganize(memory, *, mode='default') -> MemoryApply only exact deduplication and canonical normalization.
parammemoryMemoryparammodeLiteral['default', 'dedupe', 'normalize']= 'default'Returns
powercontext.builtin.artifacts.memory.models.Memoryfuncchanges(memory, *, since_revision=None) -> tuple[MemoryRevisionChanges, ...]Read compact change summaries without expanding entry bodies.
parammemoryMemoryparamsince_revisionint | None= NoneReturns
tuple[powercontext.builtin.artifacts.memory.models.MemoryRevisionChanges, ...]funcsearch(query, *, memories, limit=10, mode='auto', tag_filter=None, admission=None, query_embedding=None) -> MemorySearchResultSearch explicit current Memory heads with capability-safe fallback.
admission=None applies the historical fusion-time thresholds bit for bit.
query_embedding lets a caller reuse a vector it already paid for (RFC 1560's
expansion rounds reuse the round-0 vector). Reuse is best-effort and honest: it is
applied only when the supplied profile equals the one this search resolved, and the
call reports embedding_calls = 0 in that case, 1 when it embedded. The
resolved vector is handed back on :attr:MemorySearchResult.query_embedding so the
next round can reuse it. When the mode resolved to fts there is no vector to
report, so the field stays None and the next round must pay again.
paramquerystrparammemoriesSequence[Memory]paramlimitint= 10parammodeMemorySearchMode= 'auto'paramtag_filterTagFilter | None= NoneparamadmissionAdmissionFloor | None= Noneparamquery_embeddingMemoryQueryEmbedding | None= NoneReturns
powercontext.builtin.artifacts.memory.models.MemorySearchResultfunc_resolve_query_vector(*, query, requested_mode, selected_mode, profile, capabilities, reuse) -> tuple[MemoryUsedSearchMode, tuple[float, ...] | None, MemoryQueryEmbedding | None, int]Resolve — or reuse — the query vector for a vector or hybrid search.
Returns (selected_mode, query_vector, resolved_embedding, embedding_calls).
resolved_embedding is None exactly when the search fell back to fts, which
is also when the caller must drop the embedding profile from the backend request.
Reuse applies only when the supplied profile equals the resolved one; otherwise the round embeds and reports one call. A failed embedding is still counted as one call, because the call was issued — the cost is real even though it produced nothing.
paramquerystrparamrequested_modeMemorySearchModeparamselected_modeMemoryUsedSearchModeparamprofileEmbeddingProfileparamcapabilitiesMemoryCapabilitiesparamreuseMemoryQueryEmbedding | NoneReturns
tuple[powercontext.builtin.artifacts.memory.models.MemoryUsedSearchMode, tuple[float, ...] | None, powercontext.builtin.artifacts.memory.models.MemoryQueryEmbedding | None, int]func_reranked_search_result(*, mode, query, hits, limit, admission=None, embedding_calls=0, query_embedding=None) -> MemorySearchResultApply the optional reranker and report what this search paid and admitted.
generation_calls counts the RFC 0080 rerank call this search actually issued: 1
when the reranker ran over a non-empty pool, 0 otherwise. It is never inferred
from configuration — a deployment whose rerank flag is on but whose reranker did not
run reports 0, which is the honest number.
parammodeMemoryUsedSearchModeparamquerystrparamhitstuple[MemoryHit, ...]paramlimitintparamadmissionAdmissionCounts | None= Noneparamembedding_callsint= 0paramquery_embeddingMemoryQueryEmbedding | None= NoneReturns
powercontext.builtin.artifacts.memory.models.MemorySearchResultfuncexpand(hits, *, layer='full') -> tuple[MemoryEntryVersion, ...]Expand exact hit anchors and reject cross-Revision substitutions.
paramhitsSequence[MemoryHit]paramlayerLiteral['full']= 'full'Returns
tuple[powercontext.builtin.artifacts.memory.models.MemoryEntryVersion, ...]funcentries(memory, /, *, tag_filter=None) -> tuple[MemoryEntryVersion, ...]Return the entry objects referenced by one exact current Memory head.
parammemoryMemoryparamtag_filterTagFilter | None= NoneReturns
tuple[powercontext.builtin.artifacts.memory.models.MemoryEntryVersion, ...]funcrebuild_projections(embedding_model=None) -> NoneRebuild current-head search projections from authoritative Memory revisions.
paramembedding_modelEmbeddingModel | None= NoneReturns
Nonefuncvalidate_citation(citation) -> MemoryEntryVersionResolve one exact Handoff citation.
paramcitationMemoryCitationReturns
powercontext.builtin.artifacts.memory.models.MemoryEntryVersionfunc_validate_search_heads(memories) -> Noneparammemoriestuple[Memory, ...]Returns
Nonefunc_select_search_mode(requested, *, memories, capabilities) -> MemoryUsedSearchModeparamrequestedMemorySearchModeparammemoriestuple[ArtifactRef, ...]paramcapabilitiesMemoryCapabilitiesReturns
powercontext.builtin.artifacts.memory.models.MemoryUsedSearchModefunc_deduplicate_manifest(manifest, current_by_entry) -> tuple[list[MemoryChange], set[str]]parammanifestdict[str, MemoryManifestEntry]paramcurrent_by_entrydict[str, MemoryEntryVersion]Returns
tuple[list[powercontext.builtin.artifacts.memory.models.MemoryChange], set[str]]func_normalize_manifest_entries(base, manifest, current_by_entry, *, skip) -> tuple[list[MemoryChange], list[MemoryEntryVersion]]parambaseMemoryparammanifestdict[str, MemoryManifestEntry]paramcurrent_by_entrydict[str, MemoryEntryVersion]paramskipset[str]Returns
tuple[list[powercontext.builtin.artifacts.memory.models.MemoryChange], list[powercontext.builtin.artifacts.memory.models.MemoryEntryVersion]]func_set_entry_state(memory, *, entries, target_state, reason) -> MemoryparammemoryMemoryparamentriesSequence[MemoryEntryVersion]paramtarget_stateLiteral['active', 'inactive']paramreasonstr | NoneReturns
powercontext.builtin.artifacts.memory.models.Memoryfunc_commit_existing_transition(*, base, manifest, changes, current_by_entry, entry_versions, growth=frozenset()) -> MemoryparambaseMemoryparammanifestdict[str, MemoryManifestEntry]paramchangesSequence[MemoryChange]paramcurrent_by_entrydict[str, MemoryEntryVersion]paramentry_versionstuple[MemoryEntryVersion, ...]paramgrowthfrozenset[MemoryCapacityDimension]= frozenset()Returns
powercontext.builtin.artifacts.memory.models.Memoryfunc_validate_anchor(*, memory_ref, entry_id, entry_version_id, version) -> Noneparammemory_refArtifactRefparamentry_idstrparamentry_version_idstrparamversionMemoryEntryVersionReturns
Nonefunc_prepare_projections(*, base, manifest_entries, versions_by_entry, changed_version_ids) -> tuple[MemoryProjection, ...]Build active-head projections, reusing unchanged vectors and embedding only changes.
parambaseMemory | Noneparammanifest_entriestuple[MemoryManifestEntry, ...]paramversions_by_entrydict[str, MemoryEntryVersion]paramchanged_version_idsfrozenset[str]Returns
tuple[powercontext.builtin.artifacts.memory.protocols.MemoryProjection, ...]func_previous_projections(base) -> dict[str, MemoryProjection]parambaseMemory | NoneReturns
dict[str, powercontext.builtin.artifacts.memory.protocols.MemoryProjection]func_cache_projections(committed, projections) -> NoneparamcommittedMemoryparamprojectionstuple[MemoryProjection, ...]Returns
Nonefunc_attach_embeddings(projections, embed_indices) -> tuple[MemoryProjection, ...]paramprojectionstuple[MemoryProjection, ...]paramembed_indicesSequence[int]Returns
tuple[powercontext.builtin.artifacts.memory.protocols.MemoryProjection, ...]func_embed_texts(texts, profile, *, query=False) -> tuple[EmbeddingVector, ...]paramtextstuple[str, ...]paramprofileEmbeddingProfileparamquerybool= FalseReturns
tuple[powercontext.builtin.inference.EmbeddingVector, ...]func_canonical_base(memory) -> Memory | NoneparammemoryMemory | NoneReturns
powercontext.builtin.artifacts.memory.models.Memory | Nonefunc_canonical_memory(memory) -> MemoryparammemoryMemoryReturns
powercontext.builtin.artifacts.memory.models.Memoryfunc_canonical_operation_evidence(sources, artifacts) -> _OperationEvidenceparamsourcesSequence[Source]paramartifactsSequence[Artifact[object]]Returns
powercontext.builtin.artifacts.memory.service._OperationEvidencefunc_validated_entries(memory) -> tuple[MemoryEntryVersion, ...]parammemoryMemoryReturns
tuple[powercontext.builtin.artifacts.memory.models.MemoryEntryVersion, ...]func_candidates(mode, entries, evidence, current_entries, *, active_version_ids) -> tuple[MemoryEntryInput, ...]parammodeLiteral['append', 'extract']paramentriestuple[MemoryEntryInput, ...]paramevidence_OperationEvidenceparamcurrent_entriestuple[MemoryEntryVersion, ...]paramactive_version_idsfrozenset[str]Returns
tuple[powercontext.builtin.artifacts.memory.models.MemoryEntryInput, ...]func_assess_write(base, candidates, evidence, current_entries) -> MemoryWriteAssessment | NoneAsk the configured gate about one candidate set; None means no gate is active.
parambaseMemory | Noneparamcandidatestuple[MemoryEntryInput, ...]paramevidence_OperationEvidenceparamcurrent_entriestuple[MemoryEntryVersion, ...] | NoneReturns
powercontext.builtin.artifacts.memory.protocols.MemoryWriteAssessment | Nonefunc_gate_evidence(base, candidates, evidence, current_entries) -> _GateEvidenceProjectionparambaseMemory | Noneparamcandidatestuple[MemoryEntryInput, ...]paramevidence_OperationEvidenceparamcurrent_entriestuple[MemoryEntryVersion, ...] | NoneReturns
powercontext.builtin.artifacts.memory.service._GateEvidenceProjectionfunc_gate_candidate_evidence(base, candidates, evidence, current_entries) -> tuple[_GateCandidateEvidence, ...]parambaseMemory | Noneparamcandidatestuple[MemoryEntryInput, ...]paramevidence_OperationEvidenceparamcurrent_entriestuple[MemoryEntryVersion, ...] | NoneReturns
tuple[powercontext.builtin.artifacts.memory.service._GateCandidateEvidence, ...]func_candidate_gate_evidence(candidate) -> tuple[_GateEvidenceEntry, ...]paramcandidate_GateCandidateEvidenceReturns
tuple[powercontext.builtin.artifacts.memory.service._GateEvidenceEntry, ...]func_source_gate_evidence(candidate_index, source) -> _GateEvidenceEntryparamcandidate_indexintparamsourceSourceReturns
powercontext.builtin.artifacts.memory.service._GateEvidenceEntryfunc_artifact_gate_evidence(candidate_index, artifact) -> _GateEvidenceEntryparamcandidate_indexintparamartifactArtifact[object]Returns
powercontext.builtin.artifacts.memory.service._GateEvidenceEntryfunc_source_ref_gate_evidence(candidate_index, ref) -> _GateEvidenceEntryparamcandidate_indexintparamrefSourceRefReturns
powercontext.builtin.artifacts.memory.service._GateEvidenceEntryfunc_artifact_ref_gate_evidence(candidate_index, ref) -> _GateEvidenceEntryparamcandidate_indexintparamrefArtifactRefReturns
powercontext.builtin.artifacts.memory.service._GateEvidenceEntryfunc_write_gate_policy_id() -> strReturns
strfunc_prepare_commit(*, base, candidates, evidence, current_entries, flagged_reason=None) -> MemoryCommit | NoneparambaseMemory | Noneparamcandidatestuple[MemoryEntryInput, ...]paramevidence_OperationEvidenceparamcurrent_entriestuple[MemoryEntryVersion, ...] | Noneparamflagged_reasonstr | None= NoneReturns
powercontext.builtin.artifacts.memory.protocols.MemoryCommit | Nonefunc_claim_revision_target(candidate, base, current_by_entry, targeted) -> tuple[str, MemoryEntryVersion]paramcandidateMemoryEntryInputparambaseMemory | Noneparamcurrent_by_entrydict[str, MemoryEntryVersion]paramtargetedset[str]Returns
tuple[str, powercontext.builtin.artifacts.memory.models.MemoryEntryVersion]func_material_from_candidate(candidate, allowed_sources, allowed_artifacts, *, previous=None) -> _EntryMaterialparamcandidateMemoryEntryInputparamallowed_sourcesSequence[Source]paramallowed_artifactsSequence[Artifact[object]]parampreviousMemoryEntryVersion | None= NoneReturns
powercontext.builtin.artifacts.memory.service._EntryMaterialfunc_canonical_candidate_sources(values, allowed) -> tuple[Source, ...]paramvaluesSequence[Source]paramallowedSequence[Source]Returns
tuple[powercontext.sources.Source, ...]func_canonical_candidate_artifacts(values, allowed, previous) -> tuple[ArtifactRef, ...]paramvaluesSequence[Artifact[object]]paramallowedSequence[Artifact[object]]parampreviousSequence[ArtifactRef]Returns
tuple[powercontext.artifacts.ArtifactRef, ...]func_material_from_version(version) -> _EntryMaterialparamversionMemoryEntryVersionReturns
powercontext.builtin.artifacts.memory.service._EntryMaterialfunc_entry_material(*, kind, text, sources, artifacts) -> _EntryMaterialparamkindstrparamtextstrparamsourcesSequence[SourceRef]paramartifactsSequence[ArtifactRef]Returns
powercontext.builtin.artifacts.memory.service._EntryMaterialfunc_source_refs(sources) -> tuple[SourceRef, ...]paramsourcesSequence[Source]Returns
tuple[powercontext.sources.SourceRef, ...]func_new_entry_version(*, memory_id, entry_id, previous, material, created_in_revision) -> MemoryEntryVersionparammemory_idstrparamentry_idstrparampreviousMemoryEntryVersion | Noneparammaterial_EntryMaterialparamcreated_in_revisionintReturns
powercontext.builtin.artifacts.memory.models.MemoryEntryVersionfunc_new_id(kind) -> strparamkindstrReturns
str
