ontolith.core

ontolith.core

Core abstractions and ports for Ontolith.

This module contains the fundamental building blocks: - Clock and IdProvider ports for deterministic behavior - Embedder port for hybrid retrieval (SPEC §11.3) - ObservabilitySink port for metrics/events/structured logs (SPEC §18, ADR-0044) - Error taxonomy for stable error handling - Entity and Assertion value objects

The meta-model/IR/validation this docstring once listed as "Future" now lives in ontolith.schema (SchemaIR, ConceptDef/PropertyDef/ RelationDef, the class DSL) — not ontolith.core, and shipped in M1/M3.

Assertion

Bases: BaseModel

Atomic unit of knowledge with full provenance.

Attributes:

Name	Type	Description
id	str	Unique identifier (ULID)
namespace	str	Namespace slug
subject	str	Entity ID this assertion is about
predicate	str	Qualified property/relation name (e.g., "Person.name")
value_kind	Literal['literal', 'ref']	Whether value is a literal or entity reference
value_type	str | None	Type of literal value (if value_kind is "literal")
value	str	The actual value (literal or entity ID)
author	str	Principal ID who made this assertion
acting_as	str | None	Principal ID if delegated (optional)
source	str | None	Source of information (URI or description)
confidence	float | None	Author's stated belief (0.0-1.0)
rationale	str | None	Why this assertion was made
model	str | None	Model family+version for AI authors
asserted_at	datetime	When we learned this fact
valid_from	datetime | None	When the fact became/becomes true (defaults to asserted_at)
valid_to	datetime | None	When the fact stopped being true (NULL = still valid)
status	Literal['active', 'superseded', 'retracted', 'flagged']	Current lifecycle state
proposal_id	str | None	Proposal that introduced this assertion
supersedes	str | None	Previous assertion ID if this supersedes another
metadata	dict[str, Any]	Open JSON blob for future evolution

set_defaults_and_validate classmethod

set_defaults_and_validate(
    values: dict[str, Any],
) -> dict[str, Any]

Set valid_from default and validate value_type based on value_kind.

Source code in src/ontolith/core/assertion.py

@model_validator(mode="before")
@classmethod
def set_defaults_and_validate(cls, values: dict[str, Any]) -> dict[str, Any]:
    """Set valid_from default and validate value_type based on value_kind."""
    # Default valid_from to asserted_at (SPEC §5.3)
    if "valid_from" not in values or values["valid_from"] is None:
        values["valid_from"] = values.get("asserted_at")

    # Validate value_type based on value_kind
    value_kind = values.get("value_kind")
    value_type = values.get("value_type")

    if value_kind == "literal" and value_type is None:
        raise ValueError("value_type is required when value_kind is 'literal'")
    if value_kind == "ref" and value_type is not None:
        raise ValueError("value_type must be None when value_kind is 'ref'")

    return values

AssertionEvent

Bases: BaseModel

Append-only audit record of an assertion status mutation.

Assertions themselves are append-only (only status/valid_to/supersedes mutate in place) — this event log makes each such mutation independently attributable and timestamped, since the assertion row itself is overwritten in place and doesn't retain who caused a given transition or when it happened (only its current status).

Attributes:

Name	Type	Description
id	str	Unique event ID (ULID)
assertion_id	str	Assertion this event was recorded against
actor	str	Principal ID accountable for the transition — the author of the assertion that triggered supersession/contradiction during conflict routing, or the principal who explicitly retracted/ resolved it
action	Literal['superseded', 'flagged', 'retracted', 'reactivated']	Which status transition this event records
at	datetime	When the transition occurred
successor_id	str | None	For action="superseded", the id of the assertion that caused this supersession. Assertion.supersedes only records the first predecessor when one incoming assertion supersedes several concurrently-overlapping ones (KI-008) — querying events by successor_id recovers the full predecessor set. None for other actions.

Clock

Bases: ABC

Abstract clock for deterministic time handling.

now abstractmethod

now() -> datetime

Return the current time.

Returns:

Type	Description
datetime	Current datetime in UTC.

Source code in src/ontolith/core/clock.py

@abstractmethod
def now(self) -> datetime:
    """Return the current time.

    Returns:
        Current datetime in UTC.
    """
    ...

FixedClock

FixedClock(fixed_time: str | datetime | None = None)

Bases: Clock

Test clock that returns a fixed time.

Useful for testing time-sensitive logic without flakiness.

Parameters:

Name	Type	Description	Default
fixed_time	str | datetime | None	The time to always return, or current time if None.	None

Example

clock = FixedClock("2025-01-01T00:00:00Z") clock.now() datetime.datetime(2025, 1, 1, 0, 0, tzinfo=datetime.timezone.utc) clock.advance(days=1) clock.now() datetime.datetime(2025, 1, 2, 0, 0, tzinfo=datetime.timezone.utc)

Initialize with a fixed time.

Parameters:

Name	Type	Description	Default
fixed_time	str | datetime | None	ISO-8601 string, datetime, or None for current time.	None

Source code in src/ontolith/core/clock.py

def __init__(self, fixed_time: str | datetime | None = None) -> None:
    """Initialize with a fixed time.

    Args:
        fixed_time: ISO-8601 string, datetime, or None for current time.
    """
    if fixed_time is None:
        self._time = datetime.now(UTC)
    elif isinstance(fixed_time, str):
        self._time = datetime.fromisoformat(fixed_time.replace("Z", "+00:00"))
    else:
        self._time = fixed_time

    # Ensure timezone-aware
    if self._time.tzinfo is None:
        self._time = self._time.replace(tzinfo=UTC)

now

now() -> datetime

Return the fixed time.

Returns:

Type	Description
datetime	The configured fixed datetime.

Source code in src/ontolith/core/clock.py

def now(self) -> datetime:
    """Return the fixed time.

    Returns:
        The configured fixed datetime.
    """
    return self._time

set

set(time: str | datetime) -> None

Set the clock to a new time.

Parameters:

Name	Type	Description	Default
time	str | datetime	ISO-8601 string or datetime to set.	required

Source code in src/ontolith/core/clock.py

def set(self, time: str | datetime) -> None:
    """Set the clock to a new time.

    Args:
        time: ISO-8601 string or datetime to set.
    """
    if isinstance(time, str):
        self._time = datetime.fromisoformat(time.replace("Z", "+00:00"))
    else:
        self._time = time

    if self._time.tzinfo is None:
        self._time = self._time.replace(tzinfo=UTC)

advance

advance(
    *,
    days: int = 0,
    hours: int = 0,
    minutes: int = 0,
    seconds: int = 0,
) -> None

Advance the clock by the specified duration.

Parameters:

Name	Type	Description	Default
days	int	Number of days to advance.	0
hours	int	Number of hours to advance.	0
minutes	int	Number of minutes to advance.	0
seconds	int	Number of seconds to advance.	0

Source code in src/ontolith/core/clock.py

def advance(
    self,
    *,
    days: int = 0,
    hours: int = 0,
    minutes: int = 0,
    seconds: int = 0,
) -> None:
    """Advance the clock by the specified duration.

    Args:
        days: Number of days to advance.
        hours: Number of hours to advance.
        minutes: Number of minutes to advance.
        seconds: Number of seconds to advance.
    """
    self._time += timedelta(
        days=days,
        hours=hours,
        minutes=minutes,
        seconds=seconds,
    )

SystemClock

Bases: Clock

Production clock using system time.

now

now() -> datetime

Return the current system time in UTC.

Returns:

Type	Description
datetime	Current datetime in UTC timezone.

Source code in src/ontolith/core/clock.py

def now(self) -> datetime:
    """Return the current system time in UTC.

    Returns:
        Current datetime in UTC timezone.
    """
    return datetime.now(UTC)

Embedder

Bases: Protocol

Converts text into L2-unit-normalized embedding vectors.

embed

embed(texts: list[str]) -> list[list[float]]

Embed a batch of texts.

Parameters:

Name	Type	Description	Default
texts	list[str]	Texts to embed.	required

Returns:

Type	Description
list[list[float]]	One L2-unit-normalized vector of length dim per input text, in
list[list[float]]	the same order. A text with no extractable signal (empty or
list[list[float]]	whitespace-only) yields the zero vector, not an error.

Source code in src/ontolith/core/embedder.py

def embed(self, texts: list[str]) -> list[list[float]]:
    """Embed a batch of texts.

    Args:
        texts: Texts to embed.

    Returns:
        One L2-unit-normalized vector of length `dim` per input text, in
        the same order. A text with no extractable signal (empty or
        whitespace-only) yields the zero vector, not an error.
    """
    ...

HashingEmbedder

HashingEmbedder(dim: int = 256)

Deterministic, dependency-free default Embedder.

Implements the feature-hashing trick (Weinberger et al., 2009): each token hashes to a bucket index and a sign, avoiding the need for a pre-built vocabulary or any numeric library. This is a placeholder for real semantic quality, not a substitute for a model-backed Embedder — it guarantees deterministic, reproducible vectors for a given text, not meaningful semantic similarity between unrelated texts.

Uses hashlib.sha256, not Python's built-in hash(), because str hashing is randomly salted per process (PYTHONHASHSEED) and would break determinism across runs.

Initialize with a fixed output dimensionality.

Parameters:

Name	Type	Description	Default
dim	int	Length of each output vector.	256

Source code in src/ontolith/core/embedder.py

def __init__(self, dim: int = 256) -> None:
    """Initialize with a fixed output dimensionality.

    Args:
        dim: Length of each output vector.
    """
    self.dim = dim

embed

embed(texts: list[str]) -> list[list[float]]

Embed a batch of texts via feature hashing.

Parameters:

Name	Type	Description	Default
texts	list[str]	Texts to embed.	required

Returns:

Type	Description
list[list[float]]	One L2-unit-normalized vector of length dim per input text.

Source code in src/ontolith/core/embedder.py

def embed(self, texts: list[str]) -> list[list[float]]:
    """Embed a batch of texts via feature hashing.

    Args:
        texts: Texts to embed.

    Returns:
        One L2-unit-normalized vector of length `dim` per input text.
    """
    return [self._embed_one(text) for text in texts]

LookupEmbedder

LookupEmbedder(
    mapping: dict[str, list[float]],
    dim: int,
    default: list[float] | None = None,
)

Deterministic test double backed by an explicit text-to-vector map.

Unlike HashingEmbedder, output is fully predictable — use this in tests that need to assert exact rank order from semantic search.

Initialize with an explicit text-to-vector mapping.

Parameters:

Name	Type	Description	Default
mapping	dict[str, list[float]]	Exact text to vector lookup.	required
dim	int	Length of vectors in mapping and default.	required
default	list[float] | None	Vector to return for unmapped texts, or None to raise.	None

Source code in src/ontolith/core/embedder.py

def __init__(
    self, mapping: dict[str, list[float]], dim: int, default: list[float] | None = None
) -> None:
    """Initialize with an explicit text-to-vector mapping.

    Args:
        mapping: Exact text to vector lookup.
        dim: Length of vectors in `mapping` and `default`.
        default: Vector to return for unmapped texts, or None to raise.
    """
    self._mapping = mapping
    self.dim = dim
    self._default = default

embed

embed(texts: list[str]) -> list[list[float]]

Look up each text's vector.

Parameters:

Name	Type	Description	Default
texts	list[str]	Texts to look up.	required

Returns:

Type	Description
list[list[float]]	The mapped (or default) vector per input text.

Raises:

Type	Description
KeyError	A text is not in mapping and no default was given.

Source code in src/ontolith/core/embedder.py

def embed(self, texts: list[str]) -> list[list[float]]:
    """Look up each text's vector.

    Args:
        texts: Texts to look up.

    Returns:
        The mapped (or default) vector per input text.

    Raises:
        KeyError: A text is not in `mapping` and no `default` was given.
    """
    result = []
    for text in texts:
        if text in self._mapping:
            result.append(self._mapping[text])
        elif self._default is not None:
            result.append(self._default)
        else:
            raise KeyError(f"LookupEmbedder: no vector mapped for {text!r}")
    return result

Entity

Bases: BaseModel

Concrete individual of a concept.

Attributes:

Name	Type	Description
id	str	Unique identifier (ULID)
namespace	str	Namespace slug
concept	str	Concept name (e.g., "Person", "Organization")
natural_key	str | None	Optional unique key within (namespace, concept)
created_at	datetime	When this entity was created
created_by	str	Principal ID who created this entity

AuthError

AuthError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Authentication failed or no identity resolved.

Examples: - Invalid credentials - Expired token - Unknown principal

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

CapabilityError

CapabilityError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Principal lacks required capability for operation.

Examples: - Agent attempting direct write (requires 'write' capability) - User trying to review without 'review' capability - Missing admin capability for schema changes

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

ConflictError

ConflictError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Unresolved contradiction blocks an operation.

Examples: - Query returns entities with flagged contradictions - Attempt to write when contradiction is open - Conflict resolution required before proceeding

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

NotFoundError

NotFoundError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Entity, assertion, or namespace not found.

Examples: - Entity ID doesn't exist - Namespace not initialized - Assertion ID invalid

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

OntolithError

OntolithError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: Exception

Base exception for all Ontolith errors.

All errors carry: - A stable code for programmatic handling - A human-readable message - Optional structured detail for additional context

Initialize error with message and optional detail.

Parameters:

Name	Type	Description	Default
message	str	Human-readable error message.	required
detail	dict[str, Any] | None	Optional structured context (e.g., field names, constraints).	None

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

PluginError

PluginError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Plugin load or execution failure.

Examples: - Plugin not found - Plugin manifest invalid - Plugin raised exception - Sandbox violation

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

PolicyDenied

PolicyDenied(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Proposal rejected by policy engine.

The rejection reason is in the message and detail dict.

Examples: - Confidence below auto-accept threshold - Missing required source - Trust level too low

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

SchemaError

SchemaError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Invalid or incompatible schema or migration.

Examples: - Breaking schema change without migration - Invalid concept/property definition - Schema version conflict

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

StorageError

StorageError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Storage backend failure.

Examples: - Database connection failed - Transaction timeout - Disk full - Backend-specific errors

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

ValidationError

ValidationError(
    message: str, detail: dict[str, Any] | None = None
)

Bases: OntolithError

Validator or constraint failure.

Carries a list of violations in the detail dict.

Examples: - Required field missing - Type mismatch - Cardinality violation - Custom validator rejection

Source code in src/ontolith/core/errors.py

def __init__(self, message: str, detail: dict[str, Any] | None = None) -> None:
    """Initialize error with message and optional detail.

    Args:
        message: Human-readable error message.
        detail: Optional structured context (e.g., field names, constraints).
    """
    super().__init__(message)
    self.message = message
    self.detail = detail or {}

FixedIdProvider

FixedIdProvider(ids: list[str])

Bases: IdProvider

Test ID provider that cycles through a fixed list of IDs.

Useful for tests that need specific, predictable IDs.

Parameters:

Name	Type	Description	Default
ids	list[str]	List of IDs to cycle through.	required

Example

provider = FixedIdProvider(["alice-id", "bob-id"]) provider.next() 'alice-id' provider.next() 'bob-id' provider.next() # cycles back 'alice-id'

Initialize with a fixed list of IDs.

Parameters:

Name	Type	Description	Default
ids	list[str]	List of ID strings to cycle through.	required

Raises:

Type	Description
ValueError	If ids list is empty.

Source code in src/ontolith/core/ids.py

def __init__(self, ids: list[str]) -> None:
    """Initialize with a fixed list of IDs.

    Args:
        ids: List of ID strings to cycle through.

    Raises:
        ValueError: If ids list is empty.
    """
    if not ids:
        raise ValueError("FixedIdProvider requires at least one ID")
    self._ids = ids
    self._index = 0

next

next() -> str

Return the next ID from the list, cycling if necessary.

Returns:

Type	Description
str	Next ID from the configured list.

Source code in src/ontolith/core/ids.py

def next(self) -> str:
    """Return the next ID from the list, cycling if necessary.

    Returns:
        Next ID from the configured list.
    """
    id_str = self._ids[self._index]
    self._index = (self._index + 1) % len(self._ids)
    return id_str

reset

reset() -> None

Reset to the first ID in the list.

Source code in src/ontolith/core/ids.py

def reset(self) -> None:
    """Reset to the first ID in the list."""
    self._index = 0

IdProvider

Bases: ABC

Abstract ID provider for deterministic ID generation.

next abstractmethod

next() -> str

Generate the next ID.

Returns:

Type	Description
str	A unique identifier as a string.

Source code in src/ontolith/core/ids.py

@abstractmethod
def next(self) -> str:
    """Generate the next ID.

    Returns:
        A unique identifier as a string.
    """
    ...

SequentialIdProvider

SequentialIdProvider(prefix: str = 'id', start: int = 1)

Bases: IdProvider

Test ID provider that generates sequential IDs.

Useful for: - Deterministic testing (same IDs on every run) - Readable test fixtures ("id-001", "id-002", etc.) - Predictable ordering in tests

Parameters:

Name	Type	Description	Default
prefix	str	Prefix for generated IDs.	'id'
start	int	Starting number (default 1).	1

Example

provider = SequentialIdProvider(prefix="test", start=1) provider.next() 'test-001' provider.next() 'test-002'

Initialize with optional prefix and starting number.

Parameters:

Name	Type	Description	Default
prefix	str	String prefix for IDs.	'id'
start	int	Starting counter value.	1

Source code in src/ontolith/core/ids.py

def __init__(self, prefix: str = "id", start: int = 1) -> None:
    """Initialize with optional prefix and starting number.

    Args:
        prefix: String prefix for IDs.
        start: Starting counter value.
    """
    self._prefix = prefix
    self._counter = start

next

next() -> str

Generate the next sequential ID.

Returns:

Type	Description
str	A sequential ID string like "prefix-001".

Source code in src/ontolith/core/ids.py

def next(self) -> str:
    """Generate the next sequential ID.

    Returns:
        A sequential ID string like "prefix-001".
    """
    current = self._counter
    self._counter += 1
    return f"{self._prefix}-{current:03d}"

reset

reset(start: int = 1) -> None

Reset the counter.

Parameters:

Name	Type	Description	Default
start	int	Value to reset counter to.	1

Source code in src/ontolith/core/ids.py

def reset(self, start: int = 1) -> None:
    """Reset the counter.

    Args:
        start: Value to reset counter to.
    """
    self._counter = start

UlidProvider

Bases: IdProvider

Production ID provider using ULID.

ULIDs are: - Lexicographically sortable - Timestamp-based (first 48 bits) - URL-safe - 26 characters (vs UUID's 36) - Compatible with UUID tools

next

next() -> str

Generate a new ULID.

Returns:

Type	Description
str	A ULID string (26 characters).

Source code in src/ontolith/core/ids.py

def next(self) -> str:
    """Generate a new ULID.

    Returns:
        A ULID string (26 characters).
    """
    return str(ULID())

Namespace

Bases: BaseModel

A registered namespace.

Attributes:

Name	Type	Description
id	str	Namespace slug
created_at	datetime	When this namespace was registered
metadata	dict[str, Any]	Open JSON blob for future evolution

NullObservabilitySink

Bases: ObservabilitySink

No-op sink — for a deployment or test that wants Ontolith to stay silent by explicit choice, unlike StdlibLoggingSink (the default when no sink is injected at all).

log

log(level: int, message: str, **fields: object) -> None

Discard level/message/**fields — no-op.

Source code in src/ontolith/core/observability.py

def log(self, level: int, message: str, **fields: object) -> None:
    """Discard `level`/`message`/`**fields` — no-op."""

record_event

record_event(kind: str, **fields: object) -> None

Discard kind/**fields — no-op.

Source code in src/ontolith/core/observability.py

def record_event(self, kind: str, **fields: object) -> None:
    """Discard `kind`/`**fields` — no-op."""

record_metric

record_metric(name: str, value: float, **tags: str) -> None

Discard name/value/**tags — no-op.

Source code in src/ontolith/core/observability.py

def record_metric(self, name: str, value: float, **tags: str) -> None:
    """Discard `name`/`value`/`**tags` — no-op."""

ObservabilitySink

Bases: ABC

Abstract sink for structured logs, lifecycle events, and metrics (SPEC §18, ADR-0044).

Correlation fields (namespace, principal, acting_as, proposal_id, ...) are carried, not re-derived: callers thread whatever is already in scope through as keyword fields on the call itself — this port has no context-propagation mechanism of its own (ADR-0044 deliberately rejected a contextvars-based one).

govern/policy never calls this port — SPEC's purity requirement for the policy engine has no observability carve-out. Instrumentation happens one layer up, in Ontology's own orchestration and in interfaces/*'s request-level handling.

log abstractmethod

log(level: int, message: str, **fields: object) -> None

Emit a structured, correlated log line.

Parameters:

Name	Type	Description	Default
level	int	A stdlib logging level (logging.DEBUG/INFO/ WARNING/ERROR/CRITICAL), not a sink-specific scale — every sink is expected to understand these five.	required
message	str	Human-readable message, already fully formed (no %-style interpolation placeholders — callers format before calling this method, since not every sink is stdlib logging-backed).	required
**fields	object	Structured, correlated context (e.g. namespace=, principal=, acting_as=, proposal_id=, code=). exc_info=<exception> is a recognized-by-convention field for a genuinely unhandled exception a sink may want to capture a traceback for (StdlibLoggingSink does; a sink that doesn't understand it just reports it as a plain field, no different from any other).	{}

Source code in src/ontolith/core/observability.py

@abstractmethod
def log(self, level: int, message: str, **fields: object) -> None:
    """Emit a structured, correlated log line.

    Args:
        level: A stdlib `logging` level (`logging.DEBUG`/`INFO`/
            `WARNING`/`ERROR`/`CRITICAL`), not a sink-specific scale —
            every sink is expected to understand these five.
        message: Human-readable message, already fully formed (no
            %-style interpolation placeholders — callers format before
            calling this method, since not every sink is stdlib
            `logging`-backed).
        **fields: Structured, correlated context (e.g. `namespace=`,
            `principal=`, `acting_as=`, `proposal_id=`, `code=`).
            `exc_info=<exception>` is a recognized-by-convention field
            for a genuinely unhandled exception a sink may want to
            capture a traceback for (`StdlibLoggingSink` does; a sink
            that doesn't understand it just reports it as a plain
            field, no different from any other).
    """
    ...

record_event abstractmethod

record_event(kind: str, **fields: object) -> None

Record one occurrence of a named lifecycle event (SPEC §18's four: proposal lifecycle, contradiction open/resolve, schema migration, plugin load).

Parameters:

Name	Type	Description	Default
kind	str	Event kind, e.g. "proposal.accepted", "contradiction.opened".	required
**fields	object	Structured, correlated context for this occurrence.	{}

Source code in src/ontolith/core/observability.py

@abstractmethod
def record_event(self, kind: str, **fields: object) -> None:
    """Record one occurrence of a named lifecycle event (SPEC §18's
    four: proposal lifecycle, contradiction open/resolve, schema
    migration, plugin load).

    Args:
        kind: Event kind, e.g. "proposal.accepted", "contradiction.opened".
        **fields: Structured, correlated context for this occurrence.
    """
    ...

record_metric abstractmethod

record_metric(name: str, value: float, **tags: str) -> None

Record one observation of a named metric (SPEC §18's seven).

Parameters:

Name	Type	Description	Default
name	str	Metric name, e.g. "proposals.accepted", "query.latency_ms".	required
value	float	The observed value.	required
**tags	str	Metric dimensions (e.g. namespace=, backend=) — kept as plain strings, unlike log()/record_event()'s arbitrary-object fields, since every metrics backend this port is likely to sit in front of treats tags/labels as strings.	{}

Source code in src/ontolith/core/observability.py

@abstractmethod
def record_metric(self, name: str, value: float, **tags: str) -> None:
    """Record one observation of a named metric (SPEC §18's seven).

    Args:
        name: Metric name, e.g. "proposals.accepted", "query.latency_ms".
        value: The observed value.
        **tags: Metric dimensions (e.g. `namespace=`, `backend=`) — kept
            as plain strings, unlike `log()`/`record_event()`'s
            arbitrary-object fields, since every metrics backend this
            port is likely to sit in front of treats tags/labels as
            strings.
    """
    ...

RecordingObservabilitySink

RecordingObservabilitySink()

Bases: ObservabilitySink

Test double that accumulates every call (mirrors FixedClock/ FixedIdProvider's role for their own ports) — for tests asserting exactly what was logged/recorded, not just that no exception was raised.

Source code in src/ontolith/core/observability.py

def __init__(self) -> None:
    self.logs: list[tuple[int, str, dict[str, object]]] = []
    self.events: list[tuple[str, dict[str, object]]] = []
    self.metrics: list[tuple[str, float, dict[str, str]]] = []

log

log(level: int, message: str, **fields: object) -> None

Append (level, message, fields) to self.logs.

Source code in src/ontolith/core/observability.py

def log(self, level: int, message: str, **fields: object) -> None:
    """Append `(level, message, fields)` to `self.logs`."""
    self.logs.append((level, message, fields))

record_event

record_event(kind: str, **fields: object) -> None

Append (kind, fields) to self.events.

Source code in src/ontolith/core/observability.py

def record_event(self, kind: str, **fields: object) -> None:
    """Append `(kind, fields)` to `self.events`."""
    self.events.append((kind, fields))

record_metric

record_metric(name: str, value: float, **tags: str) -> None

Append (name, value, tags) to self.metrics.

Source code in src/ontolith/core/observability.py

def record_metric(self, name: str, value: float, **tags: str) -> None:
    """Append `(name, value, tags)` to `self.metrics`."""
    self.metrics.append((name, value, tags))

StdlibLoggingSink

StdlibLoggingSink(
    logger_name: str = "ontolith.observability",
)

Bases: ObservabilitySink

Production-safe default sink (mirrors Clock's SystemClock): writes through Python's own logging module, under an ontolith.observability logger by default so a deployment can filter or route it independently of any other logger this project uses directly (e.g. a module still doing its own logging.getLogger(__name__) for something outside this port's scope).

This is not the tier-(c) metrics backend ADR-0044 defers to its own follow-up ADR — record_metric/record_event land here as a visible, zero-configuration stopgap (a structured log line), not a real counter/gauge a monitoring stack could scrape. Swap in a concrete observe/-adapter sink once one exists for that.

Source code in src/ontolith/core/observability.py

def __init__(self, logger_name: str = "ontolith.observability") -> None:
    self._logger = logging.getLogger(logger_name)

log

log(level: int, message: str, **fields: object) -> None

Emit message (plus any **fields) through the wrapped logging.Logger at level.

A caller logging a genuinely unhandled exception (not a domain OntolithError) may pass exc_info=<the exception> to get a real traceback in the log record, the same as calling logging.Logger.exception()/log(..., exc_info=...) directly would — pulled out of **fields rather than a dedicated parameter so every other sink's log() signature stays exactly SPEC §18's three-argument shape; a sink that can't act on it just reports it as a plain field instead (a non-None, non-BaseException value is treated as a plain field here too, not a malformed exc_info; exc_info=None — the same as not passing it at all — is dropped silently, matching logging.Logger's own convention that None means "no exception").

Source code in src/ontolith/core/observability.py

def log(self, level: int, message: str, **fields: object) -> None:
    """Emit `message` (plus any `**fields`) through the wrapped
    `logging.Logger` at `level`.

    A caller logging a genuinely unhandled exception (not a domain
    `OntolithError`) may pass `exc_info=<the exception>` to get a real
    traceback in the log record, the same as calling
    `logging.Logger.exception()`/`log(..., exc_info=...)` directly
    would — pulled out of `**fields` rather than a dedicated parameter
    so every other sink's `log()` signature stays exactly SPEC §18's
    three-argument shape; a sink that can't act on it just reports it
    as a plain field instead (a non-`None`, non-`BaseException` value
    is treated as a plain field here too, not a malformed `exc_info`;
    `exc_info=None` — the same as not passing it at all — is dropped
    silently, matching `logging.Logger`'s own convention that `None`
    means "no exception").
    """
    exc_info_field = fields.pop("exc_info", None)
    if isinstance(exc_info_field, BaseException):
        exc_info: BaseException | None = exc_info_field
    else:
        exc_info = None
        if exc_info_field is not None:
            fields["exc_info"] = exc_info_field
    if fields:
        self._logger.log(level, "%s %s", message, fields, exc_info=exc_info)
    else:
        self._logger.log(level, message, exc_info=exc_info)

record_event

record_event(kind: str, **fields: object) -> None

Log kind and **fields as one INFO-level structured line — the tier-(a)/(b) stopgap described in this class's own docstring, not a real event store.

Source code in src/ontolith/core/observability.py

def record_event(self, kind: str, **fields: object) -> None:
    """Log `kind` and `**fields` as one INFO-level structured line —
    the tier-(a)/(b) stopgap described in this class's own docstring,
    not a real event store."""
    self._logger.info("event=%s %s", kind, fields)

record_metric

record_metric(name: str, value: float, **tags: str) -> None

Log name, value, and **tags as one INFO-level structured line — the tier-(a)/(b) stopgap described in this class's own docstring, not a real metrics backend.

Source code in src/ontolith/core/observability.py

def record_metric(self, name: str, value: float, **tags: str) -> None:
    """Log `name`, `value`, and `**tags` as one INFO-level structured
    line — the tier-(a)/(b) stopgap described in this class's own
    docstring, not a real metrics backend."""
    self._logger.info("metric=%s value=%s %s", name, value, tags)