symfonic.core.prompt.blocks.sources.database_schema

database_schema

Append-only revision schema for the database-backed block source.

Core owns this schema because core owns its invariants. Sequence allocation, scope keying and append-only discipline live in one module rather than being re-derived by every operator who writes raw SQL against the table.

The shape

One table, prompt_block_revisions. Every write is an INSERT of a new row; there is no mutable "current value" row that an edit overwrites. The head of a block is therefore not a row that gets updated -- it is the row with the highest sequence, selected by :data:SELECT_HEAD.

sequence is unique per (scope_path, block_id) -- it is the leading part of the table's primary key. That single constraint is what makes the head unambiguous under concurrency: two writers that both read head n will both try to insert n + 1, and the database rejects one of them. The loser gets an integrity error to surface as :class:~symfonic.core.prompt.blocks.protocol.RevisionConflictError; it does not get to land a second row tied for latest. A table ordered by created_at instead would accept both and leave two candidate heads with no principled way to pick one.

scope_path is part of the revision table's own key, not merely a column on some head row. Every read statement here filters on it. A revision table keyed on block_id alone would hand every tenant's history to :meth:load_revision, and a history leak is strictly worse than a current-value leak because history is cumulative.

revision and content_hash are deliberately separate columns. content_hash digests the body; revision identifies the row. Restoring an earlier revision is expressed as appending its content again, so the same content_hash legitimately recurs -- only revision is unique. :func:make_revision_id derives one from the sequence, so uniqueness of the identifier follows from the same constraint that makes the head unambiguous.

The stored scope_path value is :attr:~symfonic.core.scope.TenantScope.scope_path verbatim, obtained through :func:~symfonic.core.prompt.blocks.protocol.block_isolation_key -- the same function the rest of the block layer keys caches on. Schema, history filtering and cache keys therefore agree on one canonical representation rather than three near-identical ones.

What append-only means here, and what it does not

Append-only is enforced by core's code path -- this module emits no UPDATE and no DELETE, :func:assert_append_only fails the import if one ever appears, and the INSERT carries no ON CONFLICT DO UPDATE -- and by the unique-sequence constraint, which the database enforces against every connection.

It is not a guarantee that rows cannot change. Any principal with table access -- the table owner, a superuser, a hand-written admin script, a psql session -- can still UPDATE or DELETE a revision. Nothing in a schema can prevent that, and claiming otherwise would misrepresent what an operator is relying on when they treat this history as an audit trail.

Hardening against those principals is an operator-side privilege concern, and it is deliberately not provisioned by this migration. The migration creates no roles and issues no grants, because a library that invents roles in an adopter's database is guessing at an authorization model it cannot see. Recommended practice, to run yourself, is to let the application role insert and select but nothing else::

REVOKE UPDATE, DELETE, TRUNCATE ON prompt_block_revisions FROM app_role;
GRANT INSERT, SELECT ON prompt_block_revisions TO app_role;

That still leaves the table owner and any superuser able to mutate rows; if you need more, the answer is an audit trigger or a write-once tablespace, not this schema.

Dialect

The statements use asyncpg's numbered placeholders and stay inside the SQL subset PostgreSQL and SQLite both accept (CURRENT_TIMESTAMP rather than NOW(), no RETURNING), so the schema's own conformance tests can execute the same statement text the adapter ships rather than a hand-copied approximation. :func:qmark translates the placeholders for drivers using the qmark paramstyle. No dependency beyond the existing [postgres] extra is introduced: this module imports nothing but the standard library and symfonic itself.

assert_append_only

assert_append_only(statements: Iterable[str]) -> None

Raise unless every statement leaves existing revisions untouched.

Run against :data:ALL_STATEMENTS at import, so a mutating statement added to this module fails the import that introduced it rather than the first production write.

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

def assert_append_only(statements: Iterable[str]) -> None:
    """Raise unless every statement leaves existing revisions untouched.

    Run against :data:`ALL_STATEMENTS` at import, so a mutating statement
    added to this module fails the import that introduced it rather than
    the first production write.
    """
    offenders = mutating_statements(statements)
    if offenders:
        raise AssertionError(
            "the block revision schema is append-only, but these statements would "
            f"change or remove existing rows: {offenders!r}"
        )

make_revision_id

make_revision_id(sequence: int, content_hash: str) -> str

Return a revision identifier unique within one block and scope.

Uniqueness rides on sequence, which the primary key already enforces per (scope_path, block_id). The digest tail is carried for legibility only -- it is what makes r7-9f2c... recognisable in a log -- and must never be the sole identity: restoring an earlier revision appends the same content, so the digest repeats by design.

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

def make_revision_id(sequence: int, content_hash: str) -> str:
    """Return a revision identifier unique within one block and scope.

    Uniqueness rides on ``sequence``, which the primary key already
    enforces per ``(scope_path, block_id)``. The digest tail is carried
    for legibility only -- it is what makes ``r7-9f2c...`` recognisable
    in a log -- and must never be the sole identity: restoring an earlier
    revision appends the same content, so the digest repeats by design.
    """
    if sequence < 1:
        raise ValueError(f"revision sequence must be >= 1; got {sequence}")
    digest = content_hash.rsplit(":", 1)[-1]
    if not digest:
        raise ValueError(
            f"content_hash {content_hash!r} carries no digest to build a revision id from"
        )
    return f"r{sequence}-{digest[:16]}"

mutating_statements

mutating_statements(
    statements: Iterable[str],
) -> tuple[str, ...]

Return every statement that would change or remove an existing row.

DROP TABLE is not counted: dropping the table is the declared reverse migration (:data:MIGRATION_DOWN), and it removes the schema rather than editing a revision inside it.

That exemption is unqualified rather than scoped to :data:MIGRATION_DOWN's own statement -- a DROP TABLE reached from anywhere else in the package (a reset() helper, say) would pass this check too. It is not scoped narrower because this checker also runs, unparametrised, against every SQL-looking string literal in the package's source (including partial fragments an f-string splits across ast.Constant nodes, which never equal a complete MIGRATION_DOWN statement); a membership test against :data:MIGRATION_DOWN would silently stop matching those fragments and reintroduce the exact gap ALTER TABLE was added above to close. Narrowing this safely needs the fragment-scanning test to cooperate, not just this function.

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

def mutating_statements(statements: Iterable[str]) -> tuple[str, ...]:
    """Return every statement that would change or remove an existing row.

    ``DROP TABLE`` is not counted: dropping the table is the declared
    reverse migration (:data:`MIGRATION_DOWN`), and it removes the schema
    rather than editing a revision inside it.

    That exemption is unqualified rather than scoped to
    :data:`MIGRATION_DOWN`'s own statement -- a ``DROP TABLE`` reached
    from anywhere else in the package (a ``reset()`` helper, say) would
    pass this check too. It is not scoped narrower because this checker
    also runs, unparametrised, against every SQL-looking string literal in
    the package's source (including partial fragments an f-string splits
    across ``ast.Constant`` nodes, which never equal a complete
    ``MIGRATION_DOWN`` statement); a membership test against
    :data:`MIGRATION_DOWN` would silently stop matching those fragments
    and reintroduce the exact gap ``ALTER TABLE`` was added above to
    close. Narrowing this safely needs the fragment-scanning test to
    cooperate, not just this function.
    """
    return tuple(sql for sql in statements if _MUTATING_SQL.search(sql))

qmark

qmark(sql: str) -> str

Translate $1-style placeholders into the qmark paramstyle.

The statements are written once, in the style asyncpg speaks. A driver using qmark (sqlite3, and the schema's own conformance tests) runs the same text through this rather than keeping a second, silently divergent copy.

Positional rewriting is only sound when the placeholders appear in ascending order and each is used once, so that is checked rather than assumed.

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

def qmark(sql: str) -> str:
    """Translate ``$1``-style placeholders into the qmark paramstyle.

    The statements are written once, in the style asyncpg speaks. A
    driver using qmark (``sqlite3``, and the schema's own conformance
    tests) runs the same text through this rather than keeping a second,
    silently divergent copy.

    Positional rewriting is only sound when the placeholders appear in
    ascending order and each is used once, so that is checked rather than
    assumed.
    """
    numbers = [int(n) for n in _PLACEHOLDER.findall(sql)]
    if numbers != list(range(1, len(numbers) + 1)):
        raise ValueError(
            f"cannot rewrite placeholders positionally: {sql!r} uses them out of order "
            "or more than once, so a qmark driver would bind different values"
        )
    return _PLACEHOLDER.sub("?", sql)

revision_key

revision_key(
    scope: TenantScope, block_id: str
) -> tuple[str, str]

Return the (scope_path, block_id) values to bind into a statement.

Delegates to :func:~symfonic.core.prompt.blocks.protocol.block_isolation_key so the value stored in the scope_path column is exactly :attr:TenantScope.scope_path -- the same string the block layer keys its caches on. Deriving it here independently would be one more place for the canonical representation to drift.

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

def revision_key(scope: TenantScope, block_id: str) -> tuple[str, str]:
    """Return the ``(scope_path, block_id)`` values to bind into a statement.

    Delegates to
    :func:`~symfonic.core.prompt.blocks.protocol.block_isolation_key` so
    the value stored in the ``scope_path`` column is exactly
    :attr:`TenantScope.scope_path` -- the same string the block layer
    keys its caches on. Deriving it here independently would be one more
    place for the canonical representation to drift.
    """
    return block_isolation_key(scope, block_id)