API Reference¶

class Artifacts(BaseModel):
    """Compose shared Artifact persistence and read operations."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    catalog: ArtifactCatalog[Artifact[object]]
    store: ArtifactStore[ArtifactDraft[object], Artifact[object]]

    async def add(self, draft: ArtifactDraft[object], /) -> Artifact[object]:
        """Commit an initial Artifact Revision without semantic generation."""

        return await self.store.add(draft)

    async def revise(
        self,
        artifact: Artifact[object],
        draft: ArtifactDraft[object],
        /,
    ) -> Artifact[object]:
        """Commit a Revision using ``artifact`` as the optimistic base."""

        if artifact.family != draft.family:
            raise ArtifactFamilyMismatchError(artifact, draft)
        return await self.store.revise(artifact, draft)

    async def get(self, artifact: Artifact[object], /) -> Artifact[object]:
        """Return the canonical exact Revision matching ``artifact``."""

        return await self.catalog.get(artifact)

    async def latest(self, artifact: Artifact[object], /) -> Artifact[object]:
        """Return the latest visible Revision of ``artifact``."""

        return await self.catalog.latest(artifact)

    async def revisions(self, artifact: Artifact[object], /) -> tuple[Artifact[object], ...]:
        """Return the visible history of ``artifact``."""

        return await self.catalog.revisions(artifact)

class PowerContext(BaseModel, Generic[SourcesT, ArtifactsT, TriggersT]):
    """Bind three explicitly selected component groups without owning their lifecycle."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    sources: SourcesT
    artifacts: ArtifactsT
    triggers: TriggersT

class Sources(BaseModel):
    """Compose Source acquisition, persistence, and read operations."""

    model_config = ConfigDict(arbitrary_types_allowed=True)

    catalog: SourceCatalog
    store: SourceStore[Source]

    async def resolve(self, value: object, /) -> Source:
        """Resolve an adapter-native input without persisting it."""

        return await self.catalog.resolve(value)

    async def add(self, source: Source, /) -> Source:
        """Persist a resolved Source without deriving Artifacts."""

        self.catalog.as_ref(source)
        stored = await self.store.add(source)
        self.catalog.as_ref(stored)
        return stored

    async def get(self, source: Source, /) -> Source:
        """Return the canonical persisted Source matching ``source``."""

        return await self.catalog.get(source)

    async def list(self) -> tuple[Source, ...]:
        """Return the Sources visible in the current catalog view."""

        return await self.catalog.list()

    async def read(self, source: Source, /) -> object:
        """Read the adapter-native value described by ``source``."""

        return await self.catalog.read(source)

class Source(BaseModel):
    """Base value for an adapter-owned Source description."""

    name: str
    materialization: SourceMaterialization
    description: str | None = None

class SourceAdapter(Protocol[InputT, SourceT, ValueT_co]):
    """Resolve one exact input class and read one concrete Source class."""

    input_class: type[InputT]
    """The exact input class accepted by this adapter."""

    name: str
    """The stable registration name for this adapter."""

    source_class: type[SourceT]
    """The exact Source class produced and read by this adapter."""

    async def resolve(self, value: InputT, /) -> SourceT:
        """Resolve an adapter-native value without changing a catalog."""

        ...

    async def read(self, source: SourceT, /) -> ValueT_co:
        """Read the adapter-native value described by a Source."""

        ...

class SourceCatalog:
    """A read-only Source catalog routed by actual adapters."""

    def __init__(
        self,
        *,
        backend: SourceCatalogBackend,
        adapters: Iterable[_AnySourceAdapter],
    ) -> None:
        by_input: dict[type[object], _AnySourceAdapter] = {}
        by_source: dict[type[Source], _AnySourceAdapter] = {}
        for adapter in adapters:
            input_class, source_class = _validate_adapter(adapter)
            if input_class in by_input:
                raise SourceConflictError("input_class", input_class)
            if source_class in by_source:
                raise SourceConflictError("source_class", source_class)
            by_input[input_class] = adapter
            by_source[source_class] = adapter
        self._backend = backend
        self._by_input = by_input
        self._by_source = by_source

    async def list(self) -> tuple[Source, ...]:
        sources = await self._backend.list()
        for source in sources:
            self.as_ref(source)
        return sources

    async def get(self, source: Source, /) -> Source:
        self.as_ref(source)
        stored = await self._backend.get(source)
        self.as_ref(stored)
        if type(stored) is not type(source) or stored != source:
            raise SourceNotFoundError(source)
        return stored

    def as_ref(self, source: Source, /) -> SourceRef:
        adapter = _adapter_for_source(source, self._by_source)
        return SourceRef(source_type=adapter.name, source_id=source.name)

    async def resolve(self, value: object, /) -> Source:
        input_class = type(value)
        try:
            adapter = self._by_input[input_class]
        except KeyError:
            raise SourceAdapterNotFoundError("input", input_class) from None
        source = await adapter.resolve(value)
        if type(source) is not adapter.source_class:
            raise InvalidSourceResultError(adapter.name, "resolve", adapter.source_class, type(source))
        self.as_ref(source)
        return cast(Source, source)

    async def read(self, source: Source, /) -> object:
        adapter = _adapter_for_source(source, self._by_source)
        return await adapter.read(source)

@runtime_checkable
class SourceCatalogBackend(Protocol):
    """Provide the Source reads required by SourceCatalog."""

    async def get(self, source: Source, /) -> Source: ...

    async def list(self) -> tuple[Source, ...]:
        """Return the Sources visible in one backend view."""

        ...

class SourceMaterialization(StrEnum):
    """Describe where the value read for a Source comes from."""

    CAPTURED = "captured"
    REFERENCED = "referenced"

class SourceRef(BaseModel):
    """A stable reference to one Source in the current catalog view."""

    source_type: str
    source_id: str

    @field_validator("source_type", "source_id")
    @classmethod
    def validate_reference_part(cls, value: str, info) -> str:
        _validate_reference_part(info.field_name, value)
        return value

@runtime_checkable
class SourceStore(Protocol[SourceT]):
    """Persist resolved Sources for later catalog reads and lineage."""

    async def add(self, value: SourceT, /) -> SourceT: ...

class Artifact(BaseModel, Generic[ContentT]):
    """An immutable snapshot in an artifact lifecycle."""

    family: ClassVar[str] = "artifact"

    artifact_id: str
    revision: StrictInt = Field(ge=1)
    content: ContentT
    lineage: ArtifactLineage = Field(default_factory=ArtifactLineage)

    @field_validator("artifact_id")
    @classmethod
    def validate_artifact_id(cls, value: str) -> str:
        _validate_reference_part("artifact_id", value)
        if len(value) > MAX_ARTIFACT_ID_LENGTH:
            raise InvalidArtifactReferenceError(
                "artifact_id",
                f"must not exceed {MAX_ARTIFACT_ID_LENGTH} characters",
            )
        return value

    def as_ref(self) -> ArtifactRef:
        """Return an exact reference to this revision."""

        return ArtifactRef(family=self.family, artifact_id=self.artifact_id, revision=self.revision)

@runtime_checkable
class ArtifactCatalog(Protocol[ArtifactT]):
    """Read artifact revisions without owning their writes."""

    async def get(self, artifact: ArtifactT, /) -> ArtifactT:
        """Return the canonical exact revision matching ``artifact``."""

        ...

    async def latest(self, artifact: ArtifactT, /) -> ArtifactT:
        """Return the latest visible revision of ``artifact``."""

        ...

    async def revisions(self, artifact: ArtifactT, /) -> tuple[ArtifactT, ...]:
        """Return the visible history of ``artifact`` in ascending revision order."""

        ...

class ArtifactDraft(BaseModel, Generic[ContentT]):
    """Content and complete evidence supplied for one Artifact write."""

    family: ClassVar[str] = "artifact"

    content: ContentT
    sources: tuple[SourceRef, ...] = ()
    artifacts: tuple[ArtifactRef, ...] = ()

    @model_validator(mode="after")
    def validate_family(self):
        _validate_reference_part("family", self.family)
        return self

class ArtifactLineage(BaseModel):
    """The direct evidence used to produce one artifact revision."""

    sources: tuple[SourceRef, ...] = ()
    artifacts: tuple[ArtifactRef, ...] = ()

class ArtifactRef(BaseModel):
    """A stable reference to one exact artifact revision."""

    family: str
    artifact_id: str
    revision: StrictInt = Field(ge=1)

    @field_validator("family", "artifact_id")
    @classmethod
    def validate_identity(cls, value: str, info) -> str:
        _validate_reference_part(info.field_name, value)
        maximum = MAX_ARTIFACT_FAMILY_LENGTH if info.field_name == "family" else MAX_ARTIFACT_ID_LENGTH
        if len(value) > maximum:
            raise InvalidArtifactReferenceError(info.field_name, f"must not exceed {maximum} characters")
        return value

@runtime_checkable
class ArtifactStore(Protocol[DraftT_contra, ArtifactT]):
    """Commit new Artifact revisions from complete domain objects."""

    async def add(self, draft: DraftT_contra, /) -> ArtifactT:
        """Commit the first revision represented by ``draft``."""

        ...

    async def revise(self, artifact: ArtifactT, draft: DraftT_contra, /) -> ArtifactT:
        """Commit ``draft`` only if ``artifact`` remains the latest revision."""

        ...

class CandidatePipeline(Protocol):
    """Produce untrusted Memory candidates from canonical bounded evidence."""

    async def extract(self, request: MemoryCandidateRequest, /) -> tuple[MemoryEntryInput, ...]:
        """Return candidates that still require all service validations."""

        ...

class DefaultMemoryEvidenceProjector:
    """Expose only stable public evidence metadata unless a caller opts in to more."""

    def project_source(self, source: Source, /) -> JsonValue:
        return _validated_json({
            "name": source.name,
            "materialization": source.materialization.value,
            "description": source.description,
        })

    def project_artifact(self, artifact: Artifact[object], /) -> JsonValue:
        return _validated_json({
            "artifact_id": artifact.artifact_id,
            "revision": artifact.revision,
            "family": artifact.family,
            "content": artifact.content,
        })

class EmbeddingProfile(BaseModel):
    """The Memory embedding index contract for one deployment."""

    profile_id: str
    model: str
    dimension: int
    distance: Literal["l2"] = "l2"
    normalization: Literal["none", "unit"] = "unit"

class LLMMemoryCandidatePipeline:
    """Map schema-valid model proposals back to exact bounded domain values."""

    def __init__(
        self,
        generator: StructuredGenerator[MemoryExtractionInput, MemoryExtractionOutput],
        *,
        evidence_projector: MemoryEvidenceProjector | None = None,
    ) -> None:
        self._generator = generator
        self._evidence_projector = (
            DefaultMemoryEvidenceProjector() if evidence_projector is None else evidence_projector
        )

    async def extract(self, request: MemoryCandidateRequest, /) -> tuple[MemoryEntryInput, ...]:
        """Generate and map candidates that still require MemoryService validation."""

        extraction_input, evidence = _extraction_input(request, self._evidence_projector)
        result = await self._generator.generate(extraction_input)
        output = _validated_output(result)
        current_entries = {entry.entry_id: entry for entry in request.current_entries}
        revised_entries: set[str] = set()
        candidates: list[MemoryEntryInput] = []
        for proposal in output.candidates:
            selected_sources, selected_artifacts = _selected_evidence(proposal, evidence)
            entry = _revision_target(proposal, current_entries, revised_entries)
            candidates.append(
                MemoryEntryInput(
                    entry=entry,
                    kind=proposal.kind,
                    text=proposal.text,
                    sources=selected_sources,
                    artifacts=selected_artifacts,
                    reason=proposal.reason,
                )
            )
        return tuple(candidates)

class LLMMemoryReranker:
    """Use one structured listwise generation request to select Memory hits."""

    policy_id = MEMORY_RERANK_INSTRUCTIONS_VERSION

    def __init__(
        self,
        generator: StructuredGenerator[MemoryRerankInput, MemoryRerankOutput],
        /,
    ) -> None:
        self._generator = generator

    async def rerank(
        self,
        query: str,
        candidates: tuple[MemoryHit, ...],
        limit: int,
        /,
    ) -> MemoryRerankDecision:
        """Generate and normalize a sparse selection from the coarse ranks."""

        _validate_bounds(len(candidates), limit)
        result = await self._generator.generate(
            MemoryRerankInput(
                query=query,
                max_results=limit,
                candidates=tuple(
                    MemoryRerankCandidate(rank=rank, text=candidate.text)
                    for rank, candidate in enumerate(candidates, start=1)
                ),
            )
        )
        selected_ranks, discarded_rank_count = _normalize_ranks(
            result.output.selected_ranks,
            candidate_count=len(candidates),
            limit=limit,
        )
        if selected_ranks:
            return MemoryRerankDecision(
                selected_ranks=selected_ranks,
                usage=result.usage,
                discarded_rank_count=discarded_rank_count,
            )
        return MemoryRerankDecision(
            selected_ranks=tuple(range(1, limit + 1)),
            usage=result.usage,
            discarded_rank_count=discarded_rank_count,
            used_fallback=True,
        )

class Memory(Artifact[MemoryContent]):
    """An immutable snapshot in a Memory lifecycle."""

    family: ClassVar[str] = "memory"

class MemoryBackend(Protocol):
    """Storage and retrieval capabilities required by the Memory Family."""

    async def capabilities(self) -> MemoryCapabilities:
        """Return probed deployment capabilities."""

        ...

    async def get(self, memory: ArtifactRef, /) -> Memory:
        """Load one exact Memory Revision."""

        ...

    async def latest(self, artifact_id: str, /) -> Memory:
        """Load the current head of one Memory identity."""

        ...

    async def entries(self, memory: ArtifactRef, /) -> tuple[MemoryEntryVersion, ...]:
        """Load the entry versions referenced by an exact manifest."""

        ...

    async def projections(self, memory: ArtifactRef, /) -> tuple[MemoryProjection, ...]:
        """Load rebuildable active-head projections for an exact Memory Revision."""

        ...

    async def rebuild_projections(self, embedding_model: EmbeddingModel | None = None, /) -> None:
        """Rebuild active-head and search projections from authoritative revisions."""

        ...

    def begin(self) -> AbstractAsyncContextManager[MemoryUnitOfWork]:
        """Open the adapter-specific atomic write boundary."""

        ...

    async def changes(
        self,
        memory: ArtifactRef,
        since_revision: int | None,
        /,
    ) -> tuple[MemoryRevisionChanges, ...]:
        """Read compact Revision changes without entry bodies."""

        ...

    async def vector_complete(
        self,
        memories: tuple[ArtifactRef, ...],
        profile: EmbeddingProfile,
        /,
    ) -> bool:
        """Derive fixed-profile vector completeness for selected heads."""

        ...

    async def search(self, request: MemorySearchRequest, /) -> MemorySearchChannels:
        """Return backend-ordered FTS/vector channels after manifest checks."""

        ...

    async def expand(self, hits: tuple[MemoryHit, ...], /) -> tuple[MemoryEntryVersion, ...]:
        """Load and validate exact versions anchored by hits."""

        ...

class MemoryBackendConfigurationError(MemoryLayerError, RuntimeError):
    """Raised when a repository cannot satisfy its declared configuration."""

class MemoryCandidateRequest(BaseModel):
    """Canonical evidence and bounded current entries offered to a pipeline."""

    sources: tuple[Source, ...]
    artifacts: tuple[Artifact[object], ...]
    current_entries: tuple[MemoryEntryVersion, ...]

class MemoryCapabilities(BaseModel):
    """Backend features available for the configured deployment."""

    fts: bool
    vector: bool = False
    hybrid: bool = False
    embedding_profile: EmbeddingProfile | None = None

class MemoryChange(BaseModel):
    """A compact entry change recorded by one Memory Revision."""

    op: MemoryChangeOp
    entry_id: str
    from_entry_version_id: str | None
    to_entry_version_id: str | None
    reason: str | None = None

class MemoryChannelHit(BaseModel):
    """An identity-preserving candidate returned by one search channel."""

    memory_ref: ArtifactRef
    entry_id: str
    entry_version_id: str
    text: str
    distance: float | None = Field(default=None, ge=0.0, allow_inf_nan=False)

class MemoryCitation(BaseModel):
    """A stable Handoff anchor for one exact entry version."""

    memory_ref: ArtifactRef
    entry_id: str
    entry_version_id: str

class MemoryCommit(BaseModel):
    """A complete Memory Revision and every row changed atomically with it."""

    base: Memory | None
    memory: Memory
    content_hash: str
    entry_versions: tuple[MemoryEntryVersion, ...]
    projections: tuple[MemoryProjection, ...]

class MemoryContent(BaseModel):
    """The complete canonical content of one Memory Artifact Revision."""

    manifest: MemoryManifest
    changes: tuple[MemoryChange, ...] = ()
    schema_version: Literal["powercontext.memory.v1"] = Field(
        default="powercontext.memory.v1",
        alias="schema",
    )

class MemoryEntryInput(BaseModel):
    """An untrusted proposed entry addition or content revision."""

    kind: str
    text: str
    entry: MemoryEntryVersion | None = None
    sources: tuple[Source, ...] = ()
    artifacts: tuple[Artifact[object], ...] = ()
    reason: str | None = None

class MemoryEntryVersion(BaseModel):
    """One immutable version of a logical Memory entry."""

    memory_artifact_id: str
    entry_id: str
    entry_version_id: str
    version: int
    previous_version_id: str | None
    kind: str
    text: str
    entry_content_hash: str
    created_in_revision: int
    sources: tuple[SourceRef, ...] = ()
    artifacts: tuple[ArtifactRef, ...] = ()

class MemoryEvidenceProjector(Protocol):
    """Project adapter-owned evidence values into explicit JSON-visible content."""

    def project_source(self, source: Source, /) -> JsonValue:
        """Return the stable JSON projection exposed to the model."""

        ...

    def project_artifact(self, artifact: Artifact[object], /) -> JsonValue:
        """Return the stable JSON projection exposed to the model."""

        ...

class MemoryExtractionCandidate(BaseModel):
    """One model-proposed addition or revision with operation-local citations."""

    intent: MemoryExtractionIntent
    kind: KnownMemoryEntryKind
    text: str
    evidence_ids: tuple[str, ...]
    entry_id: str | None = None
    reason: str | None = None

class MemoryExtractionCurrentEntry(BaseModel):
    """One active entry from the selected Memory head."""

    entry_id: str
    kind: str
    text: str

class MemoryExtractionEvidence(BaseModel):
    """One operation-local evidence value visible to the generator."""

    evidence_id: str
    evidence_type: MemoryEvidenceType
    content: JsonValue

class MemoryExtractionInput(BaseModel):
    """Only bounded operation evidence and active current entries."""

    evidence: tuple[MemoryExtractionEvidence, ...]
    current_entries: tuple[MemoryExtractionCurrentEntry, ...]

class MemoryExtractionOutput(BaseModel):
    """Schema-bound Memory extraction output; an empty tuple is a no-op."""

    candidates: tuple[MemoryExtractionCandidate, ...] = ()

class MemoryExtractionProfile(StrEnum):
    """Stable built-in policies for selecting Memory from bounded evidence."""

    CODING = "coding"
    CONVERSATION = "conversation"

class MemoryHit(BaseModel):
    """A fused retrieval result anchored to exact Memory content."""

    memory_ref: ArtifactRef
    entry_id: str
    entry_version_id: str
    text: str
    score: float
    matched_by: tuple[MemoryMatchedBy, ...]

class MemoryLayerError(PowerContextError):
    """Base exception for Memory domain and repository failures."""

class MemoryManifest(BaseModel):
    """The authoritative directory for one Memory Revision."""

    entries: tuple[MemoryManifestEntry, ...] = ()
    format: Literal["flat-v1"] = "flat-v1"

class MemoryManifestEntry(BaseModel):
    """One logical entry pointer and state in an immutable Revision."""

    entry_id: str
    entry_version_id: str
    entry_content_hash: str
    state: MemoryEntryState

class MemoryProjection(BaseModel):
    """A rebuildable active-head projection prepared outside a transaction."""

    entry_version: MemoryEntryVersion
    searchable_text: str
    embedding: EmbeddingVector | None = None
    embedding_content_hash: str | None = None

class MemoryRerankCandidate(_StrictModel):
    """One identity-preserving coarse candidate exposed to the model."""

    rank: int = Field(ge=1)
    text: str = Field(min_length=1)

@dataclass(frozen=True, slots=True)
class MemoryRerankDecision:
    """A validated sparse selection and its portable inference metadata."""

    selected_ranks: tuple[int, ...]
    usage: InferenceUsage
    discarded_rank_count: int = 0
    used_fallback: bool = False

class MemoryRerankInput(_StrictModel):
    """One query and its bounded listwise candidate pool."""

    query: str = Field(min_length=1)
    max_results: int = Field(ge=1)
    candidates: tuple[MemoryRerankCandidate, ...]

class MemoryRerankMode(StrEnum):
    """Deployment-selectable Memory reranking policies."""

    NONE = "none"
    LLM = "llm"

class MemoryRerankOutput(_StrictModel):
    """Original coarse ranks selected in descending answer utility."""

    selected_ranks: tuple[int, ...]

class MemoryRerankTrace(BaseModel):
    """Observable listwise selection over one coarse retrieval pool."""

    policy_id: str = Field(min_length=1)
    candidate_hits: tuple[MemoryHit, ...]
    selected_ranks: tuple[int, ...]
    discarded_rank_count: int = 0
    used_fallback: bool = False
    latency_ms: float = Field(ge=0.0, allow_inf_nan=False)
    usage: InferenceUsage

class MemoryReranker(Protocol):
    """Select final Memory hits from one already ordered coarse pool."""

    policy_id: str

    async def rerank(
        self,
        query: str,
        candidates: tuple[MemoryHit, ...],
        limit: int,
        /,
    ) -> MemoryRerankDecision:
        """Return validated original ranks without changing hit identity."""

        ...

class MemoryRevisionChanges(BaseModel):
    """The compact changes stored by one exact Memory Revision."""

    memory_ref: ArtifactRef
    changes: tuple[MemoryChange, ...]

class MemorySearchChannels(BaseModel):
    """Backend-internal channel rankings before shared fusion."""

    fts: tuple[MemoryChannelHit, ...] = ()
    vector: tuple[MemoryChannelHit, ...] = ()

class MemorySearchRequest(BaseModel):
    """A fully validated backend search request for explicit current heads."""

    query: str
    analyzed_query: str
    memories: tuple[ArtifactRef, ...]
    candidate_limit: int
    mode: MemoryUsedSearchMode
    query_vector: EmbeddingVector | None = None
    embedding_profile: EmbeddingProfile | None = None

class MemorySearchResult(BaseModel):
    """Search hits together with the mode actually executed."""

    mode: MemoryUsedSearchMode
    hits: tuple[MemoryHit, ...] = ()
    rerank: MemoryRerankTrace | None = None