Persistence¶

async def connect(self) -> None:
    """Establish connection to the storage backend.

    Raises:
        PersistenceConnectionError: If the connection cannot be
            established.
    """
    ...

async def disconnect(self) -> None:
    """Close the storage backend connection.

    Safe to call even if not connected.
    """
    ...

async def health_check(self) -> bool:
    """Check whether the backend is healthy and responsive.

    Returns:
        ``True`` if the backend is reachable and operational.
    """
    ...

async def migrate(self) -> None:
    """Run pending schema migrations.

    Raises:
        MigrationError: If a migration fails.
    """
    ...

def get_db(self) -> Any:
    """Return the underlying database connection.

    Returns:
        The raw database connection object (backend-specific).

    Raises:
        PersistenceConnectionError: If not yet connected.
    """
    ...

def write_context(self) -> AbstractAsyncContextManager[None]:
    """Async context manager around mutating SQL on this backend.

    The mutual-exclusion guarantee is backend-specific:

    - **SQLite** acquires a shared in-process write lock so that
      multi-statement transactions on the single
      ``aiosqlite.Connection`` cannot interleave at the statement
      level. Concurrent writers on the same backend instance
      serialize.
    - **Postgres** yields immediately: each repository operation
      checks out an independent connection from the async pool, so
      writers are already isolated at the database level. The
      method exists on this backend only to keep the cross-backend
      interface uniform; it does not provide mutual exclusion
      beyond what the pool already gives.

    Use it in repository write paths so the same code path works
    on both backends::

        async with backend.write_context():
            await db.execute(...)
            await db.commit()

    Each call returns a fresh context manager. On SQLite, the
    underlying lock primitive is shared across calls so concurrent
    callers serialize. On Postgres, there is no shared primitive.
    Repositories are wired with this method (as a callable) at
    backend construction; callers that already hold a
    ``PersistenceBackend`` reference can use it directly for
    cross-repo transactional boundaries.

    Callers must not rely on ``write_context`` for distributed
    mutual exclusion or cross-backend serializability.
    """
    ...

def build_lockouts(self, auth_config: AuthConfig) -> LockoutRepository:
    """Construct a lockout repository for this backend.

    Method-based rather than property because :class:`LockoutRepository`
    needs the operator's ``AuthConfig`` (threshold, window, duration)
    which is app-layer config, not persistence-layer.  Callers supply
    the config at startup; the returned repo shares this backend's
    connection / pool.

    Raises:
        PersistenceConnectionError: If the backend is not connected.
    """
    ...

def build_escalations(
    self,
    *,
    notify_channel: str | None = None,
) -> EscalationQueueRepository:
    """Construct an escalation queue repository for this backend.

    Method-based rather than property because Postgres escalations
    accept an optional NOTIFY channel name -- cross-instance notify
    config lives on the escalation subsystem, not on persistence.
    ``notify_channel`` is ignored by the SQLite implementation.

    Raises:
        PersistenceConnectionError: If the backend is not connected.
    """
    ...

def build_ontology_versioning(
    self,
) -> VersioningService[EntityDefinition]:
    """Construct the ontology versioning service bound to this backend.

    Returns a versioning service wired to the backend's active DB
    handle.  SQLite implementations bind the service to their
    ``aiosqlite.Connection``; Postgres implementations bind to their
    ``AsyncConnectionPool``.

    Raises:
        PersistenceConnectionError: If the backend is not connected.
    """
    ...

async def get_setting(self, key: NotBlankStr) -> str | None:
    """Retrieve a setting value by key.

    Args:
        key: Setting key.

    Returns:
        The setting value, or ``None`` if not found.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def set_setting(self, key: NotBlankStr, value: str) -> None:
    """Store a setting value.

    Upserts -- creates or updates the key.

    Args:
        key: Setting key.
        value: Setting value.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def save(self, entity: AgentRuntimeState) -> None:
    """Upsert an agent runtime state by ``agent_id``.

    Args:
        entity: The agent runtime state to persist.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def get(self, entity_id: NotBlankStr) -> AgentRuntimeState | None:
    """Retrieve an agent runtime state by agent ID.

    Args:
        entity_id: The agent identifier.

    Returns:
        The agent state, or ``None`` if not found.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def list_items(
    self,
    *,
    limit: int = DEFAULT_PAGE_SIZE,
    offset: int = 0,
) -> tuple[AgentRuntimeState, ...]:
    """List all agent runtime states in ``agent_id`` order.

    Args:
        limit: Maximum rows to return.
        offset: Rows to skip from the head of the ordering.

    Returns:
        Agent states in ascending ``agent_id`` order.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def get_active(
    self,
    *,
    limit: int = DEFAULT_PAGE_SIZE,
    offset: int = 0,
) -> tuple[AgentRuntimeState, ...]:
    """Retrieve a bounded page of non-idle agent states.

    Returns states where ``status != 'idle'``, ordered by
    ``last_activity_at`` descending then ``agent_id`` ascending
    (the stable secondary key makes paging deterministic when
    activity timestamps tie). Callers that need every active
    state drain via
    :func:`synthorg.persistence._shared.collect_all`.

    Args:
        limit: Maximum rows to return.
        offset: Rows to skip from the head of the ordering.

    Returns:
        A page of active agent states.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def delete(self, entity_id: NotBlankStr) -> bool:
    """Delete an agent runtime state by agent ID.

    Args:
        entity_id: The agent identifier.

    Returns:
        ``True`` if deleted, ``False`` if not found.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def save(self, entity: Artifact) -> None:
    """Persist an artifact (insert or update by id).

    Args:
        entity: The artifact to persist.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def save_returning_outcome(self, artifact: Artifact) -> bool:
    """Persist an artifact atomically and return the insert/update outcome.

    This is a D7 bespoke method retained for the audit-event correctness
    invariant: callers need to know whether the write was an insert or
    update so they can emit the corresponding ``API_ARTIFACT_CREATED`` or
    ``API_ARTIFACT_UPDATED`` event. The outcome is computed atomically
    with the write (SQLite: ``INSERT ... ON CONFLICT(id) DO NOTHING``
    rowcount; Postgres: ``xmax = 0 AS created``) to avoid the TOCTOU
    window of a separate ``get()`` probe.

    Args:
        artifact: The artifact to persist.

    Returns:
        ``True`` when this call inserted a new row, ``False`` when
        it updated an existing row in place.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def get(self, entity_id: NotBlankStr) -> Artifact | None:
    """Retrieve an artifact by its ID.

    Args:
        entity_id: The artifact identifier.

    Returns:
        The artifact, or ``None`` if not found.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def list_items(
    self,
    *,
    limit: int = DEFAULT_PAGE_SIZE,
    offset: int = 0,
) -> tuple[Artifact, ...]:
    """List all artifacts with pagination.

    Results are ordered by artifact ID ascending to ensure
    deterministic pagination across backends.

    Args:
        limit: Maximum rows to return.
        offset: Rows to skip before the window.

    Returns:
        Artifacts ordered by id ascending.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def query(
    self,
    filter_spec: ArtifactFilterSpec,
    *,
    limit: int = DEFAULT_PAGE_SIZE,
    offset: int = 0,
) -> tuple[Artifact, ...]:
    """List artifacts matching the filter spec (paginated).

    Results are ordered by artifact ID ascending to ensure
    deterministic pagination across backends.

    Args:
        filter_spec: Carries optional filters for task_id, created_by,
            artifact_type.
        limit: Maximum rows to return.
        offset: Rows to skip before the window.

    Returns:
        Matching artifacts ordered by ID, as a tuple.

    Raises:
        PersistenceError: If the operation fails.
        QueryError: If ``limit < 1`` or ``offset < 0``.
    """
    ...

async def count(self, filter_spec: ArtifactFilterSpec) -> int:
    """Count artifacts matching the filter spec.

    Args:
        filter_spec: Carries optional filters.

    Returns:
        Total number of matching artifacts.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def delete(self, entity_id: NotBlankStr) -> bool:
    """Delete an artifact by ID.

    Args:
        entity_id: The artifact identifier.

    Returns:
        ``True`` if the artifact was deleted, ``False`` if not found.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def append(self, entry: AuditEntry) -> None:
    """Persist an audit entry (append-only).

    Args:
        entry: The audit entry to persist.

    Raises:
        DuplicateRecordError: If an entry with the same ID exists.
        QueryError: If the operation fails.
    """
    ...

async def query(
    self,
    filter_spec: AuditFilterSpec,
    *,
    limit: int = DEFAULT_PAGE_SIZE,
    offset: int = 0,
) -> tuple[AuditEntry, ...]:
    """Return audit entries matching the filter spec (paginated).

    Results are ordered by timestamp descending (newest first).

    Args:
        filter_spec: Audit filter specification with optional filters.
        limit: Maximum number of entries to return (must be >= 1).
        offset: Number of entries to skip (for pagination).

    Returns:
        Matching audit entries as a tuple.

    Raises:
        QueryError: If the operation fails, *limit* < 1, or
            *until* is earlier than *since* in the filter spec.
    """
    ...

async def purge_before(self, cutoff: AwareDatetime) -> int:
    """Delete audit entries older than *cutoff* (CFG-1 audit).

    This is the one exception to the append-only rule: it powers
    the retention sweeper which enforces the operator-configurable
    ``security.audit_retention_days`` window. Rows are removed
    permanently; the retention-vs-forensic tradeoff is decided at
    the retention-window level, not per row.

    Args:
        cutoff: Entries strictly older than this UTC timestamp
            are deleted.

    Returns:
        Number of rows removed.

    Raises:
        QueryError: If the operation fails.
    """
    ...

async def append(self, checkpoint: Checkpoint) -> None:
    """Persist a checkpoint row (append-only).

    Args:
        checkpoint: The checkpoint to persist.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...

async def query(
    self,
    filter_spec: CheckpointFilterSpec,
    *,
    limit: int = DEFAULT_PAGE_SIZE,
    offset: int = 0,
) -> tuple[Checkpoint, ...]:
    """Return checkpoints matching the filter, newest first."""
    ...

async def purge_before(self, threshold: AwareDatetime) -> int:
    """Delete checkpoints with ``saved_at < threshold``.

    ``threshold`` must be timezone-aware UTC; naive datetimes are
    rejected at the boundary so purge cut-offs do not depend on the
    caller's local-time assumption.
    """
    ...

async def get_latest(
    self,
    *,
    execution_id: NotBlankStr | None = None,
    task_id: NotBlankStr | None = None,
) -> Checkpoint | None:
    """Retrieve the latest checkpoint by turn_number.

    At least one filter (``execution_id`` or ``task_id``) is required.

    Args:
        execution_id: Filter by execution identifier.
        task_id: Filter by task identifier.

    Returns:
        The checkpoint with the highest turn_number, or ``None``.

    Raises:
        PersistenceError: If the operation fails.
        ValueError: If neither filter is provided.
    """
    ...

async def delete_by_execution(self, execution_id: NotBlankStr) -> int:
    """Delete all checkpoints for an execution.

    Args:
        execution_id: The execution identifier.

    Returns:
        Number of checkpoints deleted.

    Raises:
        PersistenceError: If the operation fails.
    """
    ...