ontolith.store

The StorageBackend port, plus the two shipped adapters.

ontolith.store

Storage layer and backend abstractions.

DEFAULT_NAMESPACE module-attribute

DEFAULT_NAMESPACE = 'default'

The one namespace this project operates in today (KI-022).

Ontolith is still single-namespace throughout (ADR-0015) — Ontology always writes to this namespace, and both backends seed a matching namespace registry row for it at schema-creation time. Shared by Ontology and both backends so that specific trio stays in sync by construction; a handful of other unrelated "default" literals elsewhere (e.g. REST/MCP route defaults, example scripts) are independent naming choices, not instances of this constant, and aren't required to match it.

VECTOR_SCOPES module-attribute

VECTOR_SCOPES = frozenset({'entity', 'assertion'})

Closed set of embedding scopes (SPEC §11.3): entity- and assertion-level.

Deliberately not an arbitrary caller-supplied string — both backends use scope to name per-scope storage (SQLite: a vec0 virtual table per scope; DuckDB: a plain table per scope), so an open string would mean dynamic DDL driven by caller input. vector_upsert/vector_search MUST reject any scope outside this set with ValidationError.

StorageBackend

Bases: Protocol

Abstract port for storage adapters.

Concrete backends implement this protocol to provide: - Transaction management - Entity and assertion persistence - Query execution - Vector search (for hybrid retrieval)

The default implementation (M1) is SQLite + sqlite-vec. Alternative backends can be plugged in via this interface.

begin

begin() -> None

Begin a new transaction.

This port makes no promise about when a concurrent writer is serialized against this one (at begin() versus at the first conflicting statement) — that's a backend-specific locking detail, not a cross-backend contract. See SQLiteBackend.begin()'s own docstring (KI-084) for the default backend's specific choice and why it matters for cross-process write safety.

Source code in src/ontolith/store/base.py

def begin(self) -> None:
    """Begin a new transaction.

    This port makes no promise about *when* a concurrent writer is
    serialized against this one (at `begin()` versus at the first
    conflicting statement) — that's a backend-specific locking detail,
    not a cross-backend contract. See `SQLiteBackend.begin()`'s own
    docstring (KI-084) for the default backend's specific choice and
    why it matters for cross-process write safety.
    """
    ...

commit

commit() -> None

Commit the current transaction.

Source code in src/ontolith/store/base.py

def commit(self) -> None:
    """Commit the current transaction."""
    ...

rollback

rollback() -> None

Rollback the current transaction.

Source code in src/ontolith/store/base.py

def rollback(self) -> None:
    """Rollback the current transaction."""
    ...

transaction

transaction() -> AbstractContextManager[None]

Context manager for atomic multi-write transactions.

Guarantees rollback on any exception. Prefer this over manual begin/commit/rollback to avoid wedged connections.

Source code in src/ontolith/store/base.py

def transaction(self) -> AbstractContextManager[None]:
    """Context manager for atomic multi-write transactions.

    Guarantees rollback on any exception. Prefer this over
    manual begin/commit/rollback to avoid wedged connections.
    """
    ...

put_principal

put_principal(principal: Principal) -> None

Persist a principal.

Parameters:

Name	Type	Description	Default
principal	Principal	Principal to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_principal(self, principal: Principal) -> None:
    """Persist a principal.

    Args:
        principal: Principal to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_principal

get_principal(principal_id: str) -> Principal | None

Retrieve a principal by ID.

Parameters:

Name	Type	Description	Default
principal_id	str	Principal ID to retrieve	required

Returns:

Type	Description
Principal | None	Principal if found, None otherwise

Source code in src/ontolith/store/base.py

def get_principal(self, principal_id: str) -> Principal | None:
    """Retrieve a principal by ID.

    Args:
        principal_id: Principal ID to retrieve

    Returns:
        Principal if found, None otherwise
    """
    ...

put_credential

put_credential(credential: PrincipalCredential) -> None

Persist a principal credential (hashed API-key token, ADR-0014).

Parameters:

Name	Type	Description	Default
credential	PrincipalCredential	PrincipalCredential to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_credential(self, credential: PrincipalCredential) -> None:
    """Persist a principal credential (hashed API-key token, ADR-0014).

    Args:
        credential: PrincipalCredential to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_principal_by_token_hash

get_principal_by_token_hash(
    token_hash: str,
) -> Principal | None

Resolve a principal via a credential's token hash.

Only unrevoked credentials resolve.

Parameters:

Name	Type	Description	Default
token_hash	str	SHA-256 hash of the raw bearer token	required

Returns:

Type	Description
Principal | None	Principal if the hash matches an active credential, None otherwise

Source code in src/ontolith/store/base.py

def get_principal_by_token_hash(self, token_hash: str) -> Principal | None:
    """Resolve a principal via a credential's token hash.

    Only unrevoked credentials resolve.

    Args:
        token_hash: SHA-256 hash of the raw bearer token

    Returns:
        Principal if the hash matches an active credential, None otherwise
    """
    ...

get_credential

get_credential(
    credential_id: str,
) -> PrincipalCredential | None

Retrieve a credential by ID.

Parameters:

Name	Type	Description	Default
credential_id	str	Credential ID to retrieve	required

Returns:

Type	Description
PrincipalCredential | None	PrincipalCredential if found, None otherwise

Source code in src/ontolith/store/base.py

def get_credential(self, credential_id: str) -> PrincipalCredential | None:
    """Retrieve a credential by ID.

    Args:
        credential_id: Credential ID to retrieve

    Returns:
        PrincipalCredential if found, None otherwise
    """
    ...

list_principals

list_principals() -> list[Principal]

List all principals (KI-022).

Returns:

Type	Description
list[Principal]	All principals, most recently created first

Source code in src/ontolith/store/base.py

def list_principals(self) -> list[Principal]:
    """List all principals (KI-022).

    Returns:
        All principals, most recently created first
    """
    ...

get_credentials_for_principal

get_credentials_for_principal(
    principal_id: str,
) -> list[PrincipalCredential]

List all credentials (active and revoked) issued to a principal.

Never exposes the raw token — only credential metadata (id, created_at, revoked_at). Used to discover a credential ID to revoke.

Parameters:

Name	Type	Description	Default
principal_id	str	Principal to list credentials for	required

Returns:

Type	Description
list[PrincipalCredential]	Credentials for this principal, most recently issued first

Source code in src/ontolith/store/base.py

def get_credentials_for_principal(self, principal_id: str) -> list[PrincipalCredential]:
    """List all credentials (active and revoked) issued to a principal.

    Never exposes the raw token — only credential metadata (id,
    created_at, revoked_at). Used to discover a credential ID to revoke.

    Args:
        principal_id: Principal to list credentials for

    Returns:
        Credentials for this principal, most recently issued first
    """
    ...

revoke_credential

revoke_credential(
    credential_id: str,
    revoked_at: datetime,
    revoked_by: str,
) -> None

Mark a credential as revoked.

Parameters:

Name	Type	Description	Default
credential_id	str	Credential to revoke	required
revoked_at	datetime	Timestamp of revocation	required
revoked_by	str	Principal ID of the admin performing the revocation (KI-060)	required

Raises:

Type	Description
StorageError	If the credential is not found

Source code in src/ontolith/store/base.py

def revoke_credential(self, credential_id: str, revoked_at: datetime, revoked_by: str) -> None:
    """Mark a credential as revoked.

    Args:
        credential_id: Credential to revoke
        revoked_at: Timestamp of revocation
        revoked_by: Principal ID of the admin performing the
            revocation (KI-060)

    Raises:
        StorageError: If the credential is not found
    """
    ...

put_admin_event

put_admin_event(event: AdminEvent) -> None

Persist an append-only admin-action event (KI-060, SPEC §17).

Parameters:

Name	Type	Description	Default
event	AdminEvent	AdminEvent to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_admin_event(self, event: AdminEvent) -> None:
    """Persist an append-only admin-action event (KI-060, SPEC §17).

    Args:
        event: AdminEvent to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_admin_events

get_admin_events(
    actor: str | None = None, target: str | None = None
) -> list[AdminEvent]

Retrieve admin events, optionally filtered by actor or target.

Parameters:

Name	Type	Description	Default
actor	str | None	Filter to events performed by this principal ID	None
target	str | None	Filter to events against this target	None

Returns:

Type	Description
list[AdminEvent]	Matching events, oldest first

Source code in src/ontolith/store/base.py

def get_admin_events(
    self, actor: str | None = None, target: str | None = None
) -> list[AdminEvent]:
    """Retrieve admin events, optionally filtered by actor or target.

    Args:
        actor: Filter to events performed by this principal ID
        target: Filter to events against this target

    Returns:
        Matching events, oldest first
    """
    ...

put_entity

put_entity(entity: Entity) -> None

Persist an entity.

Parameters:

Name	Type	Description	Default
entity	Entity	Entity to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_entity(self, entity: Entity) -> None:
    """Persist an entity.

    Args:
        entity: Entity to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

put_assertion

put_assertion(assertion: Assertion) -> None

Persist an assertion.

Parameters:

Name	Type	Description	Default
assertion	Assertion	Assertion to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_assertion(self, assertion: Assertion) -> None:
    """Persist an assertion.

    Args:
        assertion: Assertion to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_entity

get_entity(entity_id: str) -> Entity | None

Retrieve an entity by ID.

Parameters:

Name	Type	Description	Default
entity_id	str	Entity ID to retrieve	required

Returns:

Type	Description
Entity | None	Entity if found, None otherwise

Source code in src/ontolith/store/base.py

def get_entity(self, entity_id: str) -> Entity | None:
    """Retrieve an entity by ID.

    Args:
        entity_id: Entity ID to retrieve

    Returns:
        Entity if found, None otherwise
    """
    ...

get_entity_by_natural_key

get_entity_by_natural_key(
    namespace: str, concept: str, natural_key: str
) -> Entity | None

Retrieve an entity by its unique (namespace, concept, natural_key) triple (KI-091) — the same uniqueness the entity table's own UNIQUE(namespace, concept, natural_key) constraint enforces, used to pre-check a conflict before put_entity rather than surfacing one late as a redacted StorageError.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to search within	required
concept	str	Concept name	required
natural_key	str	Natural key to look up	required

Returns:

Type	Description
Entity | None	Entity if one with this exact triple exists, None otherwise

Source code in src/ontolith/store/base.py

def get_entity_by_natural_key(
    self, namespace: str, concept: str, natural_key: str
) -> Entity | None:
    """Retrieve an entity by its unique `(namespace, concept, natural_key)`
    triple (KI-091) — the same uniqueness the `entity` table's own
    `UNIQUE(namespace, concept, natural_key)` constraint enforces, used
    to pre-check a conflict before `put_entity` rather than surfacing
    one late as a redacted `StorageError`.

    Args:
        namespace: Namespace to search within
        concept: Concept name
        natural_key: Natural key to look up

    Returns:
        Entity if one with this exact triple exists, None otherwise
    """
    ...

get_assertion

get_assertion(assertion_id: str) -> Assertion | None

Retrieve a single assertion by ID, regardless of status.

Parameters:

Name	Type	Description	Default
assertion_id	str	Assertion ID to retrieve	required

Returns:

Type	Description
Assertion | None	Assertion if found, None otherwise

Source code in src/ontolith/store/base.py

def get_assertion(self, assertion_id: str) -> Assertion | None:
    """Retrieve a single assertion by ID, regardless of status.

    Args:
        assertion_id: Assertion ID to retrieve

    Returns:
        Assertion if found, None otherwise
    """
    ...

assertions

assertions(
    subject: str | None = None,
    predicate: str | None = None,
    status: str | None = "active",
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Assertion]

Query assertions with optional filters.

Parameters:

Name	Type	Description	Default
subject	str | None	Filter by subject entity ID	None
predicate	str | None	Filter by predicate	None
status	str | None	Filter by current status (ignored when as_of_time is set). Defaults to "active"; pass status=None for every status.	'active'
as_of_time	datetime | None	If set, applies bitemporal filter: valid_from <= t < (valid_to or ∞) AND asserted_at <= t	None
include_flagged	bool	When as_of_time is set, whether to include 'flagged' assertions (excluded by default); ignored when as_of_time is None (pass status=None there instead).	False
include_history	bool	When as_of_time is set, whether to opt back into seeing a 'retracted' assertion once its own retraction event's timestamp is <= as_of_time (ADR-0049, KI-095; excluded by default) — mirrors include_flagged's shape (KI-098); ignored when as_of_time is None (pass status=None there instead).	False

Returns:

Type	Description
list[Assertion]	List of matching assertions

Source code in src/ontolith/store/base.py

def assertions(
    self,
    subject: str | None = None,
    predicate: str | None = None,
    status: str | None = "active",
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Assertion]:
    """Query assertions with optional filters.

    Args:
        subject: Filter by subject entity ID
        predicate: Filter by predicate
        status: Filter by current status (ignored when as_of_time is set).
            Defaults to "active"; pass status=None for every status.
        as_of_time: If set, applies bitemporal filter:
            valid_from <= t < (valid_to or ∞) AND asserted_at <= t
        include_flagged: When as_of_time is set, whether to include
            'flagged' assertions (excluded by default); ignored when
            as_of_time is None (pass status=None there instead).
        include_history: When as_of_time is set, whether to opt back
            into seeing a 'retracted' assertion once its own retraction
            event's timestamp is <= as_of_time (ADR-0049, KI-095;
            excluded by default) — mirrors include_flagged's shape
            (KI-098); ignored when as_of_time is None (pass status=None
            there instead).

    Returns:
        List of matching assertions
    """
    ...

set_assertion_status

set_assertion_status(
    assertion_id: str,
    status: str,
    valid_to: str | None = None,
) -> None

Update assertion status and optionally close validity window.

This is the ONLY allowed mutation on assertions (append-only invariant). Used for supersession and retraction.

Parameters:

Name	Type	Description	Default
assertion_id	str	Assertion ID to update	required
status	str	New status (superseded, retracted, flagged)	required
valid_to	str | None	Optional validity end time (ISO format)	None

Raises:

Type	Description
StorageError	If update fails or assertion not found

Source code in src/ontolith/store/base.py

def set_assertion_status(
    self,
    assertion_id: str,
    status: str,
    valid_to: str | None = None,
) -> None:
    """Update assertion status and optionally close validity window.

    This is the ONLY allowed mutation on assertions (append-only invariant).
    Used for supersession and retraction.

    Args:
        assertion_id: Assertion ID to update
        status: New status (superseded, retracted, flagged)
        valid_to: Optional validity end time (ISO format)

    Raises:
        StorageError: If update fails or assertion not found
    """
    ...

put_assertion_event

put_assertion_event(event: AssertionEvent) -> None

Persist an append-only assertion status-mutation event.

Parameters:

Name	Type	Description	Default
event	AssertionEvent	AssertionEvent to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_assertion_event(self, event: AssertionEvent) -> None:
    """Persist an append-only assertion status-mutation event.

    Args:
        event: AssertionEvent to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_assertion_events

get_assertion_events(
    assertion_id: str,
) -> list[AssertionEvent]

Retrieve all status-mutation events for an assertion, oldest first.

Parameters:

Name	Type	Description	Default
assertion_id	str	Assertion to retrieve events for	required

Returns:

Type	Description
list[AssertionEvent]	Events for this assertion, ordered by occurrence

Source code in src/ontolith/store/base.py

def get_assertion_events(self, assertion_id: str) -> list[AssertionEvent]:
    """Retrieve all status-mutation events for an assertion, oldest first.

    Args:
        assertion_id: Assertion to retrieve events for

    Returns:
        Events for this assertion, ordered by occurrence
    """
    ...

get_assertion_events_by_successor

get_assertion_events_by_successor(
    successor_id: str,
) -> list[AssertionEvent]

Retrieve all 'superseded' events caused by a given successor assertion.

Recovers the full predecessor set for a supersession (KI-008): Assertion.supersedes only records the first predecessor when one incoming assertion supersedes several concurrently-overlapping ones, but every superseded predecessor gets its own event row here.

Parameters:

Name	Type	Description	Default
successor_id	str	Assertion ID that caused the supersession(s)	required

Returns:

Type	Description
list[AssertionEvent]	Events with this successor_id, ordered by occurrence. Each
list[AssertionEvent]	event's assertion_id is one predecessor that was superseded.

Source code in src/ontolith/store/base.py

def get_assertion_events_by_successor(self, successor_id: str) -> list[AssertionEvent]:
    """Retrieve all 'superseded' events caused by a given successor assertion.

    Recovers the full predecessor set for a supersession (KI-008):
    Assertion.supersedes only records the first predecessor when one
    incoming assertion supersedes several concurrently-overlapping ones,
    but every superseded predecessor gets its own event row here.

    Args:
        successor_id: Assertion ID that caused the supersession(s)

    Returns:
        Events with this successor_id, ordered by occurrence. Each
        event's assertion_id is one predecessor that was superseded.
    """
    ...

list_namespaces

list_namespaces() -> list[Namespace]

List all registered namespaces (SPEC §12.2, KI-022).

Returns:

Type	Description
list[Namespace]	All namespaces, most recently created first

Source code in src/ontolith/store/base.py

def list_namespaces(self) -> list[Namespace]:
    """List all registered namespaces (SPEC §12.2, KI-022).

    Returns:
        All namespaces, most recently created first
    """
    ...

put_schema

put_schema(schema: SchemaIR) -> None

Persist a schema version.

Parameters:

Name	Type	Description	Default
schema	SchemaIR	Schema to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Note

Schema versions are never deleted (required for time-travel).

Source code in src/ontolith/store/base.py

def put_schema(self, schema: SchemaIR) -> None:
    """Persist a schema version.

    Args:
        schema: Schema to persist

    Raises:
        StorageError: If persistence fails

    Note:
        Schema versions are never deleted (required for time-travel).
    """
    ...

get_schema

get_schema(
    namespace: str, version: int | None = None
) -> SchemaIR | None

Retrieve a schema version.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to query	required
version	int | None	Specific version, or None for latest	None

Returns:

Type	Description
SchemaIR | None	Schema if found, None otherwise

Source code in src/ontolith/store/base.py

def get_schema(self, namespace: str, version: int | None = None) -> SchemaIR | None:
    """Retrieve a schema version.

    Args:
        namespace: Namespace to query
        version: Specific version, or None for latest

    Returns:
        Schema if found, None otherwise
    """
    ...

get_schema_at

get_schema_at(
    namespace: str, at: datetime
) -> SchemaIR | None

Retrieve the schema version effective at a point in time (KI-019).

Resolves the highest version whose applied_at <= at — i.e. the schema that was current at time at, for bitemporal reconstruction (SPEC §11.4: "Schema is resolved to the schema_version effective at t"). put_schema already records applied_at via the backend's injected Clock; this method is the first reader of that column.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to query	required
at	datetime	Point in time to resolve the effective schema for	required

Returns:

Type	Description
SchemaIR | None	Schema effective at at, or None if no version had been applied
SchemaIR | None	by that time (including when the namespace has no schema at all,
SchemaIR | None	or its first version postdates at)

Source code in src/ontolith/store/base.py

def get_schema_at(self, namespace: str, at: datetime) -> SchemaIR | None:
    """Retrieve the schema version effective at a point in time (KI-019).

    Resolves the highest version whose `applied_at <= at` — i.e. the
    schema that was current at time `at`, for bitemporal reconstruction
    (SPEC §11.4: "Schema is resolved to the schema_version effective at
    t"). `put_schema` already records `applied_at` via the backend's
    injected Clock; this method is the first reader of that column.

    Args:
        namespace: Namespace to query
        at: Point in time to resolve the effective schema for

    Returns:
        Schema effective at `at`, or None if no version had been applied
        by that time (including when the namespace has no schema at all,
        or its first version postdates `at`)
    """
    ...

entities

entities(
    namespace: str | None = None,
    concept: str | None = None,
    as_of_time: datetime | None = None,
) -> list[Entity]

Query entities with optional filters.

Parameters:

Name	Type	Description	Default
namespace	str | None	Filter by namespace	None
concept	str | None	Filter by concept	None
as_of_time	datetime | None	If set, exclude entities created after this time	None

Returns:

Type	Description
list[Entity]	List of matching entities

Source code in src/ontolith/store/base.py

def entities(
    self,
    namespace: str | None = None,
    concept: str | None = None,
    as_of_time: datetime | None = None,
) -> list[Entity]:
    """Query entities with optional filters.

    Args:
        namespace: Filter by namespace
        concept: Filter by concept
        as_of_time: If set, exclude entities created after this time

    Returns:
        List of matching entities
    """
    ...

entities_where

entities_where(
    namespace: str,
    concept: str,
    predicate_filters: list[tuple[str, str, Any]],
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Entity]

Query entities matching all predicate filters in one SQL query.

Avoids the N+1 pattern of entities() + per-entity assertions() calls. Each filter is (full_predicate, operator, value); ALL must match (AND semantics) — a list, not a dict, since two different operators can target the same predicate (e.g. an age range needs both a "gte" and a "lt" filter). operator is one of:

• "eq": equality. Matches either a literal property (value_lit) or a relation's target entity id (value_ref) — KI-030.
• "contains": case-sensitive substring match against value_lit only (KI-039) — relations have no defined substring semantics, so this operator never matches against value_ref. Case-sensitive on both backends by construction: SQLite's LIKE is case-insensitive by default and DuckDB's is not, so SQLiteBackend explicitly sets PRAGMA case_sensitive_like = ON at connection time to make the two agree — a conformant third-party backend implementing this port must match that behavior, not SQLite's un-pragma'd default.
• "gt"/"lt"/"gte"/"lte": numeric range against value_lit only (KI-039) — a numeric cast (CAST/TRY_CAST, exact type backend-specific — see e.g. DuckDBBackend's docstring for why DOUBLE not REAL). Callers (in practice, only QueryBuilder) are responsible for restricting this to predicates declared numeric — see _RANGE_VALUE_TYPES in ontolith.query.builder's docstring for why the backend itself doesn't validate that. A backend is NOT responsible for validating that already-stored value_lit content actually parses as a number for a predicate declared numeric (KI-049) — implementations should fail safe (exclude the row) rather than raise for a value that doesn't parse, the way TRY_CAST does; letting a raw conversion exception escape through this port violates SPEC §16's error taxonomy.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to query	required
concept	str	Concept to filter by	required
predicate_filters	list[tuple[str, str, Any]]	List of (full_predicate, operator, value) triples (KI-039)	required
as_of_time	datetime | None	If set, applies bitemporal filter on assertions and entity creation	None
include_flagged	bool	Whether to also match 'flagged' assertions (KI-081). Honored on both the current-state and the as_of_time path (excluded by default on both). On the as_of_time path this is point-in-time, not current status (KI-097): reconstructed from the assertion_event log the same way assertions() already does, so a query pinned to a time when an assertion was disputed correctly excludes it even after the dispute has since been resolved — and, the other direction, a time strictly before any dispute existed still includes an otherwise-undisputed value, even though the same assertion is flagged now.	False
include_history	bool	Whether to also match 'superseded' and 'retracted' assertions (KI-081). On the current-state path this widens beyond 'active'. On the as_of_time path, 'superseded' is unaffected either way (its window already never restricts to 'active') — but 'retracted' does something under this flag now (ADR-0049, KI-095): the as_of_time branch additionally excludes a 'retracted' assertion once its own retraction event's timestamp is <= as_of_time; this parameter opts back out of that exclusion.	False

Returns:

Type	Description
list[Entity]	List of entities where all filters match at the given time

Source code in src/ontolith/store/base.py

def entities_where(
    self,
    namespace: str,
    concept: str,
    predicate_filters: list[tuple[str, str, Any]],
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Entity]:
    """Query entities matching all predicate filters in one SQL query.

    Avoids the N+1 pattern of entities() + per-entity assertions() calls.
    Each filter is `(full_predicate, operator, value)`; ALL must match
    (AND semantics) — a list, not a dict, since two different operators
    can target the same predicate (e.g. an `age` range needs both a
    `"gte"` and a `"lt"` filter). `operator` is one of:

    - `"eq"`: equality. Matches either a literal property (`value_lit`)
      or a relation's target entity id (`value_ref`) — KI-030.
    - `"contains"`: case-sensitive substring match against `value_lit`
      only (KI-039) — relations have no defined substring semantics, so
      this operator never matches against `value_ref`. Case-sensitive
      on both backends by construction: SQLite's `LIKE` is
      case-insensitive by default and DuckDB's is not, so
      `SQLiteBackend` explicitly sets `PRAGMA case_sensitive_like = ON`
      at connection time to make the two agree — a conformant
      third-party backend implementing this port must match that
      behavior, not SQLite's un-pragma'd default.
    - `"gt"`/`"lt"`/`"gte"`/`"lte"`: numeric range against `value_lit`
      only (KI-039) — a numeric cast (`CAST`/`TRY_CAST`, exact type
      backend-specific — see e.g. `DuckDBBackend`'s docstring for why
      `DOUBLE` not `REAL`). Callers (in practice, only `QueryBuilder`)
      are responsible for restricting this to predicates *declared*
      numeric — see `_RANGE_VALUE_TYPES` in `ontolith.query.builder`'s
      docstring for why the backend itself doesn't validate that. A
      backend is NOT responsible for validating that already-stored
      `value_lit` content actually parses as a number for a predicate
      declared numeric (KI-049) — implementations should fail safe
      (exclude the row) rather than raise for a value that doesn't
      parse, the way `TRY_CAST` does; letting a raw conversion
      exception escape through this port violates SPEC §16's error
      taxonomy.

    Args:
        namespace: Namespace to query
        concept: Concept to filter by
        predicate_filters: List of `(full_predicate, operator, value)`
            triples (KI-039)
        as_of_time: If set, applies bitemporal filter on assertions and entity creation
        include_flagged: Whether to also match 'flagged' assertions
            (KI-081). Honored on both the current-state and the
            as_of_time path (excluded by default on both). On the
            as_of_time path this is point-in-time, not current status
            (KI-097): reconstructed from the assertion_event log the
            same way assertions() already does, so a query pinned to a
            time when an assertion *was* disputed correctly excludes
            it even after the dispute has since been resolved — and,
            the other direction, a time strictly before any dispute
            existed still includes an otherwise-undisputed value, even
            though the same assertion is flagged now.
        include_history: Whether to also match 'superseded' and
            'retracted' assertions (KI-081). On the current-state path
            this widens beyond 'active'. On the as_of_time path,
            'superseded' is unaffected either way (its window already
            never restricts to 'active') — but 'retracted' does
            something under this flag now (ADR-0049, KI-095): the
            as_of_time branch additionally excludes a 'retracted'
            assertion once its own retraction event's timestamp is
            <= as_of_time; this parameter opts back out of that
            exclusion.

    Returns:
        List of entities where all filters match at the given time
    """
    ...

entities_meeting_confidence

entities_meeting_confidence(
    namespace: str,
    concept: str,
    threshold: float,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]

IDs of entities in (namespace, concept) with >=1 assertion at or above threshold confidence, active at as_of_time (or currently active, if as_of_time is None).

Avoids the N+1 pattern of calling assertions() once per candidate entity (QueryBuilder.min_confidence(), KI-028) — one SQL round trip regardless of concept size. None confidence never qualifies (ADR-0004). When as_of_time is None, "active" means status = 'active', widened by include_flagged/include_history exactly as entities_where()'s current-state path is (KI-093); when as_of_time is set, it means the same bitemporal window entities_where() uses (asserted_at <= as_of_time, valid_from/ valid_to bracketing as_of_time) — KI-036, so kb.as_of(t).query(...).min_confidence(...) evaluates against a coherent point-in-time view instead of always checking current-active assertions regardless of t. status = 'flagged' (a static contradiction, SPEC §10.3) is excluded point-in-time, not by current status (KI-097): reconstructed from the assertion_event log the same way assertions() already does, so a t before the dispute existed still qualifies and a t during a dispute that has since been resolved still correctly excludes — unless include_flagged is set — matching entities_where()'s identical as_of handling. include_history is mostly a no-op under as_of_time, also matching entities_where(): that branch never restricts to active in the first place, only conditionally excludes flagged, so a superseded assertion whose window covers t already qualifies without this flag. A retracted assertion is the one exception (ADR-0049, KI-095): as_of_time additionally excludes it once its own retraction event's timestamp is <= as_of_time, and include_history opts back out of that exclusion.

Always scoped by (namespace, concept) — this is what keeps the query's parameter count constant regardless of how many entities exist, unlike a mandatory id-list-bound design (a SQL IN (...) with one placeholder per candidate hits both SQLite's bound-variable limit and, on DuckDB, per-parameter bind overhead, at real-world scale — see KI-028's own Fix text for why that design was tried and reverted before this method first shipped). candidate_ids, when given, is an optional narrowing hint on top of that scope — not a replacement for it — for when the caller has already narrowed to a small candidate set via .where()/ .semantic() (KI-028's fix otherwise forced even a single-candidate .where() match to re-scan the entire concept; KI-037). A backend MAY use it to cut real work (e.g. SQLite binds it as one JSON-encoded parameter, avoiding the per-placeholder cost a literal IN (...) would reintroduce) or ignore it and keep scanning — both are correct, since the caller always re-intersects the returned set against its own candidate list.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to scope the scan to	required
concept	str	Concept to scope the scan to	required
threshold	float	Minimum confidence, 0.0-1.0	required
as_of_time	datetime | None	If set, evaluate against this point in time instead of current state (KI-036) — see the flagged-status point-in-time reconstruction above	None
candidate_ids	frozenset[str] | None	Optional narrowing hint (KI-037) — a backend may use this to scope the scan below (namespace, concept), but is not required to	None
include_flagged	bool	Also count 'flagged' assertions (KI-093). Honored on both the current-state and as_of_time paths, matching entities_where().	False
include_history	bool	Also count 'superseded'/'retracted' assertions (KI-093). Widens the current-state path beyond 'active'. Under as_of_time, mostly a no-op — but not for 'retracted' (ADR-0049, KI-095), matching entities_where().	False

Returns:

Type	Description
set[str]	IDs of qualifying entities (may be a superset of any candidate
set[str]	list the caller intends to intersect this against)

Source code in src/ontolith/store/base.py

def entities_meeting_confidence(
    self,
    namespace: str,
    concept: str,
    threshold: float,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]:
    """IDs of entities in `(namespace, concept)` with >=1 assertion at or
    above `threshold` confidence, active at `as_of_time` (or currently
    active, if `as_of_time` is None).

    Avoids the N+1 pattern of calling assertions() once per candidate
    entity (QueryBuilder.min_confidence(), KI-028) — one SQL round trip
    regardless of concept size. `None` confidence never qualifies
    (ADR-0004). When `as_of_time` is None, "active" means
    `status = 'active'`, widened by `include_flagged`/`include_history`
    exactly as `entities_where()`'s current-state path is (KI-093); when
    `as_of_time` is set, it means the same bitemporal window
    `entities_where()` uses (`asserted_at <= as_of_time`, `valid_from`/
    `valid_to` bracketing `as_of_time`) — KI-036, so
    `kb.as_of(t).query(...).min_confidence(...)` evaluates against a
    coherent point-in-time view instead of always checking
    current-active assertions regardless of `t`. `status = 'flagged'`
    (a static contradiction, SPEC §10.3) is excluded point-in-time, not
    by current status (KI-097): reconstructed from the assertion_event
    log the same way `assertions()` already does, so a `t` before the
    dispute existed still qualifies and a `t` during a dispute that has
    since been resolved still correctly excludes — unless
    `include_flagged` is set — matching `entities_where()`'s identical
    `as_of` handling. `include_history` is mostly a no-op under
    `as_of_time`, also matching `entities_where()`: that branch never
    restricts to `active` in the first place, only conditionally
    excludes `flagged`, so a `superseded` assertion whose window covers
    `t` already qualifies without this flag. A `retracted` assertion is
    the one exception (ADR-0049, KI-095): `as_of_time` additionally
    excludes it once its own retraction event's timestamp is <=
    `as_of_time`, and `include_history` opts back out of that
    exclusion.

    Always scoped by `(namespace, concept)` — this is what keeps the
    query's parameter count constant regardless of how many entities
    exist, unlike a mandatory id-list-bound design (a SQL `IN (...)`
    with one placeholder per candidate hits both SQLite's
    bound-variable limit and, on DuckDB, per-parameter bind overhead,
    at real-world scale — see KI-028's own Fix text for why that design
    was tried and reverted before this method first shipped).
    `candidate_ids`, when given, is an *optional* narrowing hint on top
    of that scope — not a replacement for it — for when the caller has
    already narrowed to a small candidate set via `.where()`/
    `.semantic()` (KI-028's fix otherwise forced even a single-candidate
    `.where()` match to re-scan the entire concept; KI-037). A backend
    MAY use it to cut real work (e.g. SQLite binds it as one
    JSON-encoded parameter, avoiding the per-placeholder cost a literal
    `IN (...)` would reintroduce) or ignore it and keep scanning — both
    are correct, since the caller always re-intersects the returned set
    against its own candidate list.

    Args:
        namespace: Namespace to scope the scan to
        concept: Concept to scope the scan to
        threshold: Minimum confidence, 0.0-1.0
        as_of_time: If set, evaluate against this point in time instead
            of current state (KI-036) — see the flagged-status
            point-in-time reconstruction above
        candidate_ids: Optional narrowing hint (KI-037) — a backend may
            use this to scope the scan below `(namespace, concept)`,
            but is not required to
        include_flagged: Also count 'flagged' assertions (KI-093).
            Honored on both the current-state and as_of_time paths,
            matching entities_where().
        include_history: Also count 'superseded'/'retracted' assertions
            (KI-093). Widens the current-state path beyond 'active'.
            Under as_of_time, mostly a no-op — but not for 'retracted'
            (ADR-0049, KI-095), matching entities_where().

    Returns:
        IDs of qualifying entities (may be a superset of any candidate
        list the caller intends to intersect this against)
    """
    ...

entities_meeting_trust

entities_meeting_trust(
    namespace: str,
    concept: str,
    min_trust: int,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]

IDs of entities in (namespace, concept) with >=1 assertion, active at as_of_time (or currently active, if as_of_time is None) — widened by include_flagged/include_history exactly as entities_meeting_confidence()'s identical parameters are (KI-093) — whose effective trust_level >= min_trust.

"Effective" (KI-047): when the qualifying assertion was made under delegation (acting_as set), the comparison is min(author.trust_level, acting_as.trust_level), not the author's raw trust_level alone. SPEC §8.4 states this min() rule for capability only ("the effective capability for the operation is min(capability(author), capability(acting_as))"); the trust-min is govern/policy.py's own conservative extension of that same principle, applied here by analogy, not a separate SPEC mandate. For a non-delegated assertion, this is simply the author's own trust_level, unchanged from before this method considered delegation at all.

If acting_as names a principal that doesn't resolve (there is no FK from assertion.acting_as to principal.id, so this can only happen via a direct put_assertion() call bypassing Ontology's write paths — e.g. a legacy import — since Ontology's own paths always validate the delegate exists before writing), implementations MUST fall back to the author's own trust_level rather than excluding the row or raising — i.e. treat an unresolvable delegate the same as no delegate at all. This deliberately fails open, unlike govern/policy.py's _resolve_delegation which fails closed (raises AuthError) for the same input — policy evaluation runs once, at write time, when rejecting the write outright is cheap and correct; this method runs on every query against already-committed data, where excluding or erroring on a row for data that was already accepted would be a surprising, un-auditable behavior change with no corresponding write.

Avoids the N+1 pattern of calling assertions() + get_principal() once per (candidate entity, assertion) pair (QueryBuilder. trust_at_least(), KI-028) — one SQL round trip regardless of concept size. Scoped by (namespace, concept), with the same optional candidate_ids narrowing hint (KI-037), for the same reason as entities_meeting_confidence — see its docstring.

as_of_time bitemporally scopes which assertion qualifies, the same way entities_meeting_confidence does (including its flagged-status point-in-time reconstruction, KI-097) — but each individual principal's own trust_level (author's and, if delegated, acting_as's) is always its current value, never a historical one (KI-036), and the min() this method now takes of the two (KI-047) inherits that same property. This is not an approximation: no code path updates a principal's trust_level after creation, so "trust_level as of any t at or after the principal's creation" and "trust_level now" are the same value by construction (guarded by tests/unit/test_principal_trust_immutability_invariant.py, which fails the day a mutation path is added — that would mean this method needs real principal versioning, not this shortcut). A principal cannot author an assertion before it exists, so this holds for every as_of_time an assertion's asserted_at could satisfy.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to scope the scan to	required
concept	str	Concept to scope the scan to	required
min_trust	int	Minimum effective trust level, 0-10	required
as_of_time	datetime | None	If set, evaluate assertion existence against this point in time instead of current state (KI-036)	None
candidate_ids	frozenset[str] | None	Optional narrowing hint (KI-037) — see entities_meeting_confidence's docstring	None
include_flagged	bool	Also count 'flagged' assertions (KI-093) — see entities_meeting_confidence's docstring	False
include_history	bool	Also count 'superseded'/'retracted' assertions (KI-093) — see entities_meeting_confidence's docstring	False

Returns:

Type	Description
set[str]	IDs of qualifying entities (may be a superset of any candidate
set[str]	list the caller intends to intersect this against)

Source code in src/ontolith/store/base.py

def entities_meeting_trust(
    self,
    namespace: str,
    concept: str,
    min_trust: int,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]:
    """IDs of entities in `(namespace, concept)` with >=1 assertion,
    active at `as_of_time` (or currently active, if `as_of_time` is
    None) — widened by `include_flagged`/`include_history` exactly as
    `entities_meeting_confidence()`'s identical parameters are (KI-093)
    — whose *effective* trust_level >= `min_trust`.

    "Effective" (KI-047): when the qualifying assertion was made under
    delegation (`acting_as` set), the comparison is
    `min(author.trust_level, acting_as.trust_level)`, not the author's
    raw `trust_level` alone. SPEC §8.4 states this `min()` rule for
    *capability* only ("the effective capability for the operation is
    `min(capability(author), capability(acting_as))`"); the trust-min
    is `govern/policy.py`'s own conservative extension of that same
    principle, applied here by analogy, not a separate SPEC mandate.
    For a non-delegated assertion, this is simply the author's own
    `trust_level`, unchanged from before this method considered
    delegation at all.

    If `acting_as` names a principal that doesn't resolve (there is no
    FK from `assertion.acting_as` to `principal.id`, so this can only
    happen via a direct `put_assertion()` call bypassing `Ontology`'s
    write paths — e.g. a legacy import — since `Ontology`'s own paths
    always validate the delegate exists before writing), implementations
    MUST fall back to the author's own `trust_level` rather than
    excluding the row or raising — i.e. treat an unresolvable delegate
    the same as no delegate at all. This deliberately fails *open*,
    unlike `govern/policy.py`'s `_resolve_delegation` which fails
    *closed* (raises `AuthError`) for the same input — policy
    evaluation runs once, at write time, when rejecting the write
    outright is cheap and correct; this method runs on every query
    against already-committed data, where excluding or erroring on a
    row for data that was already accepted would be a surprising,
    un-auditable behavior change with no corresponding write.

    Avoids the N+1 pattern of calling assertions() + get_principal()
    once per (candidate entity, assertion) pair (QueryBuilder.
    trust_at_least(), KI-028) — one SQL round trip regardless of
    concept size. Scoped by `(namespace, concept)`, with the same
    optional `candidate_ids` narrowing hint (KI-037), for the same
    reason as `entities_meeting_confidence` — see its docstring.

    `as_of_time` bitemporally scopes which *assertion* qualifies, the
    same way `entities_meeting_confidence` does (including its
    flagged-status point-in-time reconstruction, KI-097) — but each
    individual principal's own `trust_level` (author's and, if
    delegated, `acting_as`'s) is always its current value, never a
    historical one (KI-036), and the `min()`
    this method now takes of the two (KI-047) inherits that same
    property. This is not an approximation: no code path updates a
    principal's `trust_level` after creation, so "trust_level as of any
    t at or after the principal's creation" and "trust_level now" are
    the same value by construction (guarded by
    `tests/unit/test_principal_trust_immutability_invariant.py`, which
    fails the day a mutation path is added — that would mean this
    method needs real principal versioning, not this shortcut). A
    principal cannot author an assertion before it exists, so this
    holds for every `as_of_time` an assertion's `asserted_at` could
    satisfy.

    Args:
        namespace: Namespace to scope the scan to
        concept: Concept to scope the scan to
        min_trust: Minimum effective trust level, 0-10
        as_of_time: If set, evaluate assertion existence against this
            point in time instead of current state (KI-036)
        candidate_ids: Optional narrowing hint (KI-037) — see
            `entities_meeting_confidence`'s docstring
        include_flagged: Also count 'flagged' assertions (KI-093) — see
            `entities_meeting_confidence`'s docstring
        include_history: Also count 'superseded'/'retracted' assertions
            (KI-093) — see `entities_meeting_confidence`'s docstring

    Returns:
        IDs of qualifying entities (may be a superset of any candidate
        list the caller intends to intersect this against)
    """
    ...

vector_upsert

vector_upsert(
    scope: str, id: str, vec: list[float]
) -> None

Insert or replace the embedding vector for (scope, id).

The dimensionality of the first vector ever upserted into a scope establishes that scope's dimension for the life of the store; later upserts into the same scope must match it.

Parameters:

Name	Type	Description	Default
scope	str	Embedding scope. Must be one of VECTOR_SCOPES.	required
id	str	Entity or assertion ID the vector represents.	required
vec	list[float]	Embedding vector.	required

Raises:

Type	Description
ValidationError	scope is not in VECTOR_SCOPES, or vec's length does not match the scope's already-established dimension.
StorageError	If persistence fails.

Source code in src/ontolith/store/base.py

def vector_upsert(self, scope: str, id: str, vec: list[float]) -> None:
    """Insert or replace the embedding vector for (scope, id).

    The dimensionality of the first vector ever upserted into a scope
    establishes that scope's dimension for the life of the store; later
    upserts into the same scope must match it.

    Args:
        scope: Embedding scope. Must be one of VECTOR_SCOPES.
        id: Entity or assertion ID the vector represents.
        vec: Embedding vector.

    Raises:
        ValidationError: scope is not in VECTOR_SCOPES, or vec's length
            does not match the scope's already-established dimension.
        StorageError: If persistence fails.
    """
    ...

vector_search

vector_search(
    scope: str, vec: list[float], k: int
) -> list[tuple[str, float]]

Return the k nearest ids to vec within scope, ascending distance.

Distance is L2 (Euclidean). Embedder implementations MUST return L2-unit-normalized vectors, which makes ascending-L2-distance order equivalent to descending-cosine-similarity order.

Parameters:

Name	Type	Description	Default
scope	str	Embedding scope. Must be one of VECTOR_SCOPES.	required
vec	list[float]	Query vector.	required
k	int	Maximum number of results.	required

Returns:

Type	Description
list[tuple[str, float]]	(id, distance) tuples, nearest first. Fewer than k if the scope
list[tuple[str, float]]	has fewer than k vectors; empty list if the scope has never
list[tuple[str, float]]	been populated.

Raises:

Type	Description
ValidationError	scope is not in VECTOR_SCOPES, or vec's length does not match the scope's already-established dimension.

Source code in src/ontolith/store/base.py

def vector_search(self, scope: str, vec: list[float], k: int) -> list[tuple[str, float]]:
    """Return the k nearest ids to vec within scope, ascending distance.

    Distance is L2 (Euclidean). Embedder implementations MUST return
    L2-unit-normalized vectors, which makes ascending-L2-distance order
    equivalent to descending-cosine-similarity order.

    Args:
        scope: Embedding scope. Must be one of VECTOR_SCOPES.
        vec: Query vector.
        k: Maximum number of results.

    Returns:
        (id, distance) tuples, nearest first. Fewer than k if the scope
        has fewer than k vectors; empty list if the scope has never
        been populated.

    Raises:
        ValidationError: scope is not in VECTOR_SCOPES, or vec's length
            does not match the scope's already-established dimension.
    """
    ...

put_proposal

put_proposal(proposal: Proposal) -> None

Persist a proposal.

Parameters:

Name	Type	Description	Default
proposal	Proposal	Proposal to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_proposal(self, proposal: Proposal) -> None:
    """Persist a proposal.

    Args:
        proposal: Proposal to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_proposal

get_proposal(proposal_id: str) -> Proposal | None

Retrieve a proposal by ID.

Parameters:

Name	Type	Description	Default
proposal_id	str	Proposal ID	required

Returns:

Type	Description
Proposal | None	Proposal if found, None otherwise

Source code in src/ontolith/store/base.py

def get_proposal(self, proposal_id: str) -> Proposal | None:
    """Retrieve a proposal by ID.

    Args:
        proposal_id: Proposal ID

    Returns:
        Proposal if found, None otherwise
    """
    ...

proposals

proposals(state: str | None = None) -> list[Proposal]

Query proposals, optionally filtered by state (SPEC §14.1).

Parameters:

Name	Type	Description	Default
state	str | None	Filter by proposal state (e.g. "require_review"); None returns proposals in every state	None

Returns:

Type	Description
list[Proposal]	Matching proposals, most recently created first

Source code in src/ontolith/store/base.py

def proposals(self, state: str | None = None) -> list[Proposal]:
    """Query proposals, optionally filtered by state (SPEC §14.1).

    Args:
        state: Filter by proposal state (e.g. "require_review");
            None returns proposals in every state

    Returns:
        Matching proposals, most recently created first
    """
    ...

update_proposal_state

update_proposal_state(
    proposal_id: str,
    state: str,
    decided_at: str | None = None,
    policy_reason: str | None = None,
) -> None

Update proposal state after policy decision.

Parameters:

Name	Type	Description	Default
proposal_id	str	Proposal to update	required
state	str	New state (auto_accepted, require_review, rejected, etc.)	required
decided_at	str | None	ISO timestamp of the decision. Set unconditionally, including to None — unlike policy_reason, passing None clears the stored value rather than leaving it unchanged. Ontology.resubmit (KI-027) relies on this to re-open an already-decided proposal: a resubmission that lands back in require_review is not yet decided again, and must clear the prior decided_at rather than keep the stale value from the request_changes decision it's superseding.	None
policy_reason	str | None	Human-readable reason from policy engine. If None, the stored value is left unchanged (not cleared) — the policy-engine reason set at proposal-creation time is distinct from, and not overwritten by, review actions recorded via put_proposal_event.	None

Raises:

Type	Description
StorageError	proposal_id does not name an existing proposal

Source code in src/ontolith/store/base.py

def update_proposal_state(
    self,
    proposal_id: str,
    state: str,
    decided_at: str | None = None,
    policy_reason: str | None = None,
) -> None:
    """Update proposal state after policy decision.

    Args:
        proposal_id: Proposal to update
        state: New state (auto_accepted, require_review, rejected, etc.)
        decided_at: ISO timestamp of the decision. Set unconditionally,
            including to None — unlike policy_reason, passing None
            clears the stored value rather than leaving it unchanged.
            `Ontology.resubmit` (KI-027) relies on this to re-open an
            already-decided proposal: a resubmission that lands back in
            require_review is not yet decided again, and must clear the
            prior decided_at rather than keep the stale value from the
            request_changes decision it's superseding.
        policy_reason: Human-readable reason from policy engine. If None,
            the stored value is left unchanged (not cleared) — the
            policy-engine reason set at proposal-creation time is
            distinct from, and not overwritten by, review actions
            recorded via put_proposal_event.

    Raises:
        StorageError: proposal_id does not name an existing proposal
    """
    ...

update_proposal_reviewers

update_proposal_reviewers(
    proposal_id: str, reviewers: list[str]
) -> None

Replace a proposal's assigned reviewers (SPEC §9.4's assign action).

Unlike update_proposal_state's policy_reason, there is no "leave unchanged" sentinel here — reviewers is always replaced wholesale with what's passed, including an empty list (which clears every assignment). A dedicated method rather than folding this into update_proposal_state: assign doesn't change state, and update_proposal_state already has enough state/decided_at/policy_reason parameters with their own distinct semantics without adding a fourth (KI-078).

Parameters:

Name	Type	Description	Default
proposal_id	str	Proposal to update	required
reviewers	list[str]	New reviewer list, replacing whatever was there before	required

Raises:

Type	Description
StorageError	proposal_id does not name an existing proposal

Source code in src/ontolith/store/base.py

def update_proposal_reviewers(self, proposal_id: str, reviewers: list[str]) -> None:
    """Replace a proposal's assigned reviewers (SPEC §9.4's `assign` action).

    Unlike `update_proposal_state`'s `policy_reason`, there is no
    "leave unchanged" sentinel here — `reviewers` is always replaced
    wholesale with what's passed, including an empty list (which
    clears every assignment). A dedicated method rather than folding
    this into `update_proposal_state`: `assign` doesn't change
    `state`, and `update_proposal_state` already has enough
    state/decided_at/policy_reason parameters with their own distinct
    semantics without adding a fourth (KI-078).

    Args:
        proposal_id: Proposal to update
        reviewers: New reviewer list, replacing whatever was there before

    Raises:
        StorageError: proposal_id does not name an existing proposal
    """
    ...

put_proposal_event

put_proposal_event(event: ProposalEvent) -> None

Persist a structured review-action event (SPEC §9.4).

Parameters:

Name	Type	Description	Default
event	ProposalEvent	ProposalEvent to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_proposal_event(self, event: ProposalEvent) -> None:
    """Persist a structured review-action event (SPEC §9.4).

    Args:
        event: ProposalEvent to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_proposal_events

get_proposal_events(
    proposal_id: str,
) -> list[ProposalEvent]

Retrieve all review events for a proposal, oldest first.

Parameters:

Name	Type	Description	Default
proposal_id	str	Proposal to retrieve events for	required

Returns:

Type	Description
list[ProposalEvent]	Events for this proposal, ordered by occurrence

Source code in src/ontolith/store/base.py

def get_proposal_events(self, proposal_id: str) -> list[ProposalEvent]:
    """Retrieve all review events for a proposal, oldest first.

    Args:
        proposal_id: Proposal to retrieve events for

    Returns:
        Events for this proposal, ordered by occurrence
    """
    ...

put_contradiction

put_contradiction(contradiction: Contradiction) -> None

Persist a new contradiction.

Parameters:

Name	Type	Description	Default
contradiction	Contradiction	Contradiction to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/base.py

def put_contradiction(self, contradiction: Contradiction) -> None:
    """Persist a new contradiction.

    Args:
        contradiction: Contradiction to persist

    Raises:
        StorageError: If persistence fails
    """
    ...

get_open_contradiction

get_open_contradiction(
    namespace: str, subject: str, predicate: str
) -> Contradiction | None

Return the open contradiction for (namespace, subject, predicate), if any.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to search	required
subject	str	Subject entity ID	required
predicate	str	Predicate name	required

Returns:

Type	Description
Contradiction | None	Open Contradiction if one exists, None otherwise

Source code in src/ontolith/store/base.py

def get_open_contradiction(
    self,
    namespace: str,
    subject: str,
    predicate: str,
) -> Contradiction | None:
    """Return the open contradiction for (namespace, subject, predicate), if any.

    Args:
        namespace: Namespace to search
        subject: Subject entity ID
        predicate: Predicate name

    Returns:
        Open Contradiction if one exists, None otherwise
    """
    ...

update_contradiction_members

update_contradiction_members(
    contradiction_id: str,
    member_ids: list[str],
    metadata: dict[str, Any] | None = None,
) -> None

Replace the member_ids list (and, optionally, the metadata blob) on an existing contradiction (KI-071).

Parameters:

Name	Type	Description	Default
contradiction_id	str	Contradiction to update	required
member_ids	list[str]	New full list of member assertion IDs	required
metadata	dict[str, Any] | None	If given, replaces the contradiction's metadata blob wholesale — the caller is expected to pass the complete desired dict (e.g. built from a fresh read plus one appended entry), matching member_ids' own full-replacement convention rather than a merge/delta. None (the default) leaves metadata untouched.	None

Source code in src/ontolith/store/base.py

def update_contradiction_members(
    self,
    contradiction_id: str,
    member_ids: list[str],
    metadata: dict[str, Any] | None = None,
) -> None:
    """Replace the member_ids list (and, optionally, the metadata blob)
    on an existing contradiction (KI-071).

    Args:
        contradiction_id: Contradiction to update
        member_ids: New full list of member assertion IDs
        metadata: If given, replaces the contradiction's metadata blob
            wholesale — the caller is expected to pass the complete
            desired dict (e.g. built from a fresh read plus one
            appended entry), matching member_ids' own
            full-replacement convention rather than a merge/delta.
            ``None`` (the default) leaves metadata untouched.
    """
    ...

get_contradiction

get_contradiction(
    contradiction_id: str,
) -> Contradiction | None

Retrieve a contradiction by ID, regardless of state.

Parameters:

Name	Type	Description	Default
contradiction_id	str	Contradiction ID to retrieve	required

Returns:

Type	Description
Contradiction | None	Contradiction if found, None otherwise

Source code in src/ontolith/store/base.py

def get_contradiction(self, contradiction_id: str) -> Contradiction | None:
    """Retrieve a contradiction by ID, regardless of state.

    Args:
        contradiction_id: Contradiction ID to retrieve

    Returns:
        Contradiction if found, None otherwise
    """
    ...

contradictions

contradictions(
    state: str | None = None,
) -> list[Contradiction]

Query contradictions, optionally filtered by state (SPEC §14.1).

Parameters:

Name	Type	Description	Default
state	str | None	Filter by contradiction state ("open" or "resolved"); None returns contradictions in every state	None

Returns:

Type	Description
list[Contradiction]	Matching contradictions, most recently created first

Source code in src/ontolith/store/base.py

def contradictions(self, state: str | None = None) -> list[Contradiction]:
    """Query contradictions, optionally filtered by state (SPEC §14.1).

    Args:
        state: Filter by contradiction state ("open" or "resolved");
            None returns contradictions in every state

    Returns:
        Matching contradictions, most recently created first
    """
    ...

resolve_contradiction

resolve_contradiction(
    contradiction_id: str,
    resolved_by: str,
    resolved_at: datetime,
) -> None

Mark a contradiction as resolved (SPEC §10.3).

Parameters:

Name	Type	Description	Default
contradiction_id	str	Contradiction to resolve	required
resolved_by	str	Principal ID who resolved it	required
resolved_at	datetime	Timestamp of resolution	required

Raises:

Type	Description
StorageError	If the contradiction is not found or update fails

Source code in src/ontolith/store/base.py

def resolve_contradiction(
    self,
    contradiction_id: str,
    resolved_by: str,
    resolved_at: datetime,
) -> None:
    """Mark a contradiction as resolved (SPEC §10.3).

    Args:
        contradiction_id: Contradiction to resolve
        resolved_by: Principal ID who resolved it
        resolved_at: Timestamp of resolution

    Raises:
        StorageError: If the contradiction is not found or update fails
    """
    ...

close

close() -> None

Close the storage backend and release resources.

Source code in src/ontolith/store/base.py

def close(self) -> None:
    """Close the storage backend and release resources."""
    ...

ontolith.store.sqlite

SQLite storage backend.

Default storage adapter for Ontolith using SQLite3 with bitemporal schema.

SQLiteBackend

SQLiteBackend(
    path: str | Path, *, clock: Clock | None = None
)

SQLite implementation of StorageBackend.

Schema follows SPEC §12.2: - entity table with ULID primary key - assertion table with ULID primary key - Bitemporal columns (asserted_at, valid_from, valid_to) - Status tracking for append-only invariant

KI-066: assertion_event/proposal_event's append-only invariant (SPEC §17) is backed here by three triggers per table — BEFORE UPDATE, BEFORE DELETE, and BEFORE INSERT ... WHEN EXISTS(...) — raising sqlite3.IntegrityError on any raw mutation attempt, not just the port surface exposing no update/delete method. The BEFORE INSERT trigger is what actually closes INSERT OR REPLACE: it is schema state (persisted in the file, enforced on every connection), unlike PRAGMA recursive_triggers = ON (also set below, as defense in depth) which is per-connection and does not by itself stop a second raw connection to the same file from reviving the REPLACE bypass. DuckDBBackend has no equivalent — see its own docstring.

Initialize SQLite backend.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to SQLite database file (created if doesn't exist)	required
clock	Clock | None	Clock for timestamps (defaults to SystemClock)	None

Source code in src/ontolith/store/sqlite/backend.py

def __init__(self, path: str | Path, *, clock: Clock | None = None) -> None:
    """Initialize SQLite backend.

    Args:
        path: Path to SQLite database file (created if doesn't exist)
        clock: Clock for timestamps (defaults to SystemClock)
    """
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    # isolation_level=None: autocommit mode (ADR-0010).
    # Each write auto-commits unless _in_transaction is True.
    # check_same_thread=False: an ASGI server (REST, interfaces/rest.py)
    # dispatches sync route handlers onto a worker threadpool, which is
    # a different OS thread than the one that constructed this backend
    # — stock sqlite3 blocks that regardless of whether the access is
    # ever actually concurrent. This flag only lifts that same-thread
    # check; it does not by itself serialize concurrent access — that is
    # what self._lock (below) does (KI-023).
    # timeout=5.0 (KI-084): 5.0 is already Python's own sqlite3 default
    # when this kwarg is omitted (verified directly), so this line is a
    # behavioral no-op by itself — pinning it explicitly rather than
    # inheriting a stdlib default is what makes it a deliberate,
    # documented value instead of an unexamined implicit one. It only
    # becomes load-bearing given the change below: begin() now issues
    # BEGIN IMMEDIATE, so SQLITE_BUSY (and its extended variants) is
    # reachable at begin() time and genuinely retries against this
    # timeout for up to 5s before raising — under the old deferred
    # BEGIN, a stale-snapshot-upgrade failure (SQLITE_BUSY_SNAPSHOT)
    # bypassed the busy handler entirely, so no busy_timeout value,
    # explicit or not, would have helped against the specific failure
    # this KI closes — see begin()'s own docstring for the full
    # explanation.
    self.conn = sqlite3.connect(
        str(self.path), timeout=5.0, isolation_level=None, check_same_thread=False
    )
    self.conn.row_factory = sqlite3.Row
    # ADR-0052: refuses (SchemaError) an existing file below
    # migrations.CURRENT_FORMAT_VERSION rather than silently applying
    # pending migrations — see require_current_format's own docstring.
    # Runs before every PRAGMA/DDL below, not just _create_schema() —
    # round-1 review found `PRAGMA journal_mode = WAL` (further down)
    # still ran ahead of this check, switching a stale file's on-disk
    # journal mode before refusing to open it, which the neighboring
    # "a stale file must not have any DDL touch it on this connect at
    # all" claim didn't actually hold for a session-level PRAGMA like
    # this one. Closes the connection on *any* failure here, not just
    # SchemaError (a corrupt/non-database file raises sqlite3.Error
    # instead) — a refused open shouldn't leak a live handle either way.
    try:
        migrations.require_current_format(self.conn.cursor(), path=self.path)
    except BaseException:
        self.conn.close()
        raise
    self.conn.execute("PRAGMA foreign_keys = ON")
    # KI-066 review: recursive_triggers defaults OFF, and SQLite only
    # fires a BEFORE DELETE trigger for an `INSERT OR REPLACE`
    # conflict-row removal when this is ON — without it, `INSERT OR
    # REPLACE INTO assertion_event ...` with an existing id silently
    # rewrites the row (including `actor`, laundering attribution)
    # instead of tripping trg_assertion_event_no_delete/
    # trg_proposal_event_no_delete below. Verified empirically: the
    # bypass reproduces with this OFF and is blocked with it ON.
    self.conn.execute("PRAGMA recursive_triggers = ON")
    # entities_where()'s __contains filter (KI-039) uses LIKE; SQLite's
    # default LIKE is ASCII-case-insensitive, DuckDB's is case-sensitive
    # — without this, the same .where(x__contains=...) call would
    # silently return different result sets per backend. Case-sensitive
    # matches DuckDB's default and is the less surprising choice for a
    # substring filter (matches Python's own `in` semantics).
    self.conn.execute("PRAGMA case_sensitive_like = ON")
    # SPEC §12.1 MUST: default backend uses WAL mode — readers don't
    # block behind writers, which matters for a long-lived MCP server
    # process handling concurrent tool calls. No-op (falls back to a
    # different mode) for in-memory/`:memory:` databases.
    self.conn.execute("PRAGMA journal_mode = WAL")
    self.conn.enable_load_extension(True)
    sqlite_vec.load(self.conn)
    self.conn.enable_load_extension(False)
    self._in_transaction: bool = False
    # Serializes all access to self.conn across threads (KI-023): the
    # connection and _in_transaction are shared mutable state that stock
    # sqlite3 does not protect once check_same_thread=False lifts the
    # same-thread check. RLock (not Lock): begin() holds it across
    # multiple public-method calls inside a transaction() block, each of
    # which re-acquires it via the @_synchronized decorator.
    self._lock = threading.RLock()
    self._clock: Clock = clock or SystemClock()
    self._create_schema()

begin

begin() -> None

Begin an explicit transaction (ADR-0010, KI-084).

Acquires self._lock (KI-023) — held across every subsequent @_synchronized call until commit()/rollback() releases it, so no other thread's operation can interleave with this transaction. Only serializes this process's own threads — see KI-084's docs/adr/ADR-0001-storage-default.md update for the cross-process constraint this alone doesn't cover.

BEGIN IMMEDIATE, not a plain deferred BEGIN (KI-084): a deferred transaction takes its read snapshot lazily, on first statement. assert_literal/assert_ref's own conflict-routing read (SPEC §10) and propose/propose_ref's auto-accept branch, accept_proposal, and resubmit's auto-accept branch (via _replay_proposal_operations) all do that read inside the transaction() block this method opens — so a deferred BEGIN let a concurrent writer (a second OS process; self._lock above only protects this process's own threads) commit between that read and this connection's own later write. The resulting write then hit SQLITE_BUSY_SNAPSHOT — a stale-snapshot-upgrade failure SQLite deliberately never routes through the busy handler, so it failed immediately no matter how long busy_timeout (set in __init__) allowed. BEGIN IMMEDIATE claims the write lock right here, before any read this transaction goes on to do, so a concurrent writer is instead serialized behind it — blocked and retried by the busy handler, same as any other reachable SQLITE_BUSY, for up to busy_timeout before genuinely failing.

This is not a blanket claim that every Ontology write is now cross-process-safe — see KI-084's docs/known-issues.md entry and its own KI-092 follow-up for exactly which write paths this does and doesn't reach (several either write outside any transaction() block at all, or, per the already-resolved KI-035, evaluate policy against a read taken before the transaction opens).

Trade-off worth knowing (KI-084 review): self._lock is acquired before the BEGIN IMMEDIATE call below, so while this call is parked in SQLite's busy handler waiting out a cross-process writer, every other @_synchronized call on this process — reads included — blocks behind it too, for up to the full busy_timeout. This can't be avoided by acquiring the lock later: only one Python thread may safely touch the single shared self.conn at a time regardless of which statement is running, so narrowing the lock's span here would just reopen KI-023 (two threads issuing statements on one connection concurrently) instead. The WAL claim elsewhere in this file ("readers don't block behind writers") holds at the SQLite level; it does not hold at this process's own read availability once a begin() here is genuinely contended by another process. See KI-084's docs/adr/ADR-0001-storage-default.md update for the deployment-facing version of this same trade-off.

Source code in src/ontolith/store/sqlite/backend.py

def begin(self) -> None:
    """Begin an explicit transaction (ADR-0010, KI-084).

    Acquires self._lock (KI-023) — held across every subsequent
    @_synchronized call until commit()/rollback() releases it, so no
    other thread's operation can interleave with this transaction.
    Only serializes *this process*'s own threads — see KI-084's
    `docs/adr/ADR-0001-storage-default.md` update for the
    cross-process constraint this alone doesn't cover.

    ``BEGIN IMMEDIATE``, not a plain deferred ``BEGIN`` (KI-084): a
    deferred transaction takes its read snapshot lazily, on first
    statement. `assert_literal`/`assert_ref`'s own conflict-routing
    read (SPEC §10) and `propose`/`propose_ref`'s auto-accept branch,
    `accept_proposal`, and `resubmit`'s auto-accept branch (via
    `_replay_proposal_operations`) all do that read *inside* the
    `transaction()` block this method opens — so a deferred `BEGIN`
    let a concurrent writer (a second OS process; `self._lock` above
    only protects this process's own threads) commit between that read
    and this connection's own later write. The resulting write then
    hit `SQLITE_BUSY_SNAPSHOT` — a stale-snapshot-upgrade failure
    SQLite deliberately never routes through the busy handler, so it
    failed immediately no matter how long `busy_timeout` (set in
    `__init__`) allowed. `BEGIN IMMEDIATE` claims the write lock right
    here, before any read this transaction goes on to do, so a
    concurrent writer is instead serialized behind it — blocked and
    retried by the busy handler, same as any other reachable
    `SQLITE_BUSY`, for up to `busy_timeout` before genuinely failing.

    This is not a blanket claim that every `Ontology` write is now
    cross-process-safe — see KI-084's `docs/known-issues.md` entry and
    its own KI-092 follow-up for exactly which write paths this does
    and doesn't reach (several either write outside any `transaction()`
    block at all, or, per the already-resolved KI-035, evaluate policy
    against a read taken before the transaction opens).

    Trade-off worth knowing (KI-084 review): `self._lock` is acquired
    *before* the `BEGIN IMMEDIATE` call below, so while this call is
    parked in SQLite's busy handler waiting out a cross-process writer,
    every other `@_synchronized` call on this process — reads included
    — blocks behind it too, for up to the full `busy_timeout`. This
    can't be avoided by acquiring the lock later: only one Python
    thread may safely touch the single shared `self.conn` at a time
    regardless of which statement is running, so narrowing the lock's
    span here would just reopen KI-023 (two threads issuing statements
    on one connection concurrently) instead. The WAL claim elsewhere in
    this file ("readers don't block behind writers") holds at the
    SQLite level; it does not hold at this process's own read
    availability once a `begin()` here is genuinely contended by
    another process. See KI-084's `docs/adr/ADR-0001-storage-default.md`
    update for the deployment-facing version of this same trade-off.
    """
    self._lock.acquire()
    try:
        self.conn.execute("BEGIN IMMEDIATE")
    except sqlite3.OperationalError as e:
        self._lock.release()
        # SQLITE_BUSY and its extended variants (SQLITE_BUSY_SNAPSHOT,
        # SQLITE_BUSY_RECOVERY, SQLITE_BUSY_TIMEOUT) all share primary
        # code 5 in their low byte (extended code & 0xFF == primary
        # code) — Python's sqlite3 does surface the extended code on
        # `.sqlite_errorcode` (verified directly: a snapshot-upgrade
        # failure reports 517, not just 5). `BEGIN IMMEDIATE` above
        # means `SQLITE_BUSY_SNAPSHOT` specifically can no longer occur
        # at *this* call site (the write lock is claimed before any
        # read, so there is no stale snapshot left to upgrade here) —
        # but the masking itself stays as defense-in-depth: it's still
        # what's needed to catch `SQLITE_BUSY_RECOVERY`/`_TIMEOUT`
        # should either become reachable here, and to not silently stop
        # matching if a future SQLite/Python change alters which
        # variant this specific contention surfaces as. Distinguishable
        # from other begin() failures (a lock this connection's own
        # busy_timeout couldn't clear within its window) rather than
        # folded into the same generic message a non-transient failure
        # below would get — still a StorageError (redacted at every
        # interface boundary, KI-083's own precedent for why 5xx
        # messages carry real detail only in server-side logs, never in
        # the response), but a caller reading logs can now tell "this
        # was contention, safe to retry the whole operation" from
        # "something is actually broken" without guessing from the raw
        # sqlite3 message.
        code = getattr(e, "sqlite_errorcode", None)
        if code is not None and code & 0xFF == sqlite3.SQLITE_BUSY:
            raise StorageError(
                f"Transaction start failed due to lock contention (safe to retry): {e}"
            ) from e
        raise StorageError(f"Failed to begin transaction: {e}") from e
    except sqlite3.Error as e:
        self._lock.release()
        raise StorageError(f"Failed to begin transaction: {e}") from e
    self._in_transaction = True

commit

commit() -> None

Commit the current explicit transaction.

Releases self._lock only on success. A failed commit leaves the transaction (and the lock) open: transaction()'s except block calls rollback() next, which is then the sole path that releases the lock — releasing here too on failure would double-release it (the lock is not reentrant-safe against being released twice), raising a RuntimeError that masks the real StorageError and leaves _in_transaction stuck True.

Source code in src/ontolith/store/sqlite/backend.py

def commit(self) -> None:
    """Commit the current explicit transaction.

    Releases self._lock only on success. A failed commit leaves the
    transaction (and the lock) open: transaction()'s except block calls
    rollback() next, which is then the sole path that releases the
    lock — releasing here too on failure would double-release it (the
    lock is not reentrant-safe against being released twice), raising a
    RuntimeError that masks the real StorageError and leaves
    _in_transaction stuck True.
    """
    try:
        self.conn.commit()
    except sqlite3.Error as e:
        raise StorageError(f"Failed to commit transaction: {e}") from e
    self._in_transaction = False
    self._lock.release()

rollback

rollback() -> None

Rollback the current explicit transaction. Always releases self._lock.

Unlike commit(), this always resolves the transaction (successful or not) — it's the terminal cleanup path, including when called after a failed commit() (which deliberately did not release the lock itself, see commit()'s docstring). _in_transaction is reset unconditionally too, even if the underlying rollback itself fails: leaving it True after the lock is released would let a future caller believe it must skip autocommit for a transaction no one will ever commit or roll back again.

Source code in src/ontolith/store/sqlite/backend.py

def rollback(self) -> None:
    """Rollback the current explicit transaction. Always releases self._lock.

    Unlike commit(), this always resolves the transaction (successful
    or not) — it's the terminal cleanup path, including when called
    after a failed commit() (which deliberately did not release the
    lock itself, see commit()'s docstring). _in_transaction is reset
    unconditionally too, even if the underlying rollback itself fails:
    leaving it True after the lock is released would let a future
    caller believe it must skip autocommit for a transaction no one
    will ever commit or roll back again.
    """
    try:
        self.conn.rollback()
    except sqlite3.Error as e:
        raise StorageError(f"Failed to rollback transaction: {e}") from e
    finally:
        self._in_transaction = False
        self._lock.release()

transaction

transaction() -> Iterator[None]

Context manager for atomic multi-write transactions (ADR-0010).

Usage

with backend.transaction(): backend.put_entity(entity) backend.put_assertion(assertion)

Source code in src/ontolith/store/sqlite/backend.py

@contextmanager
def transaction(self) -> Iterator[None]:
    """Context manager for atomic multi-write transactions (ADR-0010).

    Usage:
        with backend.transaction():
            backend.put_entity(entity)
            backend.put_assertion(assertion)
    """
    self.begin()
    try:
        yield
        self.commit()
    except Exception:
        self.rollback()
        raise

put_principal

put_principal(principal: Principal) -> None

Persist a principal.

Parameters:

Name	Type	Description	Default
principal	Principal	Principal to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_principal(self, principal: Principal) -> None:
    """Persist a principal.

    Args:
        principal: Principal to persist

    Raises:
        StorageError: If persistence fails
    """
    import json

    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO principal (id, kind, owner, auth_method, default_capability, trust_level, created_at, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                principal.id,
                principal.kind,
                principal.owner,
                principal.auth_method,
                principal.default_capability,
                principal.trust_level,
                principal.created_at.isoformat(),
                json.dumps(principal.metadata),
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Principal conflict (id={principal.id}): {e}") from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist principal (id={principal.id}): {e}") from e

get_principal

get_principal(principal_id: str) -> Principal | None

Retrieve a principal by ID.

Parameters:

Name	Type	Description	Default
principal_id	str	Principal ID to retrieve	required

Returns:

Type	Description
Principal | None	Principal if found, None otherwise

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_principal(self, principal_id: str) -> Principal | None:
    """Retrieve a principal by ID.

    Args:
        principal_id: Principal ID to retrieve

    Returns:
        Principal if found, None otherwise
    """
    cursor = self.conn.cursor()
    cursor.execute(
        "SELECT * FROM principal WHERE id = ?",
        (principal_id,),
    )
    row = cursor.fetchone()
    if row is None:
        return None
    return self._row_to_principal(row)

list_principals

list_principals() -> list[Principal]

List all principals (KI-022).

Returns:

Type	Description
list[Principal]	All principals, most recently created first

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def list_principals(self) -> list[Principal]:
    """List all principals (KI-022).

    Returns:
        All principals, most recently created first
    """
    cursor = self.conn.cursor()
    # created_at DESC for most-recent-first, same as
    # get_credentials_for_principal. The id DESC tiebreak means
    # something different here, though: credential ids are ULIDs
    # (id DESC ~= recency), but principal ids are user-supplied
    # emails/slugs — id DESC is just a deterministic lexical tiebreak
    # for same-timestamp rows, not a recency proxy.
    cursor.execute("SELECT * FROM principal ORDER BY created_at DESC, id DESC")
    return [self._row_to_principal(row) for row in cursor.fetchall()]

list_namespaces

list_namespaces() -> list[Namespace]

List all registered namespaces (SPEC §12.2, KI-022).

Returns:

Type	Description
list[Namespace]	All namespaces, most recently created first

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def list_namespaces(self) -> list[Namespace]:
    """List all registered namespaces (SPEC §12.2, KI-022).

    Returns:
        All namespaces, most recently created first
    """
    cursor = self.conn.cursor()
    # id DESC tiebreak is purely lexical (namespace ids are slugs, not
    # ULIDs) — same convention as list_principals, not a recency proxy.
    cursor.execute("SELECT * FROM namespace ORDER BY created_at DESC, id DESC")
    return [self._row_to_namespace(row) for row in cursor.fetchall()]

put_credential

put_credential(credential: PrincipalCredential) -> None

Persist a principal credential (hashed API-key token).

Parameters:

Name	Type	Description	Default
credential	PrincipalCredential	PrincipalCredential to persist (token_hash, never the raw token)	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_credential(self, credential: PrincipalCredential) -> None:
    """Persist a principal credential (hashed API-key token).

    Args:
        credential: PrincipalCredential to persist (token_hash, never the
            raw token)

    Raises:
        StorageError: If persistence fails
    """
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO principal_credential
                (id, principal_id, token_hash, created_at, revoked_at, issued_by, revoked_by)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            (
                credential.id,
                credential.principal_id,
                credential.token_hash,
                credential.created_at.isoformat(),
                credential.revoked_at.isoformat() if credential.revoked_at else None,
                credential.issued_by,
                credential.revoked_by,
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Credential conflict (id={credential.id}): {e}") from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist credential (id={credential.id}): {e}") from e

get_principal_by_token_hash

get_principal_by_token_hash(
    token_hash: str,
) -> Principal | None

Resolve a principal via a credential's token hash.

Only unrevoked credentials resolve. This is the sole read path used for MCP authentication — it never trusts a caller-supplied principal ID directly.

Parameters:

Name	Type	Description	Default
token_hash	str	SHA-256 hash of the raw bearer token	required

Returns:

Type	Description
Principal | None	Principal if the hash matches an active (unrevoked) credential,
Principal | None	None otherwise

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_principal_by_token_hash(self, token_hash: str) -> Principal | None:
    """Resolve a principal via a credential's token hash.

    Only unrevoked credentials resolve. This is the sole read path used
    for MCP authentication — it never trusts a caller-supplied principal
    ID directly.

    Args:
        token_hash: SHA-256 hash of the raw bearer token

    Returns:
        Principal if the hash matches an active (unrevoked) credential,
        None otherwise
    """
    cursor = self.conn.cursor()
    cursor.execute(
        """
        SELECT p.* FROM principal p
        JOIN principal_credential c ON c.principal_id = p.id
        WHERE c.token_hash = ? AND c.revoked_at IS NULL
        """,
        (token_hash,),
    )
    row = cursor.fetchone()
    if row is None:
        return None
    return self._row_to_principal(row)

get_credential

get_credential(
    credential_id: str,
) -> PrincipalCredential | None

Retrieve a credential by ID (never exposes the raw token or hash to callers).

Parameters:

Name	Type	Description	Default
credential_id	str	Credential ID to retrieve	required

Returns:

Type	Description
PrincipalCredential | None	PrincipalCredential if found, None otherwise

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_credential(self, credential_id: str) -> PrincipalCredential | None:
    """Retrieve a credential by ID (never exposes the raw token or hash to callers).

    Args:
        credential_id: Credential ID to retrieve

    Returns:
        PrincipalCredential if found, None otherwise
    """
    cursor = self.conn.cursor()
    cursor.execute(
        "SELECT * FROM principal_credential WHERE id = ?",
        (credential_id,),
    )
    row = cursor.fetchone()
    if row is None:
        return None
    return self._row_to_credential(row)

get_credentials_for_principal

get_credentials_for_principal(
    principal_id: str,
) -> list[PrincipalCredential]

List all credentials (active and revoked) issued to a principal.

Parameters:

Name	Type	Description	Default
principal_id	str	Principal to list credentials for	required

Returns:

Type	Description
list[PrincipalCredential]	Credentials for this principal, most recently issued first

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_credentials_for_principal(self, principal_id: str) -> list[PrincipalCredential]:
    """List all credentials (active and revoked) issued to a principal.

    Args:
        principal_id: Principal to list credentials for

    Returns:
        Credentials for this principal, most recently issued first
    """
    cursor = self.conn.cursor()
    # id DESC tiebreaks two credentials issued at the same timestamp
    # (coarse/injected Clock) deterministically, so this listing has a
    # stable, reproducible order (ids are monotonically assigned).
    # issue_token() itself returns its new credential's id directly
    # (KI-024) rather than relying on this ordering to recover it.
    cursor.execute(
        "SELECT * FROM principal_credential WHERE principal_id = ? "
        "ORDER BY created_at DESC, id DESC",
        (principal_id,),
    )
    return [self._row_to_credential(row) for row in cursor.fetchall()]

revoke_credential

revoke_credential(
    credential_id: str,
    revoked_at: datetime,
    revoked_by: str,
) -> None

Mark a credential as revoked. Idempotent-safe: re-revoking an already-revoked credential is a true no-op, not a silent re-stamp — it doesn't overwrite revoked_by/revoked_at with a second caller's values (KI-060: that would launder the first revocation's real attribution).

Parameters:

Name	Type	Description	Default
credential_id	str	Credential to revoke	required
revoked_at	datetime	Timestamp of revocation	required
revoked_by	str	Principal ID of the admin performing the revocation	required

Raises:

Type	Description
StorageError	If the credential is not found

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def revoke_credential(self, credential_id: str, revoked_at: datetime, revoked_by: str) -> None:
    """Mark a credential as revoked. Idempotent-safe: re-revoking an
    already-revoked credential is a true no-op, not a silent
    re-stamp — it doesn't overwrite `revoked_by`/`revoked_at` with a
    second caller's values (KI-060: that would launder the first
    revocation's real attribution).

    Args:
        credential_id: Credential to revoke
        revoked_at: Timestamp of revocation
        revoked_by: Principal ID of the admin performing the revocation

    Raises:
        StorageError: If the credential is not found
    """
    cursor = self.conn.cursor()
    cursor.execute(
        "UPDATE principal_credential SET revoked_at = ?, revoked_by = ? "
        "WHERE id = ? AND revoked_at IS NULL",
        (revoked_at.isoformat(), revoked_by, credential_id),
    )
    if cursor.rowcount == 0:
        exists = cursor.execute(
            "SELECT 1 FROM principal_credential WHERE id = ?", (credential_id,)
        ).fetchone()
        if exists is None:
            raise StorageError(f"Credential not found: {credential_id}")
        # Already revoked - no-op, first revocation's attribution stands.
    if not self._in_transaction:
        self.conn.commit()

put_admin_event

put_admin_event(event: AdminEvent) -> None

Persist an append-only admin-action event (KI-060).

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_admin_event(self, event: AdminEvent) -> None:
    """Persist an append-only admin-action event (KI-060)."""
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO admin_event (id, actor, action, target, at, detail)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                event.id,
                event.actor,
                event.action,
                event.target,
                event.at.isoformat(),
                event.detail,
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Admin event conflict (id={event.id}): {e}") from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist admin event (id={event.id}): {e}") from e

get_admin_events

get_admin_events(
    actor: str | None = None, target: str | None = None
) -> list[AdminEvent]

Retrieve admin events, optionally filtered by actor or target, oldest first.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_admin_events(
    self, actor: str | None = None, target: str | None = None
) -> list[AdminEvent]:
    """Retrieve admin events, optionally filtered by actor or target, oldest first."""
    cursor = self.conn.cursor()
    query = "SELECT * FROM admin_event WHERE 1=1"
    params: list[str] = []
    if actor is not None:
        query += " AND actor = ?"
        params.append(actor)
    if target is not None:
        query += " AND target = ?"
        params.append(target)
    query += " ORDER BY at ASC, id ASC"
    cursor.execute(query, params)
    return [
        AdminEvent(
            id=row["id"],
            actor=row["actor"],
            action=row["action"],
            target=row["target"],
            at=datetime.fromisoformat(row["at"]),
            detail=row["detail"],
        )
        for row in cursor.fetchall()
    ]

put_entity

put_entity(entity: Entity) -> None

Persist an entity.

Parameters:

Name	Type	Description	Default
entity	Entity	Entity to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_entity(self, entity: Entity) -> None:
    """Persist an entity.

    Args:
        entity: Entity to persist

    Raises:
        StorageError: If persistence fails
    """
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO entity (id, namespace, concept, natural_key, created_at, created_by)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                entity.id,
                entity.namespace,
                entity.concept,
                entity.natural_key,
                entity.created_at.isoformat(),
                entity.created_by,
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Entity conflict: {e}") from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist entity: {e}") from e

put_assertion

put_assertion(assertion: Assertion) -> None

Persist an assertion.

Parameters:

Name	Type	Description	Default
assertion	Assertion	Assertion to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_assertion(self, assertion: Assertion) -> None:
    """Persist an assertion.

    Args:
        assertion: Assertion to persist

    Raises:
        StorageError: If persistence fails
    """
    import json

    # Map unified value field to value_lit/value_ref based on kind
    value_lit = assertion.value if assertion.value_kind == "literal" else None
    value_ref = assertion.value if assertion.value_kind == "ref" else None

    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO assertion (
                id, namespace, subject, predicate,
                value_kind, value_type, value_lit, value_ref,
                author, acting_as, source, confidence, rationale, model,
                asserted_at, valid_from, valid_to,
                status, proposal_id, supersedes, metadata
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                assertion.id,
                assertion.namespace,
                assertion.subject,
                assertion.predicate,
                assertion.value_kind,
                assertion.value_type,
                value_lit,
                value_ref,
                assertion.author,
                assertion.acting_as,
                assertion.source,
                assertion.confidence,
                assertion.rationale,
                assertion.model,
                assertion.asserted_at.isoformat(),
                assertion.valid_from.isoformat() if assertion.valid_from else None,
                assertion.valid_to.isoformat() if assertion.valid_to else None,
                assertion.status,
                assertion.proposal_id,
                assertion.supersedes,
                json.dumps(assertion.metadata),
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(
            f"Assertion conflict (id={assertion.id}, subject={assertion.subject}): {e}"
        ) from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist assertion (id={assertion.id}): {e}") from e

get_entity

get_entity(entity_id: str) -> Entity | None

Retrieve an entity by ID.

Parameters:

Name	Type	Description	Default
entity_id	str	Entity ID to retrieve	required

Returns:

Type	Description
Entity | None	Entity if found, None otherwise

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_entity(self, entity_id: str) -> Entity | None:
    """Retrieve an entity by ID.

    Args:
        entity_id: Entity ID to retrieve

    Returns:
        Entity if found, None otherwise
    """
    cursor = self.conn.cursor()
    cursor.execute(
        "SELECT * FROM entity WHERE id = ?",
        (entity_id,),
    )
    row = cursor.fetchone()
    if row is None:
        return None

    return Entity(
        id=row["id"],
        namespace=row["namespace"],
        concept=row["concept"],
        natural_key=row["natural_key"],
        created_at=datetime.fromisoformat(row["created_at"]),
        created_by=row["created_by"],
    )

get_entity_by_natural_key

get_entity_by_natural_key(
    namespace: str, concept: str, natural_key: str
) -> Entity | None

Retrieve an entity by its unique (namespace, concept, natural_key) triple.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_entity_by_natural_key(
    self, namespace: str, concept: str, natural_key: str
) -> Entity | None:
    """Retrieve an entity by its unique (namespace, concept, natural_key) triple."""
    cursor = self.conn.cursor()
    cursor.execute(
        "SELECT * FROM entity WHERE namespace = ? AND concept = ? AND natural_key = ?",
        (namespace, concept, natural_key),
    )
    row = cursor.fetchone()
    if row is None:
        return None

    return Entity(
        id=row["id"],
        namespace=row["namespace"],
        concept=row["concept"],
        natural_key=row["natural_key"],
        created_at=datetime.fromisoformat(row["created_at"]),
        created_by=row["created_by"],
    )

assertions

assertions(
    subject: str | None = None,
    predicate: str | None = None,
    status: str | None = "active",
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Assertion]

Query assertions with optional filters.

Parameters:

Name	Type	Description	Default
subject	str | None	Filter by subject entity ID	None
predicate	str | None	Filter by predicate	None
status	str | None	Filter by current status (ignored when as_of_time is set). Defaults to "active"; pass status=None for every status.	'active'
as_of_time	datetime | None	If set, applies bitemporal filter: asserted_at <= t AND valid_from <= t AND (valid_to IS NULL OR valid_to > t)	None
include_flagged	bool	When as_of_time is set, whether to include 'flagged' assertions (excluded by default — a flagged assertion is disputed, not confirmed-valid; pass True for explicit audit/history views); ignored when as_of_time is None.	False
include_history	bool	When as_of_time is set, whether to opt back into seeing a 'retracted' assertion once its own retraction event's timestamp is <= as_of_time (excluded by default, ADR-0049/KI-095) — mirrors include_flagged's shape (KI-098); ignored when as_of_time is None.	False

Returns:

Type	Description
list[Assertion]	List of matching assertions

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def assertions(
    self,
    subject: str | None = None,
    predicate: str | None = None,
    status: str | None = "active",
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Assertion]:
    """Query assertions with optional filters.

    Args:
        subject: Filter by subject entity ID
        predicate: Filter by predicate
        status: Filter by current status (ignored when as_of_time is
            set). Defaults to "active"; pass status=None for every
            status.
        as_of_time: If set, applies bitemporal filter:
            asserted_at <= t AND valid_from <= t AND (valid_to IS NULL OR valid_to > t)
        include_flagged: When as_of_time is set, whether to include
            'flagged' assertions (excluded by default — a flagged
            assertion is disputed, not confirmed-valid; pass True for
            explicit audit/history views); ignored when as_of_time is
            None.
        include_history: When as_of_time is set, whether to opt back
            into seeing a 'retracted' assertion once its own retraction
            event's timestamp is <= as_of_time (excluded by default,
            ADR-0049/KI-095) — mirrors include_flagged's shape (KI-098);
            ignored when as_of_time is None.

    Returns:
        List of matching assertions
    """
    query = "SELECT * FROM assertion WHERE 1=1"
    params: list[str] = []

    if subject is not None:
        query += " AND subject = ?"
        params.append(subject)

    if predicate is not None:
        query += " AND predicate = ?"
        params.append(predicate)

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        query += " AND asserted_at <= ?"
        params.append(t_iso)
        query += " AND (valid_from IS NULL OR valid_from <= ?)"
        params.append(t_iso)
        query += " AND (valid_to IS NULL OR valid_to > ?)"
        params.append(t_iso)
        if not include_flagged:
            # Flagged-at-t, not current status: a static conflict flags an
            # assertion permanently (no valid_to change), so using current
            # status here would hide it from as_of() queries for times
            # before the dispute existed. Reconstruct from the event log
            # instead — every flagged transition (including an assertion
            # born already-flagged) has a 'flagged' event, see
            # Ontology._apply_with_conflict_routing. Tiebreak on ae.id:
            # two events can share the same `at` under a clock that
            # hasn't advanced (e.g. flag-then-resolve in the same tick),
            # and `at` alone would make "last recorded wins"
            # nondeterministic. Under SequentialIdProvider/FixedIdProvider
            # (used in tests) id order matches recording order exactly;
            # under the production UlidProvider, id is monotonic across
            # milliseconds but not guaranteed within one, so same-`at`
            # AND same-millisecond ties are a residual (low-probability,
            # not exploitable) nondeterminism.
            query += """ AND COALESCE(
                (SELECT ae.action FROM assertion_event ae
                 WHERE ae.assertion_id = assertion.id AND ae.at <= ?
                   AND ae.action IN ('flagged', 'reactivated')
                 ORDER BY ae.at DESC, ae.id DESC LIMIT 1),
                'reactivated'
            ) != 'flagged'"""
            params.append(t_iso)
        # ADR-0049 (KI-095): same retraction-aware exclusion the
        # entities_where()/entities_meeting_confidence()/
        # entities_meeting_trust() family has — see entities_where()'s
        # comment for the full reasoning (KI-051 guarantees at most one
        # 'retracted' event per assertion, so unlike flagged/reactivated
        # above this needs no ORDER BY tiebreak: retraction is a
        # one-way terminal transition, never followed by another event).
        # include_history opts back out of this check, the same
        # mechanism the QueryBuilder-facing trio already uses (KI-098 —
        # this method had no such opt-out when ADR-0049 first shipped).
        if not include_history:
            query += (
                " AND (status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = assertion.id"
                " AND ae.action = 'retracted' AND ae.at > ?"
                "))"
            )
            params.append(t_iso)
    elif status is not None:
        query += " AND status = ?"
        params.append(status)

    cursor = self.conn.cursor()
    cursor.execute(query, params)

    return [self._row_to_assertion(row) for row in cursor.fetchall()]

get_assertion

get_assertion(assertion_id: str) -> Assertion | None

Retrieve a single assertion by ID, regardless of status.

Parameters:

Name	Type	Description	Default
assertion_id	str	Assertion ID to retrieve	required

Returns:

Type	Description
Assertion | None	Assertion if found, None otherwise

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_assertion(self, assertion_id: str) -> Assertion | None:
    """Retrieve a single assertion by ID, regardless of status.

    Args:
        assertion_id: Assertion ID to retrieve

    Returns:
        Assertion if found, None otherwise
    """
    cursor = self.conn.cursor()
    cursor.execute("SELECT * FROM assertion WHERE id = ?", (assertion_id,))
    row = cursor.fetchone()
    return self._row_to_assertion(row) if row else None

set_assertion_status

set_assertion_status(
    assertion_id: str,
    status: str,
    valid_to: str | None = None,
) -> None

Update assertion status and optionally close validity window.

This is the ONLY allowed mutation on assertions (append-only invariant).

Parameters:

Name	Type	Description	Default
assertion_id	str	Assertion ID to update	required
status	str	New status (superseded, retracted, flagged)	required
valid_to	str | None	Optional validity end time (ISO format)	None

Raises:

Type	Description
StorageError	If update fails or assertion not found

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def set_assertion_status(
    self,
    assertion_id: str,
    status: str,
    valid_to: str | None = None,
) -> None:
    """Update assertion status and optionally close validity window.

    This is the ONLY allowed mutation on assertions (append-only invariant).

    Args:
        assertion_id: Assertion ID to update
        status: New status (superseded, retracted, flagged)
        valid_to: Optional validity end time (ISO format)

    Raises:
        StorageError: If update fails or assertion not found
    """
    try:
        cursor = self.conn.cursor()
        if valid_to is not None:
            cursor.execute(
                "UPDATE assertion SET status = ?, valid_to = ? WHERE id = ?",
                (status, valid_to, assertion_id),
            )
        else:
            cursor.execute(
                "UPDATE assertion SET status = ? WHERE id = ?",
                (status, assertion_id),
            )

        if cursor.rowcount == 0:
            raise StorageError(f"Assertion not found: {assertion_id}")
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.Error as e:
        raise StorageError(f"Failed to update assertion status (id={assertion_id}): {e}") from e

put_schema

put_schema(schema: SchemaIR) -> None

Persist a schema version.

Also registers schema.namespace in the namespace registry if not already present (KI-022) — a namespace that only ever has a schema applied, never an entity, is still discoverable via list_namespaces().

Parameters:

Name	Type	Description	Default
schema	SchemaIR	Schema to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_schema(self, schema: SchemaIR) -> None:
    """Persist a schema version.

    Also registers ``schema.namespace`` in the namespace registry if
    not already present (KI-022) — a namespace that only ever has a
    schema applied, never an entity, is still discoverable via
    `list_namespaces()`.

    Args:
        schema: Schema to persist

    Raises:
        StorageError: If persistence fails
    """
    import json

    try:
        cursor = self.conn.cursor()
        self._ensure_namespace_registered(cursor, schema.namespace)
        cursor.execute(
            """
            INSERT INTO schema_version (namespace, version, definition, applied_at)
            VALUES (?, ?, ?, ?)
            """,
            (
                schema.namespace,
                schema.version,
                json.dumps(schema.to_json()),
                self._clock.now().isoformat(),
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(
            f"Schema conflict (namespace={schema.namespace}, version={schema.version}): {e}"
        ) from e
    except sqlite3.Error as e:
        raise StorageError(
            f"Failed to persist schema (namespace={schema.namespace}): {e}"
        ) from e

get_schema

get_schema(
    namespace: str, version: int | None = None
) -> SchemaIR | None

Retrieve a schema version.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to query	required
version	int | None	Specific version, or None for latest	None

Returns:

Type	Description
SchemaIR | None	Schema if found, None otherwise

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_schema(self, namespace: str, version: int | None = None) -> SchemaIR | None:
    """Retrieve a schema version.

    Args:
        namespace: Namespace to query
        version: Specific version, or None for latest

    Returns:
        Schema if found, None otherwise
    """
    import json

    cursor = self.conn.cursor()

    if version is None:
        # Get latest version
        cursor.execute(
            """
            SELECT definition FROM schema_version
            WHERE namespace = ?
            ORDER BY version DESC
            LIMIT 1
            """,
            (namespace,),
        )
    else:
        # Get specific version
        cursor.execute(
            """
            SELECT definition FROM schema_version
            WHERE namespace = ? AND version = ?
            """,
            (namespace, version),
        )

    row = cursor.fetchone()
    if row is None:
        return None

    definition = json.loads(row["definition"])
    return SchemaIR.from_json(definition)

get_schema_at

get_schema_at(
    namespace: str, at: datetime
) -> SchemaIR | None

Retrieve the schema version effective at a point in time (KI-019).

Orders by applied_at (the actual "effective at" moment), with version as a tiebreak for same-timestamp rows under a coarse or injected Clock — not by version alone, so this stays correct even if a future write path ever persisted schema rows out of temporal order relative to their version numbers.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_schema_at(self, namespace: str, at: datetime) -> SchemaIR | None:
    """Retrieve the schema version effective at a point in time (KI-019).

    Orders by applied_at (the actual "effective at" moment), with
    version as a tiebreak for same-timestamp rows under a coarse or
    injected Clock — not by version alone, so this stays correct even
    if a future write path ever persisted schema rows out of temporal
    order relative to their version numbers.
    """
    import json

    cursor = self.conn.cursor()
    cursor.execute(
        """
        SELECT definition FROM schema_version
        WHERE namespace = ? AND applied_at <= ?
        ORDER BY applied_at DESC, version DESC
        LIMIT 1
        """,
        (namespace, at.isoformat()),
    )
    row = cursor.fetchone()
    if row is None:
        return None

    definition = json.loads(row["definition"])
    return SchemaIR.from_json(definition)

entities

entities(
    namespace: str | None = None,
    concept: str | None = None,
    as_of_time: datetime | None = None,
) -> list[Entity]

Query entities with optional filters.

Parameters:

Name	Type	Description	Default
namespace	str | None	Filter by namespace	None
concept	str | None	Filter by concept	None
as_of_time	datetime | None	If set, exclude entities created after this time	None

Returns:

Type	Description
list[Entity]	List of matching entities

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def entities(
    self,
    namespace: str | None = None,
    concept: str | None = None,
    as_of_time: datetime | None = None,
) -> list[Entity]:
    """Query entities with optional filters.

    Args:
        namespace: Filter by namespace
        concept: Filter by concept
        as_of_time: If set, exclude entities created after this time

    Returns:
        List of matching entities
    """
    query = "SELECT * FROM entity WHERE 1=1"
    params: list[str] = []

    if namespace is not None:
        query += " AND namespace = ?"
        params.append(namespace)

    if concept is not None:
        query += " AND concept = ?"
        params.append(concept)

    if as_of_time is not None:
        query += " AND created_at <= ?"
        params.append(as_of_time.isoformat())

    cursor = self.conn.cursor()
    cursor.execute(query, params)

    results = []
    for row in cursor.fetchall():
        results.append(
            Entity(
                id=row["id"],
                namespace=row["namespace"],
                concept=row["concept"],
                natural_key=row["natural_key"],
                created_at=datetime.fromisoformat(row["created_at"]),
                created_by=row["created_by"],
            )
        )

    return results

put_proposal

put_proposal(proposal: Proposal) -> None

Persist a proposal.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_proposal(self, proposal: Proposal) -> None:
    """Persist a proposal."""
    import json

    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO proposal (id, namespace, author, acting_as, state,
                created_at, decided_at, policy_reason, reviewers, payload, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                proposal.id,
                proposal.namespace,
                proposal.author,
                proposal.acting_as,
                proposal.state,
                proposal.created_at.isoformat(),
                proposal.decided_at.isoformat() if proposal.decided_at else None,
                proposal.policy_reason,
                json.dumps(proposal.reviewers),
                json.dumps(proposal.payload),
                json.dumps(proposal.metadata),
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Proposal conflict (id={proposal.id}): {e}") from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist proposal (id={proposal.id}): {e}") from e

get_proposal

get_proposal(proposal_id: str) -> Proposal | None

Retrieve a proposal by ID.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_proposal(self, proposal_id: str) -> Proposal | None:
    """Retrieve a proposal by ID."""
    cursor = self.conn.cursor()
    cursor.execute("SELECT * FROM proposal WHERE id = ?", (proposal_id,))
    row = cursor.fetchone()
    return self._row_to_proposal(row) if row else None

proposals

proposals(state: str | None = None) -> list[Proposal]

Query proposals, optionally filtered by state (SPEC §14.1).

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def proposals(self, state: str | None = None) -> list[Proposal]:
    """Query proposals, optionally filtered by state (SPEC §14.1)."""
    cursor = self.conn.cursor()
    if state is not None:
        cursor.execute(
            "SELECT * FROM proposal WHERE state = ? ORDER BY created_at DESC, id DESC",
            (state,),
        )
    else:
        cursor.execute("SELECT * FROM proposal ORDER BY created_at DESC, id DESC")
    return [self._row_to_proposal(row) for row in cursor.fetchall()]

update_proposal_state

update_proposal_state(
    proposal_id: str,
    state: str,
    decided_at: str | None = None,
    policy_reason: str | None = None,
) -> None

Update proposal state after policy decision.

policy_reason=None leaves the stored value unchanged (COALESCE), it does not clear it — see the port docstring for why.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def update_proposal_state(
    self,
    proposal_id: str,
    state: str,
    decided_at: str | None = None,
    policy_reason: str | None = None,
) -> None:
    """Update proposal state after policy decision.

    policy_reason=None leaves the stored value unchanged (COALESCE), it
    does not clear it — see the port docstring for why.
    """
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            "UPDATE proposal SET state = ?, decided_at = ?, "
            "policy_reason = COALESCE(?, policy_reason) WHERE id = ?",
            (state, decided_at, policy_reason, proposal_id),
        )
        if cursor.rowcount == 0:
            raise StorageError(f"Proposal not found: {proposal_id}")
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.Error as e:
        raise StorageError(f"Failed to update proposal (id={proposal_id}): {e}") from e

update_proposal_reviewers

update_proposal_reviewers(
    proposal_id: str, reviewers: list[str]
) -> None

Replace a proposal's assigned reviewers (SPEC §9.4's assign action).

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def update_proposal_reviewers(self, proposal_id: str, reviewers: list[str]) -> None:
    """Replace a proposal's assigned reviewers (SPEC §9.4's `assign` action)."""
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            "UPDATE proposal SET reviewers = ? WHERE id = ?",
            (json.dumps(reviewers), proposal_id),
        )
        if cursor.rowcount == 0:
            raise StorageError(f"Proposal not found: {proposal_id}")
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.Error as e:
        raise StorageError(
            f"Failed to update proposal reviewers (id={proposal_id}): {e}"
        ) from e

put_proposal_event

put_proposal_event(event: ProposalEvent) -> None

Persist a structured review-action event (SPEC §9.4).

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_proposal_event(self, event: ProposalEvent) -> None:
    """Persist a structured review-action event (SPEC §9.4)."""
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO proposal_event (id, proposal_id, actor, type, detail, at)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                event.id,
                event.proposal_id,
                event.actor,
                event.type,
                event.detail,
                event.at.isoformat(),
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Proposal event conflict (id={event.id}): {e}") from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist proposal event (id={event.id}): {e}") from e

get_proposal_events

get_proposal_events(
    proposal_id: str,
) -> list[ProposalEvent]

Retrieve all review events for a proposal, oldest first.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_proposal_events(self, proposal_id: str) -> list[ProposalEvent]:
    """Retrieve all review events for a proposal, oldest first."""
    cursor = self.conn.cursor()
    cursor.execute(
        "SELECT * FROM proposal_event WHERE proposal_id = ? ORDER BY at ASC",
        (proposal_id,),
    )
    return [
        ProposalEvent(
            id=row["id"],
            proposal_id=row["proposal_id"],
            actor=row["actor"],
            type=row["type"],
            detail=row["detail"],
            at=datetime.fromisoformat(row["at"]),
        )
        for row in cursor.fetchall()
    ]

put_assertion_event

put_assertion_event(event: AssertionEvent) -> None

Persist an append-only assertion status-mutation event.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_assertion_event(self, event: AssertionEvent) -> None:
    """Persist an append-only assertion status-mutation event."""
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO assertion_event (id, assertion_id, actor, action, at, successor_id)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            (
                event.id,
                event.assertion_id,
                event.actor,
                event.action,
                event.at.isoformat(),
                event.successor_id,
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Assertion event conflict (id={event.id}): {e}") from e
    except sqlite3.Error as e:
        raise StorageError(f"Failed to persist assertion event (id={event.id}): {e}") from e

get_assertion_events

get_assertion_events(
    assertion_id: str,
) -> list[AssertionEvent]

Retrieve all status-mutation events for an assertion, oldest first.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_assertion_events(self, assertion_id: str) -> list[AssertionEvent]:
    """Retrieve all status-mutation events for an assertion, oldest first."""
    cursor = self.conn.cursor()
    cursor.execute(
        "SELECT * FROM assertion_event WHERE assertion_id = ? ORDER BY at ASC",
        (assertion_id,),
    )
    return [self._row_to_assertion_event(row) for row in cursor.fetchall()]

get_assertion_events_by_successor

get_assertion_events_by_successor(
    successor_id: str,
) -> list[AssertionEvent]

Retrieve all 'superseded' events caused by a given successor assertion.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_assertion_events_by_successor(self, successor_id: str) -> list[AssertionEvent]:
    """Retrieve all 'superseded' events caused by a given successor assertion."""
    cursor = self.conn.cursor()
    cursor.execute(
        "SELECT * FROM assertion_event WHERE successor_id = ? ORDER BY at ASC, id ASC",
        (successor_id,),
    )
    return [self._row_to_assertion_event(row) for row in cursor.fetchall()]

put_contradiction

put_contradiction(contradiction: Contradiction) -> None

Persist a new contradiction.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def put_contradiction(self, contradiction: Contradiction) -> None:
    """Persist a new contradiction."""
    import json

    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            INSERT INTO contradiction (id, namespace, subject, predicate, state,
                member_ids, created_at, raised_by, resolved_by, resolved_at, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            (
                contradiction.id,
                contradiction.namespace,
                contradiction.subject,
                contradiction.predicate,
                contradiction.state,
                json.dumps(contradiction.member_ids),
                contradiction.created_at.isoformat(),
                contradiction.raised_by,
                contradiction.resolved_by,
                contradiction.resolved_at.isoformat() if contradiction.resolved_at else None,
                json.dumps(contradiction.metadata),
            ),
        )
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.IntegrityError as e:
        raise StorageError(f"Contradiction conflict (id={contradiction.id}): {e}") from e
    except sqlite3.Error as e:
        raise StorageError(
            f"Failed to persist contradiction (id={contradiction.id}): {e}"
        ) from e

get_open_contradiction

get_open_contradiction(
    namespace: str, subject: str, predicate: str
) -> Contradiction | None

Return the open contradiction for (namespace, subject, predicate), if any.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_open_contradiction(
    self, namespace: str, subject: str, predicate: str
) -> Contradiction | None:
    """Return the open contradiction for (namespace, subject, predicate), if any."""
    cursor = self.conn.cursor()
    cursor.execute(
        """
        SELECT * FROM contradiction
        WHERE namespace = ? AND subject = ? AND predicate = ? AND state = 'open'
        LIMIT 1
        """,
        (namespace, subject, predicate),
    )
    row = cursor.fetchone()
    return self._row_to_contradiction(row) if row else None

update_contradiction_members

update_contradiction_members(
    contradiction_id: str,
    member_ids: list[str],
    metadata: dict[str, Any] | None = None,
) -> None

Add member IDs to an existing open contradiction (KI-071: optionally replace metadata too, in the same UPDATE).

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def update_contradiction_members(
    self,
    contradiction_id: str,
    member_ids: list[str],
    metadata: dict[str, Any] | None = None,
) -> None:
    """Add member IDs to an existing open contradiction (KI-071:
    optionally replace metadata too, in the same UPDATE)."""
    import json

    try:
        cursor = self.conn.cursor()
        if metadata is not None:
            cursor.execute(
                "UPDATE contradiction SET member_ids = ?, metadata = ? WHERE id = ?",
                (json.dumps(member_ids), json.dumps(metadata), contradiction_id),
            )
        else:
            cursor.execute(
                "UPDATE contradiction SET member_ids = ? WHERE id = ?",
                (json.dumps(member_ids), contradiction_id),
            )
        if cursor.rowcount == 0:
            raise StorageError(f"Contradiction not found: {contradiction_id}")
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.Error as e:
        raise StorageError(
            f"Failed to update contradiction (id={contradiction_id}): {e}"
        ) from e

get_contradiction

get_contradiction(
    contradiction_id: str,
) -> Contradiction | None

Retrieve a contradiction by ID, regardless of state.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def get_contradiction(self, contradiction_id: str) -> Contradiction | None:
    """Retrieve a contradiction by ID, regardless of state."""
    cursor = self.conn.cursor()
    cursor.execute("SELECT * FROM contradiction WHERE id = ?", (contradiction_id,))
    row = cursor.fetchone()
    return self._row_to_contradiction(row) if row else None

contradictions

contradictions(
    state: str | None = None,
) -> list[Contradiction]

Query contradictions, optionally filtered by state (SPEC §14.1).

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def contradictions(self, state: str | None = None) -> list[Contradiction]:
    """Query contradictions, optionally filtered by state (SPEC §14.1)."""
    cursor = self.conn.cursor()
    if state is not None:
        cursor.execute(
            "SELECT * FROM contradiction WHERE state = ? ORDER BY created_at DESC, id DESC",
            (state,),
        )
    else:
        cursor.execute("SELECT * FROM contradiction ORDER BY created_at DESC, id DESC")
    return [self._row_to_contradiction(row) for row in cursor.fetchall()]

resolve_contradiction

resolve_contradiction(
    contradiction_id: str,
    resolved_by: str,
    resolved_at: datetime,
) -> None

Mark a contradiction as resolved (SPEC §10.3).

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def resolve_contradiction(
    self,
    contradiction_id: str,
    resolved_by: str,
    resolved_at: datetime,
) -> None:
    """Mark a contradiction as resolved (SPEC §10.3)."""
    try:
        cursor = self.conn.cursor()
        cursor.execute(
            """
            UPDATE contradiction
            SET state = 'resolved', resolved_by = ?, resolved_at = ?
            WHERE id = ?
            """,
            (resolved_by, resolved_at.isoformat(), contradiction_id),
        )
        if cursor.rowcount == 0:
            raise StorageError(f"Contradiction not found: {contradiction_id}")
        if not self._in_transaction:
            self.conn.commit()
    except sqlite3.Error as e:
        raise StorageError(
            f"Failed to resolve contradiction (id={contradiction_id}): {e}"
        ) from e

entities_where

entities_where(
    namespace: str,
    concept: str,
    predicate_filters: list[tuple[str, str, Any]],
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Entity]

Query entities matching all predicate filters in one SQL query.

Uses correlated subqueries so each filter hits the idx_assertion_pred_value/idx_assertion_pred_ref indexes instead of doing one round-trip per entity. "eq" matches either a literal property (value_lit) or a relation's target entity id (value_ref) — KI-030: relation filters like employer="org-123" are equality checks against value_ref, not traversal into the target entity's own properties. The two are checked via a UNION ALL of two single-column point lookups rather than one value_lit = ? OR value_ref = ? predicate — SQLite's planner doesn't reliably pick a seekable plan for the latter (falls back to a full table SCAN on the as_of branch; confirmed via EXPLAIN QUERY PLAN), which turned every .where() call — not just relation filters — into an unindexed scan. "contains"/"gt"/"lt"/"gte"/"lte" (KI-039) check value_lit only — see this port method's own docstring for why relations don't get a UNION ALL branch for those.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to query	required
concept	str	Concept to filter by	required
predicate_filters	list[tuple[str, str, Any]]	List of (full_predicate, operator, value) triples (AND semantics) — see the port method's docstring for the operator set	required
as_of_time	datetime | None	If set, applies bitemporal filter on assertions and entity creation	None
include_flagged	bool	Also match 'flagged' assertions — honored on both the current-state and as_of_time paths (KI-081; excluded by default on both, see assertions()). On the as_of_time path this is point-in-time, not current status (KI-097): reconstructed from the assertion_event log the same way assertions() already does, so a query pinned to a time when an assertion was disputed correctly excludes it even after the dispute has since been resolved — and, the other direction, a time strictly before any dispute existed still includes an otherwise-undisputed value, even though the same assertion is flagged now.	False
include_history	bool	Also match 'superseded'/'retracted' assertions (KI-081). On the current-state path this widens the status set beyond 'active'. On the as_of_time path, 'superseded' is unaffected either way (its window already never restricts to 'active') — but 'retracted' does something under this flag now (ADR-0049, KI-095): as_of_time excludes a retracted assertion once its own retraction event's at is <= as_of_time (a stale, un-narrowed window otherwise keeps matching indefinitely); include_history opts back out of that exclusion.	False

Returns:

Type	Description
list[Entity]	List of entities where all filters match at the given time

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def entities_where(
    self,
    namespace: str,
    concept: str,
    predicate_filters: list[tuple[str, str, Any]],
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Entity]:
    """Query entities matching all predicate filters in one SQL query.

    Uses correlated subqueries so each filter hits the
    idx_assertion_pred_value/idx_assertion_pred_ref indexes instead of
    doing one round-trip per entity. `"eq"` matches either a literal
    property (`value_lit`) or a relation's target entity id
    (`value_ref`) — KI-030: relation filters like `employer="org-123"`
    are equality checks against `value_ref`, not traversal into the
    target entity's own properties. The two are checked via a UNION ALL
    of two single-column point lookups rather than one `value_lit = ?
    OR value_ref = ?` predicate — SQLite's planner doesn't reliably pick
    a seekable plan for the latter (falls back to a full table SCAN on
    the as_of branch; confirmed via EXPLAIN QUERY PLAN), which turned
    every `.where()` call — not just relation filters — into an
    unindexed scan. `"contains"`/`"gt"`/`"lt"`/`"gte"`/`"lte"` (KI-039)
    check `value_lit` only — see this port method's own docstring for
    why relations don't get a UNION ALL branch for those.

    Args:
        namespace: Namespace to query
        concept: Concept to filter by
        predicate_filters: List of `(full_predicate, operator, value)`
            triples (AND semantics) — see the port method's docstring
            for the operator set
        as_of_time: If set, applies bitemporal filter on assertions and entity creation
        include_flagged: Also match 'flagged' assertions — honored on
            both the current-state and as_of_time paths (KI-081;
            excluded by default on both, see assertions()). On the
            as_of_time path this is point-in-time, not current status
            (KI-097): reconstructed from the assertion_event log the
            same way assertions() already does, so a query pinned to a
            time when an assertion *was* disputed correctly excludes
            it even after the dispute has since been resolved — and,
            the other direction, a time strictly before any dispute
            existed still includes an otherwise-undisputed value, even
            though the same assertion is flagged now.
        include_history: Also match 'superseded'/'retracted' assertions
            (KI-081). On the current-state path this widens the status
            set beyond 'active'. On the as_of_time path, 'superseded' is
            unaffected either way (its window already never restricts to
            'active') — but 'retracted' does something under this flag
            now (ADR-0049, KI-095): as_of_time excludes a retracted
            assertion once its own retraction event's `at` is <=
            as_of_time (a stale, un-narrowed window otherwise keeps
            matching indefinitely); include_history opts back out of
            that exclusion.

    Returns:
        List of entities where all filters match at the given time
    """
    query = "SELECT * FROM entity WHERE namespace = ? AND concept = ?"
    params: list[Any] = [namespace, concept]

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        query += " AND created_at <= ?"
        params.append(t_iso)
        # flagged_clause is always one of exactly two hardcoded literals,
        # never caller-controlled. No # nosec needed here (unlike the
        # match_clause consumers below, e.g. `AND id IN (...SELECT...`):
        # bandit's B608 rule only flags a SQL keyword joined into a
        # string via BinOp/.format()/f-string - never a bare literal
        # Constant, which is all flagged_clause/retracted_clause ever
        # are (confirmed by AST: both branches of the ternary are plain
        # folded string constants). The COALESCE subquery below *does*
        # now contain SELECT/FROM/WHERE/ORDER BY/LIMIT (KI-097 - it
        # didn't when this comment was first written), but since that
        # text lives entirely inside the constant rather than being
        # concatenated onto one, bandit's detector still never sees it -
        # confirmed directly, not assumed (a #nosec placed here was
        # previously dead: removing it left bandit's finding count
        # unchanged).
        # KI-097: flagged-at-t, not current status — same event-log
        # reconstruction assertions() already uses (see its own
        # comment for the full reasoning: a static conflict flags an
        # assertion permanently, so using current status would hide it
        # from as_of() queries for times before the dispute existed,
        # and — the bug this KI fixes — would wrongly *include* it for
        # times during a dispute that has since been resolved, since
        # `reactivated` flips current status back to `active`).
        # COALESCE/tiebreak reasoning identical to assertions()'s own.
        # Fail-open on a missing event (COALESCE defaults to
        # 'reactivated', i.e. "not flagged"): a status='flagged' row
        # with no matching event — reachable only via a direct
        # put_assertion() bypassing Ontology, or a pre-existing row
        # from before assertion_event tracked this — is visible at
        # every t. The opposite default from retracted_clause below
        # (fail-closed, ADR-0049), but the correct one here: matches
        # assertions()'s own identical COALESCE, and status='flagged'
        # rows always carry a real 'flagged' event through every
        # write path this codebase has (_apply_with_conflict_routing).
        flagged_clause = (
            ""
            if include_flagged
            else (
                " AND COALESCE("
                "(SELECT ae.action FROM assertion_event ae"
                " WHERE ae.assertion_id = assertion.id AND ae.at <= ?"
                " AND ae.action IN ('flagged', 'reactivated')"
                " ORDER BY ae.at DESC, ae.id DESC LIMIT 1),"
                " 'reactivated'"
                ") != 'flagged'"
            )
        )
        # retracted_clause: same two-hardcoded-literals shape as
        # flagged_clause above, same reasoning for why no #nosec is
        # needed. ADR-0049 (KI-095): a `retracted` assertion's window is
        # not reliably narrowed at retraction time, so `as_of(t)` also
        # excludes it once its own retraction event's `at` is <= t —
        # KI-051 guarantees at most one such event per assertion.
        # `.include_history()` opts back out of this check, the first
        # thing it has ever done on the `as_of` path.
        retracted_clause = (
            ""
            if include_history
            else (
                " AND (status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = assertion.id"
                " AND ae.action = 'retracted' AND ae.at > ?"
                "))"
            )
        )
        match_clause = (
            " AND asserted_at <= ?"
            " AND (valid_from IS NULL OR valid_from <= ?)"
            " AND (valid_to IS NULL OR valid_to > ?)"
            f"{flagged_clause}"
            f"{retracted_clause}"
        )
        match_params = [t_iso, t_iso, t_iso]
        if not include_flagged:
            match_params.append(t_iso)
        if not include_history:
            match_params.append(t_iso)
    else:
        # Current-state: 'active' only by default;
        # .include_flagged()/.include_history() widen the set (KI-081).
        # `status IN (?, …)` is fully parameter-bound — the interpolated
        # piece is only the placeholder string (`?, ?`), built from
        # `len(match_params)`, never caller input.
        match_params = ["active"]
        if include_flagged:
            match_params.append("flagged")
        if include_history:
            match_params += ["superseded", "retracted"]
        match_clause = f" AND status IN ({', '.join(['?'] * len(match_params))})"

    # predicate/value are always bound via `?` below, never
    # interpolated; the two interpolated pieces are match_clause (built
    # from hardcoded literals, see the flagged_clause justification
    # above) and, for range operators, sql_op — a lookup into the
    # closed, module-level _RANGE_SQL_OPERATORS dict, never the
    # caller's raw operator string. Same already-justified pattern, not
    # a new SQL injection surface.
    for predicate, operator, value in predicate_filters:
        if operator == "eq":
            query += (
                " AND id IN ("  # nosec B608
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND value_lit = ?{match_clause}"
                " UNION ALL "
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND value_ref = ?{match_clause}"
                ")"
            )
            params.extend([predicate, value, *match_params, predicate, value, *match_params])
        elif operator == "contains":
            query += (
                " AND id IN ("  # nosec B608
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND value_lit LIKE ? ESCAPE '\\'{match_clause}"
                ")"
            )
            params.extend([predicate, f"%{_like_escape(value)}%", *match_params])
        else:
            # QueryBuilder only validates the predicate's *declared*
            # value_type is Integer/Float, never that already-stored
            # value_lit content actually parses as one (KI-049).
            # SQLite has no TRY_CAST (unlike DuckDB's equivalent
            # branch): CAST('unknown' AS REAL) silently returns 0.0
            # rather than erroring or excluding the row — a known,
            # tracked gap, not something this fix can close without
            # write-time content validation (KI-049), which is a
            # materially different, larger scope than this operator.
            sql_op = _RANGE_SQL_OPERATORS[operator]
            query += (
                " AND id IN ("  # nosec B608
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND CAST(value_lit AS REAL) {sql_op} ?{match_clause}"
                ")"
            )
            params.extend([predicate, value, *match_params])

    cursor = self.conn.cursor()
    cursor.execute(query, params)

    return [
        Entity(
            id=row["id"],
            namespace=row["namespace"],
            concept=row["concept"],
            natural_key=row["natural_key"],
            created_at=datetime.fromisoformat(row["created_at"]),
            created_by=row["created_by"],
        )
        for row in cursor.fetchall()
    ]

entities_meeting_confidence

entities_meeting_confidence(
    namespace: str,
    concept: str,
    threshold: float,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]

IDs of entities in (namespace, concept) with >=1 assertion at or above threshold confidence, active at as_of_time (KI-036) or currently active if as_of_time is None. candidate_ids, if given, narrows the scan below (namespace, concept) (KI-037) via a single JSON-encoded bound parameter rather than one placeholder per id. include_flagged/include_history widen "active" the same way entities_where() does (KI-093) — see its docstring.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def entities_meeting_confidence(
    self,
    namespace: str,
    concept: str,
    threshold: float,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]:
    """IDs of entities in `(namespace, concept)` with >=1 assertion at or
    above `threshold` confidence, active at `as_of_time` (KI-036) or
    currently active if `as_of_time` is None. `candidate_ids`, if given,
    narrows the scan below `(namespace, concept)` (KI-037) via a single
    JSON-encoded bound parameter rather than one placeholder per id.
    `include_flagged`/`include_history` widen "active" the same way
    `entities_where()` does (KI-093) — see its docstring."""
    if candidate_ids is not None and not candidate_ids:
        return set()

    query = (
        "SELECT DISTINCT a.subject FROM assertion a"
        " JOIN entity e ON e.id = a.subject"
        " WHERE e.namespace = ? AND e.concept = ? AND a.confidence >= ?"
    )
    params: list[Any] = [namespace, concept, threshold]

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        # One of exactly two hardcoded literals, never caller-controlled
        # - no injection surface, and no #nosec needed (see
        # entities_where()'s identical comment for why bandit's B608
        # heuristic doesn't fire on this shape at all).
        # KI-097: flagged-at-t, not current status — same event-log
        # reconstruction entities_where()/assertions() already use;
        # see entities_where()'s comment for the full reasoning.
        flagged_clause = (
            ""
            if include_flagged
            else (
                " AND COALESCE("
                "(SELECT ae.action FROM assertion_event ae"
                " WHERE ae.assertion_id = a.id AND ae.at <= ?"
                " AND ae.action IN ('flagged', 'reactivated')"
                " ORDER BY ae.at DESC, ae.id DESC LIMIT 1),"
                " 'reactivated'"
                ") != 'flagged'"
            )
        )
        # ADR-0049 (KI-095): same retraction-aware exclusion
        # entities_where() has — see its comment for the full
        # reasoning. Two-hardcoded-literals shape, no #nosec needed.
        retracted_clause = (
            ""
            if include_history
            else (
                " AND (a.status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = a.id"
                " AND ae.action = 'retracted' AND ae.at > ?"
                "))"
            )
        )
        query += (
            " AND a.asserted_at <= ?"
            " AND (a.valid_from IS NULL OR a.valid_from <= ?)"
            " AND (a.valid_to IS NULL OR a.valid_to > ?)" + flagged_clause + retracted_clause
        )
        params.extend([t_iso, t_iso, t_iso])
        if not include_flagged:
            params.append(t_iso)
        if not include_history:
            params.append(t_iso)
    else:
        status_params = ["active"]
        if include_flagged:
            status_params.append("flagged")
        if include_history:
            status_params += ["superseded", "retracted"]
        query += f" AND a.status IN ({', '.join(['?'] * len(status_params))})"
        params.extend(status_params)

    if candidate_ids is not None:
        query += " AND e.id IN (SELECT value FROM json_each(?))"
        params.append(json.dumps(list(candidate_ids)))

    cursor = self.conn.cursor()
    cursor.execute(query, params)
    return {row["subject"] for row in cursor.fetchall()}

entities_meeting_trust

entities_meeting_trust(
    namespace: str,
    concept: str,
    min_trust: int,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]

IDs of entities in (namespace, concept) with >=1 assertion, active at as_of_time (KI-036) or currently active if as_of_time is None, whose effective trust_level >= min_trust (KI-047) — min(author.trust_level, acting_as.trust_level) when the assertion was made under delegation, matching govern/policy.py's identical formula for effective trust (by analogy with SPEC §8.4's capability rule), or just author.trust_level when it wasn't. A dangling acting_as (no resolvable delegate) falls back to author.trust_level via coalesce — see StorageBackend.entities_meeting_trust's docstring for why. candidate_ids narrows the scan the same way as entities_meeting_confidence (KI-037) — see its docstring. include_flagged/include_history widen "active" the same way entities_where() does (KI-093) — see its docstring.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def entities_meeting_trust(
    self,
    namespace: str,
    concept: str,
    min_trust: int,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]:
    """IDs of entities in `(namespace, concept)` with >=1 assertion,
    active at `as_of_time` (KI-036) or currently active if `as_of_time`
    is None, whose *effective* trust_level >= `min_trust` (KI-047) —
    `min(author.trust_level, acting_as.trust_level)` when the assertion
    was made under delegation, matching `govern/policy.py`'s identical
    formula for effective trust (by analogy with SPEC §8.4's capability
    rule), or just `author.trust_level` when it wasn't. A dangling
    `acting_as` (no resolvable delegate) falls back to `author.trust_level`
    via `coalesce` — see `StorageBackend.entities_meeting_trust`'s
    docstring for why. `candidate_ids` narrows the scan the same way as
    `entities_meeting_confidence` (KI-037) — see its docstring.
    `include_flagged`/`include_history` widen "active" the same way
    `entities_where()` does (KI-093) — see its docstring."""
    if candidate_ids is not None and not candidate_ids:
        return set()

    query = (
        "SELECT DISTINCT a.subject FROM assertion a"
        " JOIN entity e ON e.id = a.subject"
        " JOIN principal p ON p.id = a.author"
        " LEFT JOIN principal delegate ON delegate.id = a.acting_as"
        " WHERE e.namespace = ? AND e.concept = ?"
        " AND min(p.trust_level, coalesce(delegate.trust_level, p.trust_level)) >= ?"
    )
    params: list[Any] = [namespace, concept, min_trust]

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        # One of exactly two hardcoded literals, never caller-controlled
        # - no injection surface, and no #nosec needed (see
        # entities_where()'s identical comment for why bandit's B608
        # heuristic doesn't fire on this shape at all).
        # KI-097: flagged-at-t, not current status — same event-log
        # reconstruction entities_where()/assertions() already use;
        # see entities_where()'s comment for the full reasoning.
        flagged_clause = (
            ""
            if include_flagged
            else (
                " AND COALESCE("
                "(SELECT ae.action FROM assertion_event ae"
                " WHERE ae.assertion_id = a.id AND ae.at <= ?"
                " AND ae.action IN ('flagged', 'reactivated')"
                " ORDER BY ae.at DESC, ae.id DESC LIMIT 1),"
                " 'reactivated'"
                ") != 'flagged'"
            )
        )
        # ADR-0049 (KI-095): same retraction-aware exclusion
        # entities_where() has — see its comment for the full
        # reasoning. Two-hardcoded-literals shape, no #nosec needed.
        retracted_clause = (
            ""
            if include_history
            else (
                " AND (a.status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = a.id"
                " AND ae.action = 'retracted' AND ae.at > ?"
                "))"
            )
        )
        query += (
            " AND a.asserted_at <= ?"
            " AND (a.valid_from IS NULL OR a.valid_from <= ?)"
            " AND (a.valid_to IS NULL OR a.valid_to > ?)" + flagged_clause + retracted_clause
        )
        params.extend([t_iso, t_iso, t_iso])
        if not include_flagged:
            params.append(t_iso)
        if not include_history:
            params.append(t_iso)
    else:
        status_params = ["active"]
        if include_flagged:
            status_params.append("flagged")
        if include_history:
            status_params += ["superseded", "retracted"]
        query += f" AND a.status IN ({', '.join(['?'] * len(status_params))})"
        params.extend(status_params)

    if candidate_ids is not None:
        query += " AND e.id IN (SELECT value FROM json_each(?))"
        params.append(json.dumps(list(candidate_ids)))

    cursor = self.conn.cursor()
    cursor.execute(query, params)
    return {row["subject"] for row in cursor.fetchall()}

vector_upsert

vector_upsert(
    scope: str, id: str, vec: list[float]
) -> None

Insert or replace the embedding vector for (scope, id).

Parameters:

Name	Type	Description	Default
scope	str	Embedding scope. Must be one of VECTOR_SCOPES.	required
id	str	Entity or assertion ID the vector represents.	required
vec	list[float]	Embedding vector.	required

Raises:

Type	Description
ValidationError	scope is not in VECTOR_SCOPES, or vec's length does not match the scope's already-established dimension.
StorageError	If persistence fails.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def vector_upsert(self, scope: str, id: str, vec: list[float]) -> None:
    """Insert or replace the embedding vector for (scope, id).

    Args:
        scope: Embedding scope. Must be one of VECTOR_SCOPES.
        id: Entity or assertion ID the vector represents.
        vec: Embedding vector.

    Raises:
        ValidationError: scope is not in VECTOR_SCOPES, or vec's length
            does not match the scope's already-established dimension.
        StorageError: If persistence fails.
    """
    self._validate_scope(scope)
    self._ensure_vector_table(scope, len(vec))
    table = f"vector_{scope}"

    try:
        was_in_transaction = self._in_transaction
        if not was_in_transaction:
            self.begin()
        cursor = self.conn.cursor()
        existing = cursor.execute(
            "SELECT vec_rowid FROM vector_id_map WHERE scope = ? AND id = ?", (scope, id)
        ).fetchone()
        # table is built from `scope`, which _validate_scope() above
        # already checked against the closed VECTOR_SCOPES set — not
        # caller-controlled free text.
        cursor.execute(f"INSERT INTO {table}(embedding) VALUES (?)", (_pack_vector(vec),))  # nosec B608
        new_rowid = cursor.lastrowid
        if existing is not None:
            cursor.execute(f"DELETE FROM {table} WHERE rowid = ?", (existing["vec_rowid"],))  # nosec B608
        cursor.execute(
            "INSERT OR REPLACE INTO vector_id_map (scope, id, vec_rowid) VALUES (?, ?, ?)",
            (scope, id, new_rowid),
        )
        if not was_in_transaction:
            self.commit()
    except sqlite3.Error as e:
        if not was_in_transaction:
            self.rollback()
        raise StorageError(f"Failed to upsert vector (scope={scope}, id={id}): {e}") from e

vector_search

vector_search(
    scope: str, vec: list[float], k: int
) -> list[tuple[str, float]]

Return the k nearest ids to vec within scope, ascending distance.

Parameters:

Name	Type	Description	Default
scope	str	Embedding scope. Must be one of VECTOR_SCOPES.	required
vec	list[float]	Query vector.	required
k	int	Maximum number of results.	required

Returns:

Type	Description
list[tuple[str, float]]	(id, distance) tuples, nearest first. Empty list if the scope
list[tuple[str, float]]	has never been populated.

Raises:

Type	Description
ValidationError	scope is not in VECTOR_SCOPES, or vec's length does not match the scope's already-established dimension.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def vector_search(self, scope: str, vec: list[float], k: int) -> list[tuple[str, float]]:
    """Return the k nearest ids to vec within scope, ascending distance.

    Args:
        scope: Embedding scope. Must be one of VECTOR_SCOPES.
        vec: Query vector.
        k: Maximum number of results.

    Returns:
        (id, distance) tuples, nearest first. Empty list if the scope
        has never been populated.

    Raises:
        ValidationError: scope is not in VECTOR_SCOPES, or vec's length
            does not match the scope's already-established dimension.
    """
    self._validate_scope(scope)
    cursor = self.conn.cursor()
    row = cursor.execute("SELECT dim FROM vector_scope WHERE scope = ?", (scope,)).fetchone()
    if row is None:
        return []
    if row["dim"] != len(vec):
        raise ValidationError(
            f"Query vector for scope {scope!r} has dimension {len(vec)}, "
            f"but this scope is established at dimension {row['dim']}"
        )

    # table is built from `scope`, already validated above (same
    # reasoning as vector_upsert) — not caller-controlled free text.
    table = f"vector_{scope}"
    rows = cursor.execute(
        f"SELECT rowid, distance FROM {table} WHERE embedding MATCH ? ORDER BY distance LIMIT ?",  # nosec B608
        (_pack_vector(vec), k),
    ).fetchall()
    if not rows:
        return []

    # placeholders is just N repetitions of the literal "?" (N = len(rowids),
    # itself derived from `rows` above, never from external input) — an
    # IN-clause arity string, not a value; the actual rowids are still
    # bound as parameters below, not interpolated.
    rowids = [r["rowid"] for r in rows]
    placeholders = ",".join("?" for _ in rowids)
    id_rows = cursor.execute(
        f"SELECT vec_rowid, id FROM vector_id_map WHERE scope = ? AND vec_rowid IN ({placeholders})",  # nosec B608
        (scope, *rowids),
    ).fetchall()
    id_by_rowid = {r["vec_rowid"]: r["id"] for r in id_rows}

    return [(id_by_rowid[r["rowid"]], r["distance"]) for r in rows]

close

close() -> None

Close the database connection.

Source code in src/ontolith/store/sqlite/backend.py

@_synchronized
def close(self) -> None:
    """Close the database connection."""
    self.conn.close()

ontolith.store.duckdb

DuckDB storage backend.

Second pluggable storage adapter (M3, ADR-0016), proving the StorageBackend port abstraction against a structurally different embedded engine than SQLite.

DuckDBBackend

DuckDBBackend(
    path: str | Path, *, clock: Clock | None = None
)

DuckDB implementation of StorageBackend.

Schema follows SPEC §12.2, identical table shape to SQLiteBackend: - entity table with ULID primary key - assertion table with ULID primary key - Bitemporal columns (asserted_at, valid_from, valid_to) - Status tracking for append-only invariant

KI-066: unlike SQLiteBackend, assertion_event/proposal_event's append-only invariant (SPEC §17: "the audit trail MUST NOT be mutable") is enforced here only by this port's own surface exposing no update/delete method — a convention, not a store-level guarantee. DuckDB (verified against 1.5.4) has no CREATE TRIGGER support at all, so the three triggers per table (BEFORE UPDATE, BEFORE DELETE, BEFORE INSERT ... WHEN EXISTS(...)) SQLiteBackend installs have no DuckDB equivalent; code holding this backend's raw duckdb.DuckDBPyConnection can still UPDATE/DELETE/INSERT OR REPLACE either audit table directly. Not fixable without a different mechanism (e.g. a superuser-only schema plus a restricted role — not available in DuckDB's embedded, single-user connection model either).

Initialize DuckDB backend.

Parameters:

Name	Type	Description	Default
path	str | Path	Path to DuckDB database file (created if doesn't exist)	required
clock	Clock | None	Clock for timestamps (defaults to SystemClock)	None

Source code in src/ontolith/store/duckdb/backend.py

def __init__(self, path: str | Path, *, clock: Clock | None = None) -> None:
    """Initialize DuckDB backend.

    Args:
        path: Path to DuckDB database file (created if doesn't exist)
        clock: Clock for timestamps (defaults to SystemClock)
    """
    self.path = Path(path)
    self.path.parent.mkdir(parents=True, exist_ok=True)
    self.conn = duckdb.connect(str(self.path))
    self._clock: Clock = clock or SystemClock()
    # Serializes all access to self.conn across threads (KI-046): DuckDB's
    # own DB-API threadsafety level is 1 ("threads may share the module,
    # but not connections") — this single connection object is not safe
    # for concurrent use without external synchronization, the same
    # requirement SQLiteBackend's self._lock (KI-023) exists for. RLock
    # (not Lock): begin() holds it across multiple public-method calls
    # inside a transaction() block, each of which re-acquires it via the
    # @_synchronized decorator.
    self._lock = threading.RLock()
    # ADR-0052: refuses (SchemaError) an existing file below
    # migrations.CURRENT_FORMAT_VERSION — see SQLiteBackend's identical
    # check for the full reasoning; closes the connection before
    # propagating so a refused open doesn't leak a live handle. Catches
    # any failure here, not just SchemaError (round-1 review: a
    # corrupt/non-database file raises duckdb.Error instead).
    try:
        migrations.require_current_format(self.conn, path=self.path)
    except BaseException:
        self.conn.close()
        raise
    self._create_schema()

begin

begin() -> None

Begin an explicit transaction.

Acquires self._lock (KI-046) — held across every subsequent @_synchronized call until commit()/rollback() releases it, so no other thread's operation can interleave with this transaction.

Source code in src/ontolith/store/duckdb/backend.py

def begin(self) -> None:
    """Begin an explicit transaction.

    Acquires self._lock (KI-046) — held across every subsequent
    @_synchronized call until commit()/rollback() releases it, so no
    other thread's operation can interleave with this transaction.
    """
    self._lock.acquire()
    try:
        self.conn.execute("BEGIN TRANSACTION")
    except duckdb.Error as e:
        self._lock.release()
        raise StorageError(f"Failed to begin transaction: {e}") from e

commit

commit() -> None

Commit the current explicit transaction.

Releases self._lock only on success — mirrors SQLiteBackend's own asymmetric release (KI-023, ADR-0010's update): an unconditional release here would double-release the lock when transaction()'s except clause calls rollback() next after a failed commit(), which RLock.release() rejects with RuntimeError, masking the real StorageError.

Source code in src/ontolith/store/duckdb/backend.py

def commit(self) -> None:
    """Commit the current explicit transaction.

    Releases self._lock only on success — mirrors SQLiteBackend's own
    asymmetric release (KI-023, ADR-0010's update): an unconditional
    release here would double-release the lock when transaction()'s
    except clause calls rollback() next after a failed commit(), which
    RLock.release() rejects with RuntimeError, masking the real
    StorageError.
    """
    try:
        self.conn.execute("COMMIT")
    except duckdb.Error as e:
        raise StorageError(f"Failed to commit transaction: {e}") from e
    self._lock.release()

rollback

rollback() -> None

Rollback the current explicit transaction. Always releases self._lock.

Unlike commit(), this always resolves the transaction (successful or not) — it's the terminal cleanup path, including when called after a failed commit() (which deliberately did not release the lock itself, see commit()'s docstring).

Source code in src/ontolith/store/duckdb/backend.py

def rollback(self) -> None:
    """Rollback the current explicit transaction. Always releases self._lock.

    Unlike commit(), this always resolves the transaction (successful
    or not) — it's the terminal cleanup path, including when called
    after a failed commit() (which deliberately did not release the
    lock itself, see commit()'s docstring).
    """
    try:
        self.conn.execute("ROLLBACK")
    except duckdb.Error as e:
        raise StorageError(f"Failed to rollback transaction: {e}") from e
    finally:
        self._lock.release()

transaction

transaction() -> Iterator[None]

Context manager for atomic multi-write transactions.

Usage

with backend.transaction(): backend.put_entity(entity) backend.put_assertion(assertion)

Source code in src/ontolith/store/duckdb/backend.py

@contextmanager
def transaction(self) -> Iterator[None]:
    """Context manager for atomic multi-write transactions.

    Usage:
        with backend.transaction():
            backend.put_entity(entity)
            backend.put_assertion(assertion)
    """
    self.begin()
    try:
        yield
        self.commit()
    except Exception:
        self.rollback()
        raise

put_principal

put_principal(principal: Principal) -> None

Persist a principal.

Parameters:

Name	Type	Description	Default
principal	Principal	Principal to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_principal(self, principal: Principal) -> None:
    """Persist a principal.

    Args:
        principal: Principal to persist

    Raises:
        StorageError: If persistence fails
    """
    try:
        self.conn.execute(
            """
            INSERT INTO principal (id, kind, owner, auth_method, default_capability, trust_level, created_at, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                principal.id,
                principal.kind,
                principal.owner,
                principal.auth_method,
                principal.default_capability,
                principal.trust_level,
                principal.created_at.isoformat(),
                json.dumps(principal.metadata),
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Principal conflict (id={principal.id}): {e}") from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist principal (id={principal.id}): {e}") from e

get_principal

get_principal(principal_id: str) -> Principal | None

Retrieve a principal by ID.

Parameters:

Name	Type	Description	Default
principal_id	str	Principal ID to retrieve	required

Returns:

Type	Description
Principal | None	Principal if found, None otherwise

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_principal(self, principal_id: str) -> Principal | None:
    """Retrieve a principal by ID.

    Args:
        principal_id: Principal ID to retrieve

    Returns:
        Principal if found, None otherwise
    """
    cursor = self.conn.execute("SELECT * FROM principal WHERE id = ?", [principal_id])
    row = cursor.fetchone()
    if row is None:
        return None
    return self._row_to_principal(self._row_to_dict(cursor, row))

list_principals

list_principals() -> list[Principal]

List all principals (KI-022).

Returns:

Type	Description
list[Principal]	All principals, most recently created first

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def list_principals(self) -> list[Principal]:
    """List all principals (KI-022).

    Returns:
        All principals, most recently created first
    """
    # id DESC is a deterministic lexical tiebreak, not a recency proxy —
    # see SQLiteBackend's identical method for why that distinction
    # matters here (principal ids are user-supplied, unlike credential
    # ids' ULID-based get_credentials_for_principal tiebreak).
    cursor = self.conn.execute("SELECT * FROM principal ORDER BY created_at DESC, id DESC")
    return [self._row_to_principal(self._row_to_dict(cursor, row)) for row in cursor.fetchall()]

list_namespaces

list_namespaces() -> list[Namespace]

List all registered namespaces (SPEC §12.2, KI-022).

Returns:

Type	Description
list[Namespace]	All namespaces, most recently created first

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def list_namespaces(self) -> list[Namespace]:
    """List all registered namespaces (SPEC §12.2, KI-022).

    Returns:
        All namespaces, most recently created first
    """
    # id DESC tiebreak is purely lexical (namespace ids are slugs, not
    # ULIDs) — same convention as list_principals, not a recency proxy.
    cursor = self.conn.execute("SELECT * FROM namespace ORDER BY created_at DESC, id DESC")
    return [self._row_to_namespace(self._row_to_dict(cursor, row)) for row in cursor.fetchall()]

put_credential

put_credential(credential: PrincipalCredential) -> None

Persist a principal credential (hashed API-key token).

Parameters:

Name	Type	Description	Default
credential	PrincipalCredential	PrincipalCredential to persist (token_hash, never the raw token)	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_credential(self, credential: PrincipalCredential) -> None:
    """Persist a principal credential (hashed API-key token).

    Args:
        credential: PrincipalCredential to persist (token_hash, never the
            raw token)

    Raises:
        StorageError: If persistence fails
    """
    try:
        self.conn.execute(
            """
            INSERT INTO principal_credential
                (id, principal_id, token_hash, created_at, revoked_at, issued_by, revoked_by)
            VALUES (?, ?, ?, ?, ?, ?, ?)
            """,
            [
                credential.id,
                credential.principal_id,
                credential.token_hash,
                credential.created_at.isoformat(),
                credential.revoked_at.isoformat() if credential.revoked_at else None,
                credential.issued_by,
                credential.revoked_by,
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Credential conflict (id={credential.id}): {e}") from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist credential (id={credential.id}): {e}") from e

get_principal_by_token_hash

get_principal_by_token_hash(
    token_hash: str,
) -> Principal | None

Resolve a principal via a credential's token hash.

Only unrevoked credentials resolve. This is the sole read path used for MCP authentication — it never trusts a caller-supplied principal ID directly.

Parameters:

Name	Type	Description	Default
token_hash	str	SHA-256 hash of the raw bearer token	required

Returns:

Type	Description
Principal | None	Principal if the hash matches an active (unrevoked) credential,
Principal | None	None otherwise

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_principal_by_token_hash(self, token_hash: str) -> Principal | None:
    """Resolve a principal via a credential's token hash.

    Only unrevoked credentials resolve. This is the sole read path used
    for MCP authentication — it never trusts a caller-supplied principal
    ID directly.

    Args:
        token_hash: SHA-256 hash of the raw bearer token

    Returns:
        Principal if the hash matches an active (unrevoked) credential,
        None otherwise
    """
    cursor = self.conn.execute(
        """
        SELECT p.* FROM principal p
        JOIN principal_credential c ON c.principal_id = p.id
        WHERE c.token_hash = ? AND c.revoked_at IS NULL
        """,
        [token_hash],
    )
    row = cursor.fetchone()
    if row is None:
        return None
    return self._row_to_principal(self._row_to_dict(cursor, row))

get_credential

get_credential(
    credential_id: str,
) -> PrincipalCredential | None

Retrieve a credential by ID (never exposes the raw token or hash to callers).

Parameters:

Name	Type	Description	Default
credential_id	str	Credential ID to retrieve	required

Returns:

Type	Description
PrincipalCredential | None	PrincipalCredential if found, None otherwise

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_credential(self, credential_id: str) -> PrincipalCredential | None:
    """Retrieve a credential by ID (never exposes the raw token or hash to callers).

    Args:
        credential_id: Credential ID to retrieve

    Returns:
        PrincipalCredential if found, None otherwise
    """
    cursor = self.conn.execute(
        "SELECT * FROM principal_credential WHERE id = ?", [credential_id]
    )
    row = cursor.fetchone()
    if row is None:
        return None
    return self._row_to_credential(self._row_to_dict(cursor, row))

get_credentials_for_principal

get_credentials_for_principal(
    principal_id: str,
) -> list[PrincipalCredential]

List all credentials (active and revoked) issued to a principal.

Parameters:

Name	Type	Description	Default
principal_id	str	Principal to list credentials for	required

Returns:

Type	Description
list[PrincipalCredential]	Credentials for this principal, most recently issued first

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_credentials_for_principal(self, principal_id: str) -> list[PrincipalCredential]:
    """List all credentials (active and revoked) issued to a principal.

    Args:
        principal_id: Principal to list credentials for

    Returns:
        Credentials for this principal, most recently issued first
    """
    # id DESC tiebreaks two credentials issued at the same timestamp —
    # see SQLiteBackend's identical fix for the determinism rationale.
    cursor = self.conn.execute(
        "SELECT * FROM principal_credential WHERE principal_id = ? "
        "ORDER BY created_at DESC, id DESC",
        [principal_id],
    )
    return [
        self._row_to_credential(self._row_to_dict(cursor, row)) for row in cursor.fetchall()
    ]

revoke_credential

revoke_credential(
    credential_id: str,
    revoked_at: datetime,
    revoked_by: str,
) -> None

Mark a credential as revoked. Idempotent-safe: re-revoking an already-revoked credential is a true no-op — see SQLiteBackend's identical method for why (KI-060: doesn't launder attribution by overwriting revoked_by on a second call).

Parameters:

Name	Type	Description	Default
credential_id	str	Credential to revoke	required
revoked_at	datetime	Timestamp of revocation	required
revoked_by	str	Principal ID of the admin performing the revocation	required

Raises:

Type	Description
StorageError	If the credential is not found

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def revoke_credential(self, credential_id: str, revoked_at: datetime, revoked_by: str) -> None:
    """Mark a credential as revoked. Idempotent-safe: re-revoking an
    already-revoked credential is a true no-op — see SQLiteBackend's
    identical method for why (KI-060: doesn't launder attribution by
    overwriting `revoked_by` on a second call).

    Args:
        credential_id: Credential to revoke
        revoked_at: Timestamp of revocation
        revoked_by: Principal ID of the admin performing the revocation

    Raises:
        StorageError: If the credential is not found
    """
    cursor = self.conn.execute(
        "UPDATE principal_credential SET revoked_at = ?, revoked_by = ? "
        "WHERE id = ? AND revoked_at IS NULL RETURNING id",
        [revoked_at.isoformat(), revoked_by, credential_id],
    )
    if not cursor.fetchall():
        exists = self.conn.execute(
            "SELECT 1 FROM principal_credential WHERE id = ?", [credential_id]
        ).fetchone()
        if exists is None:
            raise StorageError(f"Credential not found: {credential_id}")

put_admin_event

put_admin_event(event: AdminEvent) -> None

Persist an append-only admin-action event (KI-060).

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_admin_event(self, event: AdminEvent) -> None:
    """Persist an append-only admin-action event (KI-060)."""
    try:
        self.conn.execute(
            """
            INSERT INTO admin_event (id, actor, action, target, "at", detail)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            [
                event.id,
                event.actor,
                event.action,
                event.target,
                event.at.isoformat(),
                event.detail,
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Admin event conflict (id={event.id}): {e}") from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist admin event (id={event.id}): {e}") from e

get_admin_events

get_admin_events(
    actor: str | None = None, target: str | None = None
) -> list[AdminEvent]

Retrieve admin events, optionally filtered by actor or target, oldest first.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_admin_events(
    self, actor: str | None = None, target: str | None = None
) -> list[AdminEvent]:
    """Retrieve admin events, optionally filtered by actor or target, oldest first."""
    query = "SELECT * FROM admin_event WHERE 1=1"
    params: list[str] = []
    if actor is not None:
        query += " AND actor = ?"
        params.append(actor)
    if target is not None:
        query += " AND target = ?"
        params.append(target)
    query += ' ORDER BY "at" ASC, id ASC'
    cursor = self.conn.execute(query, params)
    return [
        self._row_to_admin_event(self._row_to_dict(cursor, row)) for row in cursor.fetchall()
    ]

put_entity

put_entity(entity: Entity) -> None

Persist an entity.

Parameters:

Name	Type	Description	Default
entity	Entity	Entity to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_entity(self, entity: Entity) -> None:
    """Persist an entity.

    Args:
        entity: Entity to persist

    Raises:
        StorageError: If persistence fails
    """
    try:
        self.conn.execute(
            """
            INSERT INTO entity (id, namespace, concept, natural_key, created_at, created_by)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            [
                entity.id,
                entity.namespace,
                entity.concept,
                entity.natural_key,
                entity.created_at.isoformat(),
                entity.created_by,
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Entity conflict: {e}") from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist entity: {e}") from e

put_assertion

put_assertion(assertion: Assertion) -> None

Persist an assertion.

Parameters:

Name	Type	Description	Default
assertion	Assertion	Assertion to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_assertion(self, assertion: Assertion) -> None:
    """Persist an assertion.

    Args:
        assertion: Assertion to persist

    Raises:
        StorageError: If persistence fails
    """
    # Map unified value field to value_lit/value_ref based on kind
    value_lit = assertion.value if assertion.value_kind == "literal" else None
    value_ref = assertion.value if assertion.value_kind == "ref" else None

    try:
        self.conn.execute(
            """
            INSERT INTO assertion (
                id, namespace, subject, predicate,
                value_kind, value_type, value_lit, value_ref,
                author, acting_as, source, confidence, rationale, model,
                asserted_at, valid_from, valid_to,
                status, proposal_id, supersedes, metadata
            ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                assertion.id,
                assertion.namespace,
                assertion.subject,
                assertion.predicate,
                assertion.value_kind,
                assertion.value_type,
                value_lit,
                value_ref,
                assertion.author,
                assertion.acting_as,
                assertion.source,
                assertion.confidence,
                assertion.rationale,
                assertion.model,
                assertion.asserted_at.isoformat(),
                assertion.valid_from.isoformat() if assertion.valid_from else None,
                assertion.valid_to.isoformat() if assertion.valid_to else None,
                assertion.status,
                assertion.proposal_id,
                assertion.supersedes,
                json.dumps(assertion.metadata),
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(
            f"Assertion conflict (id={assertion.id}, subject={assertion.subject}): {e}"
        ) from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist assertion (id={assertion.id}): {e}") from e

get_entity

get_entity(entity_id: str) -> Entity | None

Retrieve an entity by ID.

Parameters:

Name	Type	Description	Default
entity_id	str	Entity ID to retrieve	required

Returns:

Type	Description
Entity | None	Entity if found, None otherwise

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_entity(self, entity_id: str) -> Entity | None:
    """Retrieve an entity by ID.

    Args:
        entity_id: Entity ID to retrieve

    Returns:
        Entity if found, None otherwise
    """
    cursor = self.conn.execute("SELECT * FROM entity WHERE id = ?", [entity_id])
    row = cursor.fetchone()
    if row is None:
        return None

    d = self._row_to_dict(cursor, row)
    return Entity(
        id=d["id"],
        namespace=d["namespace"],
        concept=d["concept"],
        natural_key=d["natural_key"],
        created_at=datetime.fromisoformat(d["created_at"]),
        created_by=d["created_by"],
    )

get_entity_by_natural_key

get_entity_by_natural_key(
    namespace: str, concept: str, natural_key: str
) -> Entity | None

Retrieve an entity by its unique (namespace, concept, natural_key) triple.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_entity_by_natural_key(
    self, namespace: str, concept: str, natural_key: str
) -> Entity | None:
    """Retrieve an entity by its unique (namespace, concept, natural_key) triple."""
    cursor = self.conn.execute(
        "SELECT * FROM entity WHERE namespace = ? AND concept = ? AND natural_key = ?",
        [namespace, concept, natural_key],
    )
    row = cursor.fetchone()
    if row is None:
        return None

    d = self._row_to_dict(cursor, row)
    return Entity(
        id=d["id"],
        namespace=d["namespace"],
        concept=d["concept"],
        natural_key=d["natural_key"],
        created_at=datetime.fromisoformat(d["created_at"]),
        created_by=d["created_by"],
    )

assertions

assertions(
    subject: str | None = None,
    predicate: str | None = None,
    status: str | None = "active",
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Assertion]

Query assertions with optional filters.

Parameters:

Name	Type	Description	Default
subject	str | None	Filter by subject entity ID	None
predicate	str | None	Filter by predicate	None
status	str | None	Filter by current status (ignored when as_of_time is set). Defaults to "active"; pass status=None for every status.	'active'
as_of_time	datetime | None	If set, applies bitemporal filter: asserted_at <= t AND valid_from <= t AND (valid_to IS NULL OR valid_to > t)	None
include_flagged	bool	When as_of_time is set, whether to include 'flagged' assertions (excluded by default — a flagged assertion is disputed, not confirmed-valid; pass True for explicit audit/history views); ignored when as_of_time is None.	False
include_history	bool	When as_of_time is set, whether to opt back into seeing a 'retracted' assertion once its own retraction event's timestamp is <= as_of_time (excluded by default, ADR-0049/KI-095) — mirrors include_flagged's shape (KI-098); ignored when as_of_time is None.	False

Returns:

Type	Description
list[Assertion]	List of matching assertions

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def assertions(
    self,
    subject: str | None = None,
    predicate: str | None = None,
    status: str | None = "active",
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Assertion]:
    """Query assertions with optional filters.

    Args:
        subject: Filter by subject entity ID
        predicate: Filter by predicate
        status: Filter by current status (ignored when as_of_time is
            set). Defaults to "active"; pass status=None for every
            status.
        as_of_time: If set, applies bitemporal filter:
            asserted_at <= t AND valid_from <= t AND (valid_to IS NULL OR valid_to > t)
        include_flagged: When as_of_time is set, whether to include
            'flagged' assertions (excluded by default — a flagged
            assertion is disputed, not confirmed-valid; pass True for
            explicit audit/history views); ignored when as_of_time is
            None.
        include_history: When as_of_time is set, whether to opt back
            into seeing a 'retracted' assertion once its own retraction
            event's timestamp is <= as_of_time (excluded by default,
            ADR-0049/KI-095) — mirrors include_flagged's shape (KI-098);
            ignored when as_of_time is None.

    Returns:
        List of matching assertions
    """
    query = "SELECT * FROM assertion WHERE 1=1"
    params: list[str] = []

    if subject is not None:
        query += " AND subject = ?"
        params.append(subject)

    if predicate is not None:
        query += " AND predicate = ?"
        params.append(predicate)

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        query += " AND asserted_at <= ?"
        params.append(t_iso)
        query += " AND (valid_from IS NULL OR valid_from <= ?)"
        params.append(t_iso)
        query += " AND (valid_to IS NULL OR valid_to > ?)"
        params.append(t_iso)
        if not include_flagged:
            # Flagged-at-t, not current status: a static conflict flags an
            # assertion permanently (no valid_to change), so using current
            # status here would hide it from as_of() queries for times
            # before the dispute existed. Reconstruct from the event log
            # instead — every flagged transition (including an assertion
            # born already-flagged) has a 'flagged' event, see
            # Ontology._apply_with_conflict_routing. "at" is quoted -
            # reserved word in DuckDB. Tiebreak on ae.id: two events can
            # share the same `at` under a clock that hasn't advanced
            # (e.g. flag-then-resolve in the same tick), and `at` alone
            # would make "last recorded wins" nondeterministic. Under
            # SequentialIdProvider/FixedIdProvider (used in tests) id
            # order matches recording order exactly; under the production
            # UlidProvider, id is monotonic across milliseconds but not
            # guaranteed within one, so same-`at` AND same-millisecond
            # ties are a residual (low-probability, not exploitable)
            # nondeterminism.
            query += """ AND COALESCE(
                (SELECT ae.action FROM assertion_event ae
                 WHERE ae.assertion_id = assertion.id AND ae."at" <= ?
                   AND ae.action IN ('flagged', 'reactivated')
                 ORDER BY ae."at" DESC, ae.id DESC LIMIT 1),
                'reactivated'
            ) != 'flagged'"""
            params.append(t_iso)
        # ADR-0049 (KI-095): same retraction-aware exclusion the
        # entities_where()/entities_meeting_confidence()/
        # entities_meeting_trust() family has — see entities_where()'s
        # comment for the full reasoning (KI-051 guarantees at most one
        # 'retracted' event per assertion, so unlike flagged/reactivated
        # above this needs no ORDER BY tiebreak: retraction is a
        # one-way terminal transition, never followed by another event).
        # include_history opts back out of this check, the same
        # mechanism the QueryBuilder-facing trio already uses (KI-098 —
        # this method had no such opt-out when ADR-0049 first shipped).
        # "at" is quoted - reserved word in DuckDB (see the flagged
        # reconstruction's identical note).
        if not include_history:
            query += (
                " AND (status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = assertion.id"
                " AND ae.action = 'retracted' AND ae.\"at\" > ?"
                "))"
            )
            params.append(t_iso)
    elif status is not None:
        query += " AND status = ?"
        params.append(status)

    cursor = self.conn.execute(query, params)
    rows = cursor.fetchall()
    return [self._row_to_assertion(self._row_to_dict(cursor, row)) for row in rows]

get_assertion

get_assertion(assertion_id: str) -> Assertion | None

Retrieve a single assertion by ID, regardless of status.

Parameters:

Name	Type	Description	Default
assertion_id	str	Assertion ID to retrieve	required

Returns:

Type	Description
Assertion | None	Assertion if found, None otherwise

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_assertion(self, assertion_id: str) -> Assertion | None:
    """Retrieve a single assertion by ID, regardless of status.

    Args:
        assertion_id: Assertion ID to retrieve

    Returns:
        Assertion if found, None otherwise
    """
    cursor = self.conn.execute("SELECT * FROM assertion WHERE id = ?", [assertion_id])
    row = cursor.fetchone()
    return self._row_to_assertion(self._row_to_dict(cursor, row)) if row else None

set_assertion_status

set_assertion_status(
    assertion_id: str,
    status: str,
    valid_to: str | None = None,
) -> None

Update assertion status and optionally close validity window.

This is the ONLY allowed mutation on assertions (append-only invariant).

Parameters:

Name	Type	Description	Default
assertion_id	str	Assertion ID to update	required
status	str	New status (superseded, retracted, flagged)	required
valid_to	str | None	Optional validity end time (ISO format)	None

Raises:

Type	Description
StorageError	If update fails or assertion not found

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def set_assertion_status(
    self,
    assertion_id: str,
    status: str,
    valid_to: str | None = None,
) -> None:
    """Update assertion status and optionally close validity window.

    This is the ONLY allowed mutation on assertions (append-only invariant).

    Args:
        assertion_id: Assertion ID to update
        status: New status (superseded, retracted, flagged)
        valid_to: Optional validity end time (ISO format)

    Raises:
        StorageError: If update fails or assertion not found
    """
    try:
        # Existence check + plain UPDATE, not `UPDATE ... RETURNING`: the
        # assertion table is FK-referenced by assertion_event.assertion_id,
        # and DuckDB's RETURNING raises a false-positive constraint
        # violation when updating a row that's the target of an incoming
        # FK from another table (confirmed empirically, duckdb==1.5.4;
        # plain UPDATE on the same row works correctly).
        if (
            self.conn.execute("SELECT 1 FROM assertion WHERE id = ?", [assertion_id]).fetchone()
            is None
        ):
            raise StorageError(f"Assertion not found: {assertion_id}")

        if valid_to is not None:
            self.conn.execute(
                "UPDATE assertion SET status = ?, valid_to = ? WHERE id = ?",
                [status, valid_to, assertion_id],
            )
        else:
            self.conn.execute(
                "UPDATE assertion SET status = ? WHERE id = ?",
                [status, assertion_id],
            )
    except duckdb.Error as e:
        raise StorageError(f"Failed to update assertion status (id={assertion_id}): {e}") from e

put_schema

put_schema(schema: SchemaIR) -> None

Persist a schema version.

Also registers schema.namespace in the namespace registry if not already present (KI-022) — a namespace that only ever has a schema applied, never an entity, is still discoverable via list_namespaces().

Parameters:

Name	Type	Description	Default
schema	SchemaIR	Schema to persist	required

Raises:

Type	Description
StorageError	If persistence fails

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_schema(self, schema: SchemaIR) -> None:
    """Persist a schema version.

    Also registers ``schema.namespace`` in the namespace registry if
    not already present (KI-022) — a namespace that only ever has a
    schema applied, never an entity, is still discoverable via
    `list_namespaces()`.

    Args:
        schema: Schema to persist

    Raises:
        StorageError: If persistence fails
    """
    try:
        self._ensure_namespace_registered(schema.namespace)
        self.conn.execute(
            """
            INSERT INTO schema_version (namespace, version, definition, applied_at)
            VALUES (?, ?, ?, ?)
            """,
            [
                schema.namespace,
                schema.version,
                json.dumps(schema.to_json()),
                self._clock.now().isoformat(),
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(
            f"Schema conflict (namespace={schema.namespace}, version={schema.version}): {e}"
        ) from e
    except duckdb.Error as e:
        raise StorageError(
            f"Failed to persist schema (namespace={schema.namespace}): {e}"
        ) from e

get_schema

get_schema(
    namespace: str, version: int | None = None
) -> SchemaIR | None

Retrieve a schema version.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to query	required
version	int | None	Specific version, or None for latest	None

Returns:

Type	Description
SchemaIR | None	Schema if found, None otherwise

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_schema(self, namespace: str, version: int | None = None) -> SchemaIR | None:
    """Retrieve a schema version.

    Args:
        namespace: Namespace to query
        version: Specific version, or None for latest

    Returns:
        Schema if found, None otherwise
    """
    if version is None:
        # Get latest version
        cursor = self.conn.execute(
            """
            SELECT definition FROM schema_version
            WHERE namespace = ?
            ORDER BY version DESC
            LIMIT 1
            """,
            [namespace],
        )
    else:
        # Get specific version
        cursor = self.conn.execute(
            """
            SELECT definition FROM schema_version
            WHERE namespace = ? AND version = ?
            """,
            [namespace, version],
        )

    row = cursor.fetchone()
    if row is None:
        return None

    definition = json.loads(row[0])
    return SchemaIR.from_json(definition)

get_schema_at

get_schema_at(
    namespace: str, at: datetime
) -> SchemaIR | None

Retrieve the schema version effective at a point in time (KI-019).

Orders by applied_at (the actual "effective at" moment), with version as a tiebreak for same-timestamp rows under a coarse or injected Clock — not by version alone, so this stays correct even if a future write path ever persisted schema rows out of temporal order relative to their version numbers.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_schema_at(self, namespace: str, at: datetime) -> SchemaIR | None:
    """Retrieve the schema version effective at a point in time (KI-019).

    Orders by applied_at (the actual "effective at" moment), with
    version as a tiebreak for same-timestamp rows under a coarse or
    injected Clock — not by version alone, so this stays correct even
    if a future write path ever persisted schema rows out of temporal
    order relative to their version numbers.
    """
    cursor = self.conn.execute(
        """
        SELECT definition FROM schema_version
        WHERE namespace = ? AND applied_at <= ?
        ORDER BY applied_at DESC, version DESC
        LIMIT 1
        """,
        [namespace, at.isoformat()],
    )
    row = cursor.fetchone()
    if row is None:
        return None

    definition = json.loads(row[0])
    return SchemaIR.from_json(definition)

entities

entities(
    namespace: str | None = None,
    concept: str | None = None,
    as_of_time: datetime | None = None,
) -> list[Entity]

Query entities with optional filters.

Parameters:

Name	Type	Description	Default
namespace	str | None	Filter by namespace	None
concept	str | None	Filter by concept	None
as_of_time	datetime | None	If set, exclude entities created after this time	None

Returns:

Type	Description
list[Entity]	List of matching entities

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def entities(
    self,
    namespace: str | None = None,
    concept: str | None = None,
    as_of_time: datetime | None = None,
) -> list[Entity]:
    """Query entities with optional filters.

    Args:
        namespace: Filter by namespace
        concept: Filter by concept
        as_of_time: If set, exclude entities created after this time

    Returns:
        List of matching entities
    """
    query = "SELECT * FROM entity WHERE 1=1"
    params: list[str] = []

    if namespace is not None:
        query += " AND namespace = ?"
        params.append(namespace)

    if concept is not None:
        query += " AND concept = ?"
        params.append(concept)

    if as_of_time is not None:
        query += " AND created_at <= ?"
        params.append(as_of_time.isoformat())

    cursor = self.conn.execute(query, params)
    rows = cursor.fetchall()

    results = []
    for row in rows:
        d = self._row_to_dict(cursor, row)
        results.append(
            Entity(
                id=d["id"],
                namespace=d["namespace"],
                concept=d["concept"],
                natural_key=d["natural_key"],
                created_at=datetime.fromisoformat(d["created_at"]),
                created_by=d["created_by"],
            )
        )

    return results

put_proposal

put_proposal(proposal: Proposal) -> None

Persist a proposal.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_proposal(self, proposal: Proposal) -> None:
    """Persist a proposal."""
    try:
        self.conn.execute(
            """
            INSERT INTO proposal (id, namespace, author, acting_as, state,
                created_at, decided_at, policy_reason, reviewers, payload, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                proposal.id,
                proposal.namespace,
                proposal.author,
                proposal.acting_as,
                proposal.state,
                proposal.created_at.isoformat(),
                proposal.decided_at.isoformat() if proposal.decided_at else None,
                proposal.policy_reason,
                json.dumps(proposal.reviewers),
                json.dumps(proposal.payload),
                json.dumps(proposal.metadata),
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Proposal conflict (id={proposal.id}): {e}") from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist proposal (id={proposal.id}): {e}") from e

get_proposal

get_proposal(proposal_id: str) -> Proposal | None

Retrieve a proposal by ID.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_proposal(self, proposal_id: str) -> Proposal | None:
    """Retrieve a proposal by ID."""
    cursor = self.conn.execute("SELECT * FROM proposal WHERE id = ?", [proposal_id])
    row = cursor.fetchone()
    if row is None:
        return None
    return self._row_to_proposal(self._row_to_dict(cursor, row))

proposals

proposals(state: str | None = None) -> list[Proposal]

Query proposals, optionally filtered by state (SPEC §14.1).

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def proposals(self, state: str | None = None) -> list[Proposal]:
    """Query proposals, optionally filtered by state (SPEC §14.1)."""
    if state is not None:
        cursor = self.conn.execute(
            "SELECT * FROM proposal WHERE state = ? ORDER BY created_at DESC, id DESC", [state]
        )
    else:
        cursor = self.conn.execute("SELECT * FROM proposal ORDER BY created_at DESC, id DESC")
    rows = cursor.fetchall()
    return [self._row_to_proposal(self._row_to_dict(cursor, row)) for row in rows]

update_proposal_state

update_proposal_state(
    proposal_id: str,
    state: str,
    decided_at: str | None = None,
    policy_reason: str | None = None,
) -> None

Update proposal state after policy decision.

policy_reason=None leaves the stored value unchanged (COALESCE), it does not clear it — see the port docstring for why.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def update_proposal_state(
    self,
    proposal_id: str,
    state: str,
    decided_at: str | None = None,
    policy_reason: str | None = None,
) -> None:
    """Update proposal state after policy decision.

    policy_reason=None leaves the stored value unchanged (COALESCE), it
    does not clear it — see the port docstring for why.
    """
    try:
        # Existence check + plain UPDATE, not `UPDATE ... RETURNING`: the
        # proposal table is FK-referenced by proposal_event.proposal_id —
        # see set_assertion_status for why RETURNING is unsafe here.
        if (
            self.conn.execute("SELECT 1 FROM proposal WHERE id = ?", [proposal_id]).fetchone()
            is None
        ):
            raise StorageError(f"Proposal not found: {proposal_id}")

        self.conn.execute(
            "UPDATE proposal SET state = ?, decided_at = ?, "
            "policy_reason = COALESCE(?, policy_reason) WHERE id = ?",
            [state, decided_at, policy_reason, proposal_id],
        )
    except duckdb.Error as e:
        raise StorageError(f"Failed to update proposal (id={proposal_id}): {e}") from e

update_proposal_reviewers

update_proposal_reviewers(
    proposal_id: str, reviewers: list[str]
) -> None

Replace a proposal's assigned reviewers (SPEC §9.4's assign action).

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def update_proposal_reviewers(self, proposal_id: str, reviewers: list[str]) -> None:
    """Replace a proposal's assigned reviewers (SPEC §9.4's `assign` action)."""
    try:
        # Existence check + plain UPDATE, not `UPDATE ... RETURNING` —
        # same FK-reference reason as update_proposal_state above.
        if (
            self.conn.execute("SELECT 1 FROM proposal WHERE id = ?", [proposal_id]).fetchone()
            is None
        ):
            raise StorageError(f"Proposal not found: {proposal_id}")

        self.conn.execute(
            "UPDATE proposal SET reviewers = ? WHERE id = ?",
            [json.dumps(reviewers), proposal_id],
        )
    except duckdb.Error as e:
        raise StorageError(
            f"Failed to update proposal reviewers (id={proposal_id}): {e}"
        ) from e

put_proposal_event

put_proposal_event(event: ProposalEvent) -> None

Persist a structured review-action event (SPEC §9.4).

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_proposal_event(self, event: ProposalEvent) -> None:
    """Persist a structured review-action event (SPEC §9.4)."""
    try:
        self.conn.execute(
            """
            INSERT INTO proposal_event (id, proposal_id, actor, type, detail, "at")
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            [
                event.id,
                event.proposal_id,
                event.actor,
                event.type,
                event.detail,
                event.at.isoformat(),
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Proposal event conflict (id={event.id}): {e}") from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist proposal event (id={event.id}): {e}") from e

get_proposal_events

get_proposal_events(
    proposal_id: str,
) -> list[ProposalEvent]

Retrieve all review events for a proposal, oldest first.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_proposal_events(self, proposal_id: str) -> list[ProposalEvent]:
    """Retrieve all review events for a proposal, oldest first."""
    cursor = self.conn.execute(
        'SELECT * FROM proposal_event WHERE proposal_id = ? ORDER BY "at" ASC',
        [proposal_id],
    )
    rows = cursor.fetchall()
    return [
        ProposalEvent(
            id=d["id"],
            proposal_id=d["proposal_id"],
            actor=d["actor"],
            type=d["type"],
            detail=d["detail"],
            at=datetime.fromisoformat(d["at"]),
        )
        for d in (self._row_to_dict(cursor, row) for row in rows)
    ]

put_assertion_event

put_assertion_event(event: AssertionEvent) -> None

Persist an append-only assertion status-mutation event.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_assertion_event(self, event: AssertionEvent) -> None:
    """Persist an append-only assertion status-mutation event."""
    try:
        self.conn.execute(
            """
            INSERT INTO assertion_event (id, assertion_id, actor, action, "at", successor_id)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            [
                event.id,
                event.assertion_id,
                event.actor,
                event.action,
                event.at.isoformat(),
                event.successor_id,
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Assertion event conflict (id={event.id}): {e}") from e
    except duckdb.Error as e:
        raise StorageError(f"Failed to persist assertion event (id={event.id}): {e}") from e

get_assertion_events

get_assertion_events(
    assertion_id: str,
) -> list[AssertionEvent]

Retrieve all status-mutation events for an assertion, oldest first.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_assertion_events(self, assertion_id: str) -> list[AssertionEvent]:
    """Retrieve all status-mutation events for an assertion, oldest first."""
    cursor = self.conn.execute(
        'SELECT * FROM assertion_event WHERE assertion_id = ? ORDER BY "at" ASC',
        [assertion_id],
    )
    rows = cursor.fetchall()
    return [
        self._row_to_assertion_event(d)
        for d in (self._row_to_dict(cursor, row) for row in rows)
    ]

get_assertion_events_by_successor

get_assertion_events_by_successor(
    successor_id: str,
) -> list[AssertionEvent]

Retrieve all 'superseded' events caused by a given successor assertion.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_assertion_events_by_successor(self, successor_id: str) -> list[AssertionEvent]:
    """Retrieve all 'superseded' events caused by a given successor assertion."""
    cursor = self.conn.execute(
        'SELECT * FROM assertion_event WHERE successor_id = ? ORDER BY "at" ASC, id ASC',
        [successor_id],
    )
    rows = cursor.fetchall()
    return [
        self._row_to_assertion_event(d)
        for d in (self._row_to_dict(cursor, row) for row in rows)
    ]

put_contradiction

put_contradiction(contradiction: Contradiction) -> None

Persist a new contradiction.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def put_contradiction(self, contradiction: Contradiction) -> None:
    """Persist a new contradiction."""
    try:
        self.conn.execute(
            """
            INSERT INTO contradiction (id, namespace, subject, predicate, state,
                member_ids, created_at, raised_by, resolved_by, resolved_at, metadata)
            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
            """,
            [
                contradiction.id,
                contradiction.namespace,
                contradiction.subject,
                contradiction.predicate,
                contradiction.state,
                json.dumps(contradiction.member_ids),
                contradiction.created_at.isoformat(),
                contradiction.raised_by,
                contradiction.resolved_by,
                contradiction.resolved_at.isoformat() if contradiction.resolved_at else None,
                json.dumps(contradiction.metadata),
            ],
        )
    except duckdb.IntegrityError as e:
        raise StorageError(f"Contradiction conflict (id={contradiction.id}): {e}") from e
    except duckdb.Error as e:
        raise StorageError(
            f"Failed to persist contradiction (id={contradiction.id}): {e}"
        ) from e

get_open_contradiction

get_open_contradiction(
    namespace: str, subject: str, predicate: str
) -> Contradiction | None

Return the open contradiction for (namespace, subject, predicate), if any.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_open_contradiction(
    self, namespace: str, subject: str, predicate: str
) -> Contradiction | None:
    """Return the open contradiction for (namespace, subject, predicate), if any."""
    cursor = self.conn.execute(
        """
        SELECT * FROM contradiction
        WHERE namespace = ? AND subject = ? AND predicate = ? AND state = 'open'
        LIMIT 1
        """,
        [namespace, subject, predicate],
    )
    row = cursor.fetchone()
    return self._row_to_contradiction(self._row_to_dict(cursor, row)) if row else None

update_contradiction_members

update_contradiction_members(
    contradiction_id: str,
    member_ids: list[str],
    metadata: dict[str, Any] | None = None,
) -> None

Add member IDs to an existing open contradiction (KI-071: optionally replace metadata too, in the same UPDATE).

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def update_contradiction_members(
    self,
    contradiction_id: str,
    member_ids: list[str],
    metadata: dict[str, Any] | None = None,
) -> None:
    """Add member IDs to an existing open contradiction (KI-071:
    optionally replace metadata too, in the same UPDATE)."""
    try:
        if metadata is not None:
            cursor = self.conn.execute(
                "UPDATE contradiction SET member_ids = ?, metadata = ? WHERE id = ? "
                "RETURNING id",
                [json.dumps(member_ids), json.dumps(metadata), contradiction_id],
            )
        else:
            cursor = self.conn.execute(
                "UPDATE contradiction SET member_ids = ? WHERE id = ? RETURNING id",
                [json.dumps(member_ids), contradiction_id],
            )
        if not cursor.fetchall():
            raise StorageError(f"Contradiction not found: {contradiction_id}")
    except duckdb.Error as e:
        raise StorageError(
            f"Failed to update contradiction (id={contradiction_id}): {e}"
        ) from e

get_contradiction

get_contradiction(
    contradiction_id: str,
) -> Contradiction | None

Retrieve a contradiction by ID, regardless of state.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def get_contradiction(self, contradiction_id: str) -> Contradiction | None:
    """Retrieve a contradiction by ID, regardless of state."""
    cursor = self.conn.execute("SELECT * FROM contradiction WHERE id = ?", [contradiction_id])
    row = cursor.fetchone()
    return self._row_to_contradiction(self._row_to_dict(cursor, row)) if row else None

contradictions

contradictions(
    state: str | None = None,
) -> list[Contradiction]

Query contradictions, optionally filtered by state (SPEC §14.1).

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def contradictions(self, state: str | None = None) -> list[Contradiction]:
    """Query contradictions, optionally filtered by state (SPEC §14.1)."""
    if state is not None:
        cursor = self.conn.execute(
            "SELECT * FROM contradiction WHERE state = ? ORDER BY created_at DESC, id DESC",
            [state],
        )
    else:
        cursor = self.conn.execute(
            "SELECT * FROM contradiction ORDER BY created_at DESC, id DESC"
        )
    rows = cursor.fetchall()
    return [self._row_to_contradiction(self._row_to_dict(cursor, row)) for row in rows]

resolve_contradiction

resolve_contradiction(
    contradiction_id: str,
    resolved_by: str,
    resolved_at: datetime,
) -> None

Mark a contradiction as resolved (SPEC §10.3).

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def resolve_contradiction(
    self,
    contradiction_id: str,
    resolved_by: str,
    resolved_at: datetime,
) -> None:
    """Mark a contradiction as resolved (SPEC §10.3)."""
    try:
        cursor = self.conn.execute(
            """
            UPDATE contradiction
            SET state = 'resolved', resolved_by = ?, resolved_at = ?
            WHERE id = ?
            RETURNING id
            """,
            [resolved_by, resolved_at.isoformat(), contradiction_id],
        )
        if not cursor.fetchall():
            raise StorageError(f"Contradiction not found: {contradiction_id}")
    except duckdb.Error as e:
        raise StorageError(
            f"Failed to resolve contradiction (id={contradiction_id}): {e}"
        ) from e

vector_upsert

vector_upsert(
    scope: str, id: str, vec: list[float]
) -> None

Insert or replace the embedding vector for (scope, id).

Parameters:

Name	Type	Description	Default
scope	str	Embedding scope. Must be one of VECTOR_SCOPES.	required
id	str	Entity or assertion ID the vector represents.	required
vec	list[float]	Embedding vector.	required

Raises:

Type	Description
ValidationError	scope is not in VECTOR_SCOPES, or vec's length does not match the scope's already-established dimension.
StorageError	If persistence fails.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def vector_upsert(self, scope: str, id: str, vec: list[float]) -> None:
    """Insert or replace the embedding vector for (scope, id).

    Args:
        scope: Embedding scope. Must be one of VECTOR_SCOPES.
        id: Entity or assertion ID the vector represents.
        vec: Embedding vector.

    Raises:
        ValidationError: scope is not in VECTOR_SCOPES, or vec's length
            does not match the scope's already-established dimension.
        StorageError: If persistence fails.
    """
    self._validate_scope(scope)
    self._ensure_vector_table(scope, len(vec))
    try:
        # vector_{scope} is built from `scope`, already validated above
        # against the closed VECTOR_SCOPES set — not caller-controlled
        # free text.
        self.conn.execute(f"INSERT OR REPLACE INTO vector_{scope} VALUES (?, ?)", [id, vec])
    except duckdb.Error as e:
        raise StorageError(f"Failed to upsert vector (scope={scope}, id={id}): {e}") from e

vector_search

vector_search(
    scope: str, vec: list[float], k: int
) -> list[tuple[str, float]]

Return the k nearest ids to vec within scope, ascending distance.

Parameters:

Name	Type	Description	Default
scope	str	Embedding scope. Must be one of VECTOR_SCOPES.	required
vec	list[float]	Query vector.	required
k	int	Maximum number of results.	required

Returns:

Type	Description
list[tuple[str, float]]	(id, distance) tuples, nearest first. Empty list if the scope
list[tuple[str, float]]	has never been populated.

Raises:

Type	Description
ValidationError	scope is not in VECTOR_SCOPES, or vec's length does not match the scope's already-established dimension.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def vector_search(self, scope: str, vec: list[float], k: int) -> list[tuple[str, float]]:
    """Return the k nearest ids to vec within scope, ascending distance.

    Args:
        scope: Embedding scope. Must be one of VECTOR_SCOPES.
        vec: Query vector.
        k: Maximum number of results.

    Returns:
        (id, distance) tuples, nearest first. Empty list if the scope
        has never been populated.

    Raises:
        ValidationError: scope is not in VECTOR_SCOPES, or vec's length
            does not match the scope's already-established dimension.
    """
    self._validate_scope(scope)
    row = self.conn.execute("SELECT dim FROM vector_scope WHERE scope = ?", [scope]).fetchone()
    if row is None:
        return []
    if row[0] != len(vec):
        raise ValidationError(
            f"Query vector for scope {scope!r} has dimension {len(vec)}, "
            f"but this scope is established at dimension {row[0]}"
        )

    # vector_{scope} is built from `scope`, already validated above
    # (same reasoning as vector_upsert) — not caller-controlled free text.
    rows = self.conn.execute(
        f"SELECT id, list_distance(embedding, ?) AS distance FROM vector_{scope} "  # nosec B608
        "ORDER BY distance LIMIT ?",
        [vec, k],
    ).fetchall()
    return [(r[0], r[1]) for r in rows]

entities_where

entities_where(
    namespace: str,
    concept: str,
    predicate_filters: list[tuple[str, str, Any]],
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Entity]

Query entities matching all predicate filters in one SQL query.

Uses correlated subqueries so each filter hits the idx_assertion_pred_value/idx_assertion_pred_ref indexes instead of doing one round-trip per entity. "eq" matches either a literal property (value_lit) or a relation's target entity id (value_ref) — KI-030: relation filters like employer="org-123" are equality checks against value_ref, not traversal into the target entity's own properties. The two are checked via a UNION ALL of two single-column point lookups rather than one value_lit = ? OR value_ref = ? predicate, matching the SQLite backend (whose planner doesn't reliably pick a seekable plan for the OR form). "contains"/"gt"/"lt"/"gte"/"lte" (KI-039) check value_lit only — see this port method's own docstring for why relations don't get a UNION ALL branch for those. Numeric range comparisons cast to DOUBLE, not REAL — this module's own header comment notes DuckDB's REAL is 4-byte single precision, unlike SQLite's always-8-byte REAL, which would silently round values.

Parameters:

Name	Type	Description	Default
namespace	str	Namespace to query	required
concept	str	Concept to filter by	required
predicate_filters	list[tuple[str, str, Any]]	List of (full_predicate, operator, value) triples (AND semantics) — see the port method's docstring for the operator set	required
as_of_time	datetime | None	If set, applies bitemporal filter on assertions and entity creation	None
include_flagged	bool	Also match 'flagged' assertions — honored on both the current-state and as_of_time paths (KI-081; excluded by default on both, see assertions()). On the as_of_time path this is point-in-time, not current status (KI-097): reconstructed from the assertion_event log the same way assertions() already does, so a query pinned to a time when an assertion was disputed correctly excludes it even after the dispute has since been resolved — and, the other direction, a time strictly before any dispute existed still includes an otherwise-undisputed value, even though the same assertion is flagged now.	False
include_history	bool	Also match 'superseded'/'retracted' assertions (KI-081). On the current-state path this widens the status set beyond 'active'. On the as_of_time path, 'superseded' is unaffected either way (its window already never restricts to 'active') — but 'retracted' does something under this flag now (ADR-0049, KI-095): as_of_time excludes a retracted assertion once its own retraction event's "at" is <= as_of_time (a stale, un-narrowed window otherwise keeps matching indefinitely); include_history opts back out of that exclusion.	False

Returns:

Type	Description
list[Entity]	List of entities where all filters match at the given time

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def entities_where(
    self,
    namespace: str,
    concept: str,
    predicate_filters: list[tuple[str, str, Any]],
    as_of_time: datetime | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> list[Entity]:
    """Query entities matching all predicate filters in one SQL query.

    Uses correlated subqueries so each filter hits the
    idx_assertion_pred_value/idx_assertion_pred_ref indexes instead of
    doing one round-trip per entity. `"eq"` matches either a literal
    property (`value_lit`) or a relation's target entity id
    (`value_ref`) — KI-030: relation filters like `employer="org-123"`
    are equality checks against `value_ref`, not traversal into the
    target entity's own properties. The two are checked via a UNION ALL
    of two single-column point lookups rather than one `value_lit = ?
    OR value_ref = ?` predicate, matching the SQLite backend (whose
    planner doesn't reliably pick a seekable plan for the OR form).
    `"contains"`/`"gt"`/`"lt"`/`"gte"`/`"lte"` (KI-039) check
    `value_lit` only — see this port method's own docstring for why
    relations don't get a UNION ALL branch for those. Numeric range
    comparisons cast to `DOUBLE`, not `REAL` — this module's own header
    comment notes DuckDB's `REAL` is 4-byte single precision, unlike
    SQLite's always-8-byte `REAL`, which would silently round values.

    Args:
        namespace: Namespace to query
        concept: Concept to filter by
        predicate_filters: List of `(full_predicate, operator, value)`
            triples (AND semantics) — see the port method's docstring
            for the operator set
        as_of_time: If set, applies bitemporal filter on assertions and entity creation
        include_flagged: Also match 'flagged' assertions — honored on
            both the current-state and as_of_time paths (KI-081;
            excluded by default on both, see assertions()). On the
            as_of_time path this is point-in-time, not current status
            (KI-097): reconstructed from the assertion_event log the
            same way assertions() already does, so a query pinned to a
            time when an assertion *was* disputed correctly excludes
            it even after the dispute has since been resolved — and,
            the other direction, a time strictly before any dispute
            existed still includes an otherwise-undisputed value, even
            though the same assertion is flagged now.
        include_history: Also match 'superseded'/'retracted' assertions
            (KI-081). On the current-state path this widens the status
            set beyond 'active'. On the as_of_time path, 'superseded' is
            unaffected either way (its window already never restricts to
            'active') — but 'retracted' does something under this flag
            now (ADR-0049, KI-095): as_of_time excludes a retracted
            assertion once its own retraction event's "at" is <=
            as_of_time (a stale, un-narrowed window otherwise keeps
            matching indefinitely); include_history opts back out of
            that exclusion.

    Returns:
        List of entities where all filters match at the given time
    """
    query = "SELECT * FROM entity WHERE namespace = ? AND concept = ?"
    params: list[Any] = [namespace, concept]

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        query += " AND created_at <= ?"
        params.append(t_iso)
        # flagged_clause is always one of exactly two hardcoded literals,
        # never caller-controlled. No # nosec needed here (unlike the
        # match_clause consumers below, e.g. `AND id IN (...SELECT...`):
        # bandit's B608 rule only flags a SQL keyword joined into a
        # string via BinOp/.format()/f-string - never a bare literal
        # Constant, which is all flagged_clause/retracted_clause ever
        # are (confirmed by AST: both branches of the ternary are plain
        # folded string constants). The COALESCE subquery below *does*
        # now contain SELECT/FROM/WHERE/ORDER BY/LIMIT (KI-097 - it
        # didn't when this comment was first written), but since that
        # text lives entirely inside the constant rather than being
        # concatenated onto one, bandit's detector still never sees it -
        # confirmed directly, not assumed (a #nosec placed here was
        # previously dead: removing it left bandit's finding count
        # unchanged).
        # KI-097: flagged-at-t, not current status — same event-log
        # reconstruction assertions() already uses (see its own
        # comment for the full reasoning: a static conflict flags an
        # assertion permanently, so using current status would hide it
        # from as_of() queries for times before the dispute existed,
        # and — the bug this KI fixes — would wrongly *include* it for
        # times during a dispute that has since been resolved, since
        # `reactivated` flips current status back to `active`).
        # COALESCE/tiebreak reasoning identical to assertions()'s own.
        # "at" is quoted — reserved word in DuckDB (see assertions()'s
        # note).
        # Fail-open on a missing event (COALESCE defaults to
        # 'reactivated', i.e. "not flagged"): a status='flagged' row
        # with no matching event — reachable only via a direct
        # put_assertion() bypassing Ontology, or a pre-existing row
        # from before assertion_event tracked this — is visible at
        # every t. The opposite default from retracted_clause below
        # (fail-closed, ADR-0049), but the correct one here: matches
        # assertions()'s own identical COALESCE, and status='flagged'
        # rows always carry a real 'flagged' event through every
        # write path this codebase has (_apply_with_conflict_routing).
        flagged_clause = (
            ""
            if include_flagged
            else (
                " AND COALESCE("
                "(SELECT ae.action FROM assertion_event ae"
                ' WHERE ae.assertion_id = assertion.id AND ae."at" <= ?'
                " AND ae.action IN ('flagged', 'reactivated')"
                ' ORDER BY ae."at" DESC, ae.id DESC LIMIT 1),'
                " 'reactivated'"
                ") != 'flagged'"
            )
        )
        # retracted_clause: same two-hardcoded-literals shape as
        # flagged_clause above, same reasoning for why no #nosec is
        # needed. ADR-0049 (KI-095): a `retracted` assertion's window is
        # not reliably narrowed at retraction time, so `as_of(t)` also
        # excludes it once its own retraction event's "at" is <= t —
        # KI-051 guarantees at most one such event per assertion. "at" is
        # quoted — reserved word in DuckDB (see assertions()'s identical
        # note). `.include_history()` opts back out of this check, the
        # first thing it has ever done on the `as_of` path.
        retracted_clause = (
            ""
            if include_history
            else (
                " AND (status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = assertion.id"
                " AND ae.action = 'retracted' AND ae.\"at\" > ?"
                "))"
            )
        )
        match_clause = (
            " AND asserted_at <= ?"
            " AND (valid_from IS NULL OR valid_from <= ?)"
            " AND (valid_to IS NULL OR valid_to > ?)"
            f"{flagged_clause}"
            f"{retracted_clause}"
        )
        match_params = [t_iso, t_iso, t_iso]
        if not include_flagged:
            match_params.append(t_iso)
        if not include_history:
            match_params.append(t_iso)
    else:
        # Current-state: 'active' only by default;
        # .include_flagged()/.include_history() widen the set (KI-081).
        # `status IN (?, …)` is fully parameter-bound — the interpolated
        # piece is only the placeholder string (`?, ?`), built from
        # `len(match_params)`, never caller input.
        match_params = ["active"]
        if include_flagged:
            match_params.append("flagged")
        if include_history:
            match_params += ["superseded", "retracted"]
        match_clause = f" AND status IN ({', '.join(['?'] * len(match_params))})"

    # predicate/value are always bound via `?` below, never
    # interpolated; the two interpolated pieces are match_clause (built
    # from hardcoded literals, see the flagged_clause justification
    # above) and, for range operators, sql_op — a lookup into the
    # closed, module-level _RANGE_SQL_OPERATORS dict, never the
    # caller's raw operator string. Same already-justified pattern, not
    # a new SQL injection surface.
    for predicate, operator, value in predicate_filters:
        if operator == "eq":
            query += (
                " AND id IN ("  # nosec B608
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND value_lit = ?{match_clause}"
                " UNION ALL "
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND value_ref = ?{match_clause}"
                ")"
            )
            params.extend([predicate, value, *match_params, predicate, value, *match_params])
        elif operator == "contains":
            query += (
                " AND id IN ("  # nosec B608
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND value_lit LIKE ? ESCAPE '\\'{match_clause}"
                ")"
            )
            params.extend([predicate, f"%{_like_escape(value)}%", *match_params])
        else:
            # TRY_CAST, not CAST: QueryBuilder only validates the
            # predicate's *declared* value_type is Integer/Float, never
            # that already-stored value_lit content actually parses as
            # one (KI-049) — a row that doesn't CAST would otherwise
            # raise duckdb.ConversionException uncaught through this
            # port. TRY_CAST returns NULL instead, and NULL compared
            # with any of >/</>=/<= is never true, so the row is simply
            # excluded rather than erroring.
            sql_op = _RANGE_SQL_OPERATORS[operator]
            query += (
                " AND id IN ("  # nosec B608
                "SELECT subject FROM assertion"
                f" WHERE predicate = ? AND TRY_CAST(value_lit AS DOUBLE) {sql_op} ?"
                f"{match_clause}"
                ")"
            )
            params.extend([predicate, value, *match_params])

    cursor = self.conn.execute(query, params)
    rows = cursor.fetchall()

    return [
        Entity(
            id=d["id"],
            namespace=d["namespace"],
            concept=d["concept"],
            natural_key=d["natural_key"],
            created_at=datetime.fromisoformat(d["created_at"]),
            created_by=d["created_by"],
        )
        for d in (self._row_to_dict(cursor, row) for row in rows)
    ]

entities_meeting_confidence

entities_meeting_confidence(
    namespace: str,
    concept: str,
    threshold: float,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]

IDs of entities in (namespace, concept) with >=1 assertion at or above threshold confidence, active at as_of_time (KI-036) or currently active if as_of_time is None. candidate_ids, if given, narrows the scan below (namespace, concept) via unnest() (KI-037) — QueryBuilder only ever passes a set bounded by _CANDIDATE_HINT_MAX (1000), the range measured to be a genuine win here (~2-3x at 10k-50k entities); an earlier, unbounded version of this hint measured unnest() as a regression on both very small candidate sets (~1k entities, where everything is already fast) and large ones (thousands of candidates against a 10k-entity concept,

100x slower) — bounding the hint's size, not avoiding unnest() altogether, is what makes it a reliable win. include_flagged/ include_history widen "active" the same way entities_where() does (KI-093) — see its docstring.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def entities_meeting_confidence(
    self,
    namespace: str,
    concept: str,
    threshold: float,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]:
    """IDs of entities in `(namespace, concept)` with >=1 assertion at or
    above `threshold` confidence, active at `as_of_time` (KI-036) or
    currently active if `as_of_time` is None. `candidate_ids`, if given,
    narrows the scan below `(namespace, concept)` via `unnest()` (KI-037)
    — `QueryBuilder` only ever passes a set bounded by
    `_CANDIDATE_HINT_MAX` (1000), the range measured to be a genuine win
    here (~2-3x at 10k-50k entities); an earlier, unbounded version of
    this hint measured `unnest()` as a regression on both very small
    candidate sets (~1k entities, where everything is already fast) and
    large ones (thousands of candidates against a 10k-entity concept,
    >100x slower) — bounding the hint's size, not avoiding `unnest()`
    altogether, is what makes it a reliable win. `include_flagged`/
    `include_history` widen "active" the same way `entities_where()`
    does (KI-093) — see its docstring."""
    if candidate_ids is not None and not candidate_ids:
        return set()

    query = (
        "SELECT DISTINCT a.subject FROM assertion a"
        " JOIN entity e ON e.id = a.subject"
        " WHERE e.namespace = ? AND e.concept = ? AND a.confidence >= ?"
    )
    params: list[Any] = [namespace, concept, threshold]

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        # One of exactly two hardcoded literals, never caller-controlled
        # - no injection surface, and no #nosec needed (see
        # entities_where()'s identical comment for why bandit's B608
        # heuristic doesn't fire on this shape at all).
        # KI-097: flagged-at-t, not current status — same event-log
        # reconstruction entities_where()/assertions() already use;
        # see entities_where()'s comment for the full reasoning.
        flagged_clause = (
            ""
            if include_flagged
            else (
                " AND COALESCE("
                "(SELECT ae.action FROM assertion_event ae"
                ' WHERE ae.assertion_id = a.id AND ae."at" <= ?'
                " AND ae.action IN ('flagged', 'reactivated')"
                ' ORDER BY ae."at" DESC, ae.id DESC LIMIT 1),'
                " 'reactivated'"
                ") != 'flagged'"
            )
        )
        # ADR-0049 (KI-095): same retraction-aware exclusion
        # entities_where() has — see its comment for the full
        # reasoning. Two-hardcoded-literals shape, no #nosec needed.
        retracted_clause = (
            ""
            if include_history
            else (
                " AND (a.status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = a.id"
                " AND ae.action = 'retracted' AND ae.\"at\" > ?"
                "))"
            )
        )
        query += (
            " AND a.asserted_at <= ?"
            " AND (a.valid_from IS NULL OR a.valid_from <= ?)"
            " AND (a.valid_to IS NULL OR a.valid_to > ?)" + flagged_clause + retracted_clause
        )
        params.extend([t_iso, t_iso, t_iso])
        if not include_flagged:
            params.append(t_iso)
        if not include_history:
            params.append(t_iso)
    else:
        status_params = ["active"]
        if include_flagged:
            status_params.append("flagged")
        if include_history:
            status_params += ["superseded", "retracted"]
        query += f" AND a.status IN ({', '.join(['?'] * len(status_params))})"
        params.extend(status_params)

    if candidate_ids is not None:
        query += " AND e.id IN (SELECT unnest(?))"
        params.append(list(candidate_ids))

    cursor = self.conn.execute(query, params)
    return {row[0] for row in cursor.fetchall()}

entities_meeting_trust

entities_meeting_trust(
    namespace: str,
    concept: str,
    min_trust: int,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]

IDs of entities in (namespace, concept) with >=1 assertion, active at as_of_time (KI-036) or currently active if as_of_time is None, whose effective trust_level >= min_trust (KI-047) — min(author.trust_level, acting_as.trust_level) when the assertion was made under delegation, matching govern/policy.py's identical formula for effective trust (by analogy with SPEC §8.4's capability rule), or just author.trust_level when it wasn't. A dangling acting_as (no resolvable delegate) falls back to author.trust_level via coalesce — see StorageBackend.entities_meeting_trust's docstring for why. candidate_ids narrows the scan the same way as entities_meeting_confidence (KI-037) — see its docstring. include_flagged/include_history widen "active" the same way entities_where() does (KI-093) — see its docstring.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def entities_meeting_trust(
    self,
    namespace: str,
    concept: str,
    min_trust: int,
    as_of_time: datetime | None = None,
    candidate_ids: frozenset[str] | None = None,
    include_flagged: bool = False,
    include_history: bool = False,
) -> set[str]:
    """IDs of entities in `(namespace, concept)` with >=1 assertion,
    active at `as_of_time` (KI-036) or currently active if `as_of_time`
    is None, whose *effective* trust_level >= `min_trust` (KI-047) —
    `min(author.trust_level, acting_as.trust_level)` when the assertion
    was made under delegation, matching `govern/policy.py`'s identical
    formula for effective trust (by analogy with SPEC §8.4's capability
    rule), or just `author.trust_level` when it wasn't. A dangling
    `acting_as` (no resolvable delegate) falls back to `author.trust_level`
    via `coalesce` — see `StorageBackend.entities_meeting_trust`'s
    docstring for why. `candidate_ids` narrows the scan the same way as
    `entities_meeting_confidence` (KI-037) — see its docstring.
    `include_flagged`/`include_history` widen "active" the same way
    `entities_where()` does (KI-093) — see its docstring."""
    if candidate_ids is not None and not candidate_ids:
        return set()

    # DuckDB's `min(a, b)` is aggregate-only (returns a list for two
    # scalar args, confirmed empirically) — `least(a, b)` is the
    # multi-arg scalar form SQLite's `min(a, b)` already is; see this
    # method's SQLite counterpart for the identical formula. The two
    # are NOT equivalent for a NULL operand (sqlite `min(3, NULL)` is
    # NULL, excluding the row; duckdb `least(3, NULL)` is 3, including
    # it) — this can't fire today because `principal.trust_level` is
    # `INTEGER NOT NULL` on both backends and `coalesce` already
    # excludes the no-delegate case, but that NULL-freedom is exactly
    # what this comment is pinning down: relaxing that constraint on
    # either backend would silently diverge the two in opposite
    # directions with no test to catch it.
    query = (
        "SELECT DISTINCT a.subject FROM assertion a"
        " JOIN entity e ON e.id = a.subject"
        " JOIN principal p ON p.id = a.author"
        " LEFT JOIN principal delegate ON delegate.id = a.acting_as"
        " WHERE e.namespace = ? AND e.concept = ?"
        " AND least(p.trust_level, coalesce(delegate.trust_level, p.trust_level)) >= ?"
    )
    params: list[Any] = [namespace, concept, min_trust]

    if as_of_time is not None:
        t_iso = as_of_time.isoformat()
        # One of exactly two hardcoded literals, never caller-controlled
        # - no injection surface, and no #nosec needed (see
        # entities_where()'s identical comment for why bandit's B608
        # heuristic doesn't fire on this shape at all).
        # KI-097: flagged-at-t, not current status — same event-log
        # reconstruction entities_where()/assertions() already use;
        # see entities_where()'s comment for the full reasoning.
        flagged_clause = (
            ""
            if include_flagged
            else (
                " AND COALESCE("
                "(SELECT ae.action FROM assertion_event ae"
                ' WHERE ae.assertion_id = a.id AND ae."at" <= ?'
                " AND ae.action IN ('flagged', 'reactivated')"
                ' ORDER BY ae."at" DESC, ae.id DESC LIMIT 1),'
                " 'reactivated'"
                ") != 'flagged'"
            )
        )
        # ADR-0049 (KI-095): same retraction-aware exclusion
        # entities_where() has — see its comment for the full
        # reasoning. Two-hardcoded-literals shape, no #nosec needed.
        retracted_clause = (
            ""
            if include_history
            else (
                " AND (a.status != 'retracted' OR EXISTS ("
                "SELECT 1 FROM assertion_event ae"
                " WHERE ae.assertion_id = a.id"
                " AND ae.action = 'retracted' AND ae.\"at\" > ?"
                "))"
            )
        )
        query += (
            " AND a.asserted_at <= ?"
            " AND (a.valid_from IS NULL OR a.valid_from <= ?)"
            " AND (a.valid_to IS NULL OR a.valid_to > ?)" + flagged_clause + retracted_clause
        )
        params.extend([t_iso, t_iso, t_iso])
        if not include_flagged:
            params.append(t_iso)
        if not include_history:
            params.append(t_iso)
    else:
        status_params = ["active"]
        if include_flagged:
            status_params.append("flagged")
        if include_history:
            status_params += ["superseded", "retracted"]
        query += f" AND a.status IN ({', '.join(['?'] * len(status_params))})"
        params.extend(status_params)

    if candidate_ids is not None:
        query += " AND e.id IN (SELECT unnest(?))"
        params.append(list(candidate_ids))

    cursor = self.conn.execute(query, params)
    return {row[0] for row in cursor.fetchall()}

close

close() -> None

Close the database connection.

Source code in src/ontolith/store/duckdb/backend.py

@_synchronized
def close(self) -> None:
    """Close the database connection."""
    self.conn.close()

ontolith.store.migrations

Backend-agnostic storage-format migration reporting types (SPEC §15, ADR-0052).

Each concrete adapter (store/sqlite/migrations.py, store/duckdb/migrations.py) owns its own migration registry — the actual up/down DDL, which is inherently backend-specific — and returns results shaped as the dataclasses below so a caller (the CLI, or an SDK user) can report on either backend the same way. See ADR-0052 for the full design: why format_version exists separately from the per-namespace schema_version SPEC §6.4 already tracks, why opening an out-of-date file is refused rather than silently upgraded, and why reversible is declared per migration rather than assumed.

MigrationStep dataclass

MigrationStep(
    version: int,
    description: str,
    reversible: bool,
    applied: bool,
)

One migration's outcome within a MigrationReport.

applied is False for every step in a dry-run report (SPEC §15's "dry-run mode" requirement) — the step is pending, not yet run against the file. In a real (non-dry-run) report every step has applied=True; a migration whose up() itself fails raises instead of appearing here half-applied, so a report is always either fully successful or absent.

version instance-attribute

version: int

The format_version this step moves the database to.

description instance-attribute

description: str

Human-readable summary of what this migration does (and, for an irreversible one, why it can't be undone).

reversible instance-attribute

reversible: bool

Whether this migration's down() can restore the prior version. See ADR-0052's Decision section for what "reversible" means here — the DDL rewrite is invertible; data held only in a column a migration drops is still lost on down(), same as any additive migration's inverse.

applied instance-attribute

applied: bool

True once this step has actually run; False in a dry-run report.

MigrationReport dataclass

MigrationReport(
    from_version: int,
    to_version: int,
    steps: tuple[MigrationStep, ...],
    dry_run: bool,
)

Result of a migrate_file(..., dry_run=...) call, either backend.

from_version instance-attribute

from_version: int

The format_version the file was at before this call (as read, or inferred for a file that predates format_version tracking — see ADR-0052).

to_version instance-attribute

to_version: int

The backend's current format_version (CURRENT_FORMAT_VERSION).

steps instance-attribute

steps: tuple[MigrationStep, ...]

Every migration between from_version and to_version, in the order applied (or that would be applied, for a dry run). Empty when the file was already current.

dry_run instance-attribute

dry_run: bool

Whether this report describes a preview (no writes made) or a real migration that already ran.

up_to_date property

up_to_date: bool

True when there was nothing to migrate (steps is empty).