symfonic.core.prompt.blocks.sources.database

database

DatabaseBlockSource -- a prompt block read from, and appended to, SQL.

This is the adapter an operator's own admin system sits behind. Identity and rules become per-tenant rows rather than redeployed files: the operator edits a block in their admin surface, the host application calls :meth:DatabaseBlockSource.append_revision, and the next turn's prompt reads the new head.

It implements all three block Protocols -- :class:~symfonic.core.prompt.blocks.protocol.BlockSource, :class:~symfonic.core.prompt.blocks.protocol.HistoryCapableBlockSource and :class:~symfonic.core.prompt.blocks.protocol.WritableBlockSource -- over the append-only schema in :mod:~symfonic.core.prompt.blocks.sources.database_schema. It defines no SQL of its own; every statement it executes is a constant imported from that module, which is where the append-only assertion runs.

The head is the highest sequence, never the newest timestamp

:meth:load returns the row selected by SELECT_HEAD: highest sequence for this (scope_path, block_id). Ordering by created_at would be wrong in a way that only shows up under load -- two rows can share a timestamp (coarse clock granularity, a batch insert inside one transaction), leaving two candidate heads and no principled tiebreak, and a clock adjustment can order them backwards outright. sequence is the leading part of the primary key, so it is unique by construction and the head is a single unambiguous row.

block_id is in every WHERE clause alongside scope_path. One table backs every block of every tenant; a read that filtered on scope alone would hand back whichever block's row sorted first.

Writing is append-only, and operator-facing

:meth:append_revision issues one INSERT. There is no update, no delete, no rewind and no restore-in-place on this class -- the verbs simply do not exist, so no caller can reach for one. Restoring revision N is performed by reading it with :meth:load_revision and appending its content as a new revision, which leaves the intervening revisions in :meth:list_revisions and makes the restore itself an auditable entry.

This method is not an agent tool. It is called by the host application from its own admin surface, where the operator's identity and authorization model already live. This stage registers no block-edit tool in the agent's palette, ships no admin endpoint, and ships no authorization surface: core cannot see an adopter's permission model, and inventing one here would be a guess wearing the costume of a security control. Authorizing the caller is the host's job, and it happens before this method is reached.

Concurrency has two lines of defence

1. :func:~symfonic.core.prompt.blocks.protocol.ensure_expected_head compares expected_head against the head just read, and rejects a caller working from a revision that has already been superseded.
2. The primary key on (scope_path, block_id, sequence) catches the race the pre-check cannot: two writers that both read head n both pass the pre-check, both try to claim sequence n+1, and the database rejects one of them. That loser's integrity error is translated into the same :class:~symfonic.core.prompt.blocks.protocol.RevisionConflictError the pre-check raises, so a caller mapping conflicts to a 409 handles one exception type and not two.

The pre-check alone would be a check-then-act race; the constraint alone would surface as a driver-specific integrity error with no indication of what the current head actually is. Both are needed.

Offline and scope posture

offline_safe is False: every read crosses a connection to another process. A block served from here does not survive the datastore being unreachable, and :func:~symfonic.core.prompt.blocks.validation.warn_if_offline_unsafe says so when such a block is placed in an authored tier.

scope_aware is True: the isolation key is :attr:~symfonic.core.scope.TenantScope.scope_path (via :func:~symfonic.core.prompt.blocks.protocol.block_isolation_key), the full root-first path rather than tenant_id, so two brands under one org keep separate blocks and separate histories.

Schema creation is explicit

:meth:ensure_schema exists but is never called from a read or write path. A library that runs DDL lazily on first use is a library that creates tables in whichever database a misconfigured DSN happened to point at. The host calls it once at boot, or runs :data:~symfonic.core.prompt.blocks.sources.database_schema.MIGRATION_UP through its own migration tool.

BlockDatabaseUnavailableError

Bases: StorageError

The backing database could not serve the block.

Typed as a :class:~symfonic.core.protocols.StorageError so the resolver routes it through the block's on_source_failure policy rather than letting a driver exception escape past that policy and turn an optional block into a failed turn.

BlockRevisionNotFoundError

Bases: BlockDatabaseUnavailableError, NotFoundError

No such revision for this block in this scope.

Raised by :meth:DatabaseBlockSource.load when a block has never been written, and by :meth:DatabaseBlockSource.load_revision when the named revision does not exist for this scope -- a revision id belonging to another tenant is not found here, which is the same answer an id that never existed gets.

DatabaseBlockSource

DatabaseBlockSource(pool: Any)

Serves prompt blocks from the append-only revision table.

One instance backs every block of every scope in one database: the (scope_path, block_id) pair in each statement selects the rows, so a deployment needs one of these rather than one per block.

Parameters:

Name	Type	Description	Default
pool	Any	Anything exposing await acquire() and await release(conn) and handing back connections that speak the asyncpg connection API (execute, fetch, fetchrow, with $n placeholders). Both :class:~symfonic.memory.backends.pool.PostgresPoolManager and a raw asyncpg.Pool qualify. Injected rather than constructed so this adapter opens no second connection pool beside the one the deployment already runs.	required

Attributes:

Name	Type	Description
offline_safe	bool	Always False -- every read crosses a connection, so a block served from here does not survive the datastore being unreachable.
scope_aware	bool	Always True -- every statement is filtered on scope.scope_path.

Source code in src/symfonic/core/prompt/blocks/sources/database.py

def __init__(self, pool: Any) -> None:
    if pool is None:
        raise ValueError(
            "DatabaseBlockSource requires a connection pool; without one there "
            "is nothing to read blocks from, and deferring the failure to the "
            "first turn would surface it as a mid-prompt source failure"
        )
    self._pool = pool

append_revision async

append_revision(
    scope: TenantScope,
    block_id: str,
    content: str,
    *,
    author: str | None,
    message: str | None,
    expected_head: str | None,
) -> BlockRevision

Append a new revision of block_id for scope.

Operator-facing: called by the host application from its own admin surface, never registered as an agent tool. Authorizing the caller happens before this method is reached.

One INSERT, no ON CONFLICT clause. Existing revisions are left exactly as they were, and the previous head stays reachable through :meth:load_revision.

Parameters:

Name	Type	Description	Default
scope	TenantScope	Isolation argument; the stored key is scope.scope_path.	required
block_id	str	The block spec's name.	required
content	str	The new body.	required
author	str | None	Who asked for the write, or None when genuinely unknown -- never a defaulted "system".	required
message	str | None	Why, or None.	required
expected_head	str | None	The revision the caller believes is current, or None to assert the block has none yet.	required

Returns:

Type	Description
BlockRevision	The appended :class:BlockRevision.

Raises:

Type	Description
RevisionConflictError	expected_head is not the current head, or another writer claimed the same sequence first.

Source code in src/symfonic/core/prompt/blocks/sources/database.py

async def append_revision(
    self,
    scope: TenantScope,
    block_id: str,
    content: str,
    *,
    author: str | None,
    message: str | None,
    expected_head: str | None,
) -> BlockRevision:
    """Append a new revision of ``block_id`` for ``scope``.

    Operator-facing: called by the host application from its own
    admin surface, never registered as an agent tool. Authorizing the
    caller happens before this method is reached.

    One ``INSERT``, no ``ON CONFLICT`` clause. Existing revisions are
    left exactly as they were, and the previous head stays reachable
    through :meth:`load_revision`.

    Args:
        scope: Isolation argument; the stored key is
            ``scope.scope_path``.
        block_id: The block spec's name.
        content: The new body.
        author: Who asked for the write, or ``None`` when genuinely
            unknown -- never a defaulted ``"system"``.
        message: Why, or ``None``.
        expected_head: The revision the caller believes is current,
            or ``None`` to assert the block has none yet.

    Returns:
        The appended :class:`BlockRevision`.

    Raises:
        RevisionConflictError: ``expected_head`` is not the current
            head, or another writer claimed the same sequence first.
    """
    scope_path, bid = revision_key(scope, block_id)
    conn = await self._acquire()
    try:
        try:
            head = await conn.fetchrow(SELECT_HEAD, scope_path, bid)
        except Exception as exc:
            raise BlockDatabaseUnavailableError(
                f"reading the current head of block {block_id!r} at scope_path "
                f"{scope_path!r} raised {type(exc).__name__}: {exc}"
            ) from exc
        ensure_expected_head(
            scope,
            block_id,
            expected_head=expected_head,
            actual_head=None if head is None else head["revision"],
        )
        # The sequence is derived from the head just validated, so
        # the row claims exactly one past the revision the caller
        # was working from. Re-reading MAX(sequence) separately
        # would let it land past a revision nobody checked.
        sequence = 1 if head is None else int(head["sequence"]) + 1
        parent = None if head is None else head["revision"]
        content_hash = content_revision(content)
        revision = make_revision_id(sequence, content_hash)
        try:
            await conn.execute(
                INSERT_REVISION,
                scope_path,
                bid,
                sequence,
                revision,
                parent,
                content,
                content_hash,
                author,
                message,
            )
        except Exception as exc:
            if not is_unique_violation(exc):
                raise BlockDatabaseUnavailableError(
                    f"appending a revision of block {block_id!r} at scope_path "
                    f"{scope_path!r} raised {type(exc).__name__}: {exc}"
                ) from exc
            # The pre-check passed and the constraint still refused:
            # a concurrent writer claimed this sequence in between.
            # Re-read so the caller is told what the head actually
            # is now, and raise the same error the pre-check does.
            try:
                current = await conn.fetchrow(SELECT_HEAD, scope_path, bid)
            except Exception as read_exc:
                raise BlockDatabaseUnavailableError(
                    f"re-reading the head of block {block_id!r} at scope_path "
                    f"{scope_path!r} after a rejected append raised "
                    f"{type(read_exc).__name__}: {read_exc}"
                ) from read_exc
            current_head = None if current is None else current["revision"]
            if current_head == expected_head:
                # The head has not moved since expected_head was
                # validated, so this unique-violation-shaped exception
                # was not caused by a concurrent writer claiming our
                # sequence -- something else in the row violated a
                # constraint this adapter did not anticipate, and
                # reporting it as a conflict would tell the caller
                # retrying is the fix, when it is not.
                raise BlockDatabaseUnavailableError(
                    f"appending a revision of block {block_id!r} at scope_path "
                    f"{scope_path!r} raised {type(exc).__name__}: {exc}, but the "
                    "head is unchanged since expected_head was validated -- this "
                    "is not the concurrent-writer race this adapter reports as a "
                    "conflict"
                ) from exc
            raise RevisionConflictError(
                scope_path=scope_path,
                block_id=block_id,
                expected_head=expected_head,
                actual_head=current_head,
            ) from exc
        try:
            written = await conn.fetchrow(SELECT_REVISION, scope_path, bid, revision)
        except Exception as exc:
            raise BlockDatabaseUnavailableError(
                f"reading back revision {revision!r} of block {block_id!r} at "
                f"scope_path {scope_path!r} raised {type(exc).__name__}: {exc}"
            ) from exc
    finally:
        await self._pool.release(conn)
    if written is None:  # pragma: no cover - the insert just succeeded
        raise BlockDatabaseUnavailableError(
            f"revision {revision!r} of block {block_id!r} was inserted at "
            f"scope_path {scope_path!r} but could not be read back"
        )
    return _to_revision(written)

ensure_schema async

ensure_schema() -> None

Create the revision table if it is absent. Idempotent.

Never called from :meth:load or :meth:append_revision; see the module docstring for why DDL stays off the hot paths.

CREATE TABLE IF NOT EXISTS is not race-safe on every engine: two replicas booting at once can both attempt it, and the loser may get a duplicate-key error on a system catalog index instead of a silent no-op. That shape is indistinguishable from any other unique-violation-shaped exception, so it is recognised with the same :func:is_unique_violation the write path uses rather than a second heuristic, and treated as success: the table exists either way, which is everything this method promises.

Raises:

Type	Description
BlockDatabaseUnavailableError	The pool or the driver failed for a reason other than losing this race.

Source code in src/symfonic/core/prompt/blocks/sources/database.py

async def ensure_schema(self) -> None:
    """Create the revision table if it is absent. Idempotent.

    Never called from :meth:`load` or :meth:`append_revision`; see
    the module docstring for why DDL stays off the hot paths.

    ``CREATE TABLE IF NOT EXISTS`` is not race-safe on every engine:
    two replicas booting at once can both attempt it, and the loser
    may get a duplicate-key error on a system catalog index instead of
    a silent no-op. That shape is indistinguishable from *any other*
    unique-violation-shaped exception, so it is recognised with the
    same :func:`is_unique_violation` the write path uses rather than a
    second heuristic, and treated as success: the table exists either
    way, which is everything this method promises.

    Raises:
        BlockDatabaseUnavailableError: The pool or the driver failed
            for a reason other than losing this race.
    """
    conn = await self._acquire()
    try:
        for statement in MIGRATION_UP:
            try:
                await conn.execute(statement)
            except Exception as exc:
                if is_unique_violation(exc):
                    # Another writer's concurrent CREATE TABLE won the
                    # race; the table this call wanted now exists.
                    continue
                raise BlockDatabaseUnavailableError(
                    f"running the block-revision migration raised "
                    f"{type(exc).__name__}: {exc}"
                ) from exc
    finally:
        await self._pool.release(conn)

list_revisions async

list_revisions(
    scope: TenantScope,
    block_id: str,
    *,
    limit: int | None = None,
) -> Sequence[BlockRevision]

Return revisions of block_id for scope, newest first.

Scoped and block-filtered like every other read. History is cumulative, so a history read that dropped either filter would leak strictly more than a current-value read.

Parameters:

Name	Type	Description	Default
limit	int | None	Not part of :class:HistoryCapableBlockSource; an optional bound so a caller paging through a long-edited block's history is not forced to materialise every revision -- and every restore appends a row rather than replacing one, so that history only grows. None (the default) returns the full history, unchanged from before this parameter existed.	None

Raises:

Type	Description
ValueError	limit is not a positive integer.
BlockDatabaseUnavailableError	The pool or the driver failed.

Source code in src/symfonic/core/prompt/blocks/sources/database.py

async def list_revisions(
    self, scope: TenantScope, block_id: str, *, limit: int | None = None
) -> Sequence[BlockRevision]:
    """Return revisions of ``block_id`` for ``scope``, newest first.

    Scoped and block-filtered like every other read. History is
    cumulative, so a history read that dropped either filter would
    leak strictly more than a current-value read.

    Args:
        limit: Not part of :class:`HistoryCapableBlockSource`; an
            optional bound so a caller paging through a long-edited
            block's history is not forced to materialise every
            revision -- and every restore appends a row rather than
            replacing one, so that history only grows. ``None`` (the
            default) returns the full history, unchanged from before
            this parameter existed.

    Raises:
        ValueError: ``limit`` is not a positive integer.
        BlockDatabaseUnavailableError: The pool or the driver failed.
    """
    if limit is not None and limit <= 0:
        raise ValueError(f"list_revisions limit must be a positive integer; got {limit!r}")
    key = revision_key(scope, block_id)
    statement = LIST_REVISIONS if limit is None else LIST_REVISIONS_LIMITED
    args = key if limit is None else (*key, limit)
    conn = await self._acquire()
    try:
        try:
            rows = await conn.fetch(statement, *args)
        except Exception as exc:
            raise BlockDatabaseUnavailableError(
                f"listing revisions of block {block_id!r} at scope_path {key[0]!r} "
                f"raised {type(exc).__name__}: {exc}"
            ) from exc
    finally:
        await self._pool.release(conn)
    return tuple(_to_revision(row) for row in rows)

load async

load(scope: TenantScope, block_id: str) -> BlockRevision

Return the head revision of block_id for scope.

The head is the highest sequence for the pair, not the newest created_at and not whichever row the table returns first.

Raises:

Type	Description
BlockRevisionNotFoundError	The block has never been written for this scope.

Source code in src/symfonic/core/prompt/blocks/sources/database.py

async def load(self, scope: TenantScope, block_id: str) -> BlockRevision:
    """Return the head revision of ``block_id`` for ``scope``.

    The head is the highest ``sequence`` for the pair, not the newest
    ``created_at`` and not whichever row the table returns first.

    Raises:
        BlockRevisionNotFoundError: The block has never been written
            for this scope.
    """
    key = revision_key(scope, block_id)
    row = await self._fetchrow(SELECT_HEAD, *key)
    if row is None:
        raise BlockRevisionNotFoundError(
            f"block {block_id!r} has no revision at scope_path {key[0]!r}; the "
            "block's on_source_failure policy decides whether this fails the "
            "turn or omits the block"
        )
    return _to_revision(row)

load_revision async

load_revision(
    scope: TenantScope, block_id: str, revision: str
) -> BlockRevision

Return one named prior revision of block_id for scope.

scope is part of the lookup, not inferred from revision: revision ids are unique only within a scope's block, so trusting the id alone would read across tenants.

Raises:

Type	Description
BlockRevisionNotFoundError	No such revision in this scope.

Source code in src/symfonic/core/prompt/blocks/sources/database.py

async def load_revision(
    self, scope: TenantScope, block_id: str, revision: str
) -> BlockRevision:
    """Return one named prior revision of ``block_id`` for ``scope``.

    ``scope`` is part of the lookup, not inferred from ``revision``:
    revision ids are unique only within a scope's block, so trusting
    the id alone would read across tenants.

    Raises:
        BlockRevisionNotFoundError: No such revision in this scope.
    """
    key = revision_key(scope, block_id)
    row = await self._fetchrow(SELECT_REVISION, *key, revision)
    if row is None:
        raise BlockRevisionNotFoundError(
            f"block {block_id!r} at scope_path {key[0]!r} has no revision "
            f"{revision!r}; a revision recorded for a different scope is not "
            "reachable from this one"
        )
    return _to_revision(row)

is_unique_violation

is_unique_violation(exc: BaseException) -> bool

Return True when exc reports a duplicate-key rejection.

Matched without importing any driver: asyncpg, psycopg and friends carry sqlstate (psycopg2 spells it pgcode), and DB-API drivers without one (sqlite3) raise a class named IntegrityError. Importing every driver this adapter might be handed a connection from would drag optional extras into a code path that only needs to classify an error.

A driver-reported code is authoritative and decides the answer on its own: NOT NULL (23502), CHECK and FOREIGN KEY (23503) violations all subclass IntegrityError too, so matching on the class name alone -- the pre-fix fallback -- reported every one of them as a duplicate key. Only a driver that exposes no code at all falls through to the name-plus-message heuristic, and even then the message must actually say so; IntegrityError alone is not enough.

Source code in src/symfonic/core/prompt/blocks/sources/database.py

def is_unique_violation(exc: BaseException) -> bool:
    """Return ``True`` when ``exc`` reports a duplicate-key rejection.

    Matched without importing any driver: asyncpg, psycopg and friends
    carry ``sqlstate`` (psycopg2 spells it ``pgcode``), and DB-API drivers
    without one (``sqlite3``) raise a class named ``IntegrityError``.
    Importing every driver this adapter might be handed a connection from
    would drag optional extras into a code path that only needs to
    classify an error.

    A driver-reported code is authoritative and decides the answer on its
    own: NOT NULL (``23502``), CHECK and FOREIGN KEY (``23503``)
    violations all subclass ``IntegrityError`` too, so matching on the
    class name alone -- the pre-fix fallback -- reported every one of them
    as a duplicate key. Only a driver that exposes no code at all falls
    through to the name-plus-message heuristic, and even then the message
    must actually say so; ``IntegrityError`` alone is not enough.
    """
    code = getattr(exc, "sqlstate", None) or getattr(exc, "pgcode", None)
    if code is not None:
        return code == UNIQUE_VIOLATION_SQLSTATE
    return any(base.__name__ == "IntegrityError" for base in type(exc).__mro__) and (
        "unique" in str(exc).lower()
    )