symfonic.core.prompt.blocks.sources

sources

Concrete :class:~symfonic.core.prompt.blocks.protocol.BlockSource adapters.

BlockComputer module-attribute

BlockComputer: TypeAlias = Callable[
    [TenantScope, str],
    ComputedBlockResult | Awaitable[ComputedBlockResult],
]

The host-supplied callable. Sync or async def; both are accepted.

ComputedBlockResult module-attribute

ComputedBlockResult: TypeAlias = tuple[str, str]

What a host callable returns: (content, revision), in that order.

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.

BlockFileNotFoundError

Bases: BlockFileUnavailableError, NotFoundError

The backing file does not exist.

Derives from both :class:BlockFileUnavailableError (so a resolver catching the source-failure family catches it) and :class:~symfonic.core.protocols.NotFoundError (so "absent" stays distinguishable from "present but unreadable" -- an operator who mistyped a path and one whose deploy dropped read permissions need different fixes).

BlockFileUnavailableError

Bases: StorageError

The backing file exists but could not be read as block content.

A permission denial, a path that is a directory, or content that is not decodable in the declared encoding. Typed as a :class:~symfonic.core.protocols.StorageError so the resolver routes it through the block's on_source_failure policy instead of letting a raw :class:OSError escape past it.

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.

ComputedBlockContractError

Bases: TypeError

The host callable returned something that is not (content, revision).

Deliberately a :class:TypeError and not a :class:~symfonic.core.protocols.StorageError: this is a deterministic bug in host code, not an outage. It must not be swallowed by an omit failure policy, because a block that vanishes silently from every prompt is discovered months later, if at all. See the module docstring.

ComputedBlockSource

ComputedBlockSource(
    compute: BlockComputer,
    *,
    offline_safe: bool = False,
    timeout: float | None = None,
)

Serves one prompt block by calling a host-supplied callable.

Satisfies :class:~symfonic.core.prompt.blocks.protocol.BlockSource structurally and stops there -- see below for why history is not presented.

Parameters:

Name	Type	Description	Default
compute	BlockComputer	Called as compute(scope, block_id), returning (content, revision). May be async def; a sync callable is run in a worker thread. Both arguments are always passed, so a host serving several blocks selects on block_id and a host serving several tenants keys on scope.scope_path.	required
offline_safe	bool	Host-declared, defaulting to False. Whether this block can still be produced when the network and the datastore are unreachable. The default is the pessimistic one because this adapter cannot see inside the callable: a closure over a dict is offline-safe and a query against Postgres is not, and both arrive here as "a callable". An undeclared property is not a guarantee, so the quieter outcome is never the accidental one -- declare True only when the callable genuinely touches nothing remote. Must be an actual bool; a truthy non-bool ("false") is rejected rather than coerced, so a config mistake cannot silently declare an unsafe source safe.	False
timeout	float | None	Seconds to wait for compute before treating it as failed. None (the default) waits indefinitely, matching the adapter's original behaviour. A sync callable that times out still occupies its worker thread in the shared executor until it eventually returns; this bounds the turn, not the thread.	None

Attributes:

Name	Type	Description
scope_aware	bool	Always True. The callable receives the scope and may key on scope.scope_path, so this is the one built-in source a tenant- or profile-scoped block may use. Note what the flag does and does not assert: it permits a non-deployment block scope in :func:~symfonic.core.prompt.blocks.validation.check_scope_pairing. It cannot verify that a given callable actually reads the argument -- one that ignores scope serves global content under a per-tenant declaration, and only that host's own tests can catch it.

Why there is no history: a callable computes a value, it does not retain the values it computed. list_revisions / load_revision are therefore not presented, isinstance(src, HistoryCapableBlockSource) is False, and :func:~symfonic.core.prompt.blocks.validation.check_operator_editable rejects operator_editable=True against it at construction. A host whose backing store does keep history should ship an adapter that presents it, not declare it through this one.

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

def __init__(
    self,
    compute: BlockComputer,
    *,
    offline_safe: bool = False,
    timeout: float | None = None,
) -> None:
    if not callable(compute):
        raise TypeError(
            "ComputedBlockSource requires a callable taking (scope, block_id) "
            f"and returning (content, revision); got {type(compute).__name__}. "
            "A non-callable here would fail on the first turn that resolved "
            "the block, not at the config site that supplied it"
        )
    _reject_wrong_arity(compute)
    if not isinstance(offline_safe, bool):
        raise TypeError(
            f"ComputedBlockSource offline_safe must be a bool; got "
            f"{type(offline_safe).__name__} {offline_safe!r}. Coercing it with "
            "bool() would let a truthy non-bool such as the string 'false' "
            "silently declare an unsafe source safe -- the quieter outcome must "
            "never be the accidental one"
        )
    if timeout is not None and timeout <= 0:
        raise ValueError(f"ComputedBlockSource timeout must be positive; got {timeout!r}")
    self._compute = compute
    self._is_async = _returns_awaitable(compute)
    # Instance attribute rather than a class default: this is the one
    # capability the host declares, and two sources in one process may
    # legitimately declare it differently.
    self.offline_safe = offline_safe
    self._timeout = timeout

compute property

compute: BlockComputer

The host callable this source delegates to.

current_revision async

current_revision(scope: TenantScope, block_id: str) -> str

Return the current revision id without building a revision object.

Not part of :class:BlockSource; offered for parity with :class:~symfonic.core.prompt.blocks.sources.static.StaticBlockSource and :class:~symfonic.core.prompt.blocks.sources.file.FileBlockSource so a cache-validity check has one shape across adapters. Unlike those two, this is not cheap here: it calls compute in full, the same as :meth:load, because a computed value has no separate metadata read -- the callable's return value is the only source of the revision. A caller doing a cache-validity check against a computed-backed block pays the full call, not a shortcut.

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

async def current_revision(self, scope: TenantScope, block_id: str) -> str:
    """Return the current revision id without building a revision object.

    Not part of :class:`BlockSource`; offered for parity with
    :class:`~symfonic.core.prompt.blocks.sources.static.StaticBlockSource`
    and :class:`~symfonic.core.prompt.blocks.sources.file.FileBlockSource`
    so a cache-validity check has one shape across adapters. Unlike
    those two, this is **not** cheap here: it calls ``compute`` in
    full, the same as :meth:`load`, because a computed value has no
    separate metadata read -- the callable's return value *is* the
    only source of the revision. A caller doing a cache-validity check
    against a computed-backed block pays the full call, not a
    shortcut.
    """
    return (await self.load(scope, block_id)).revision

load async

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

Compute the current revision of block_id for scope.

Both arguments are passed through to the host callable verbatim, so content may differ per tenant and per block.

Raises:

Type	Description
ComputedBlockUnavailableError	The callable raised. Routed through the block's on_source_failure policy.
ComputedBlockContractError	The callable returned something other than a (str, str) pair with a non-blank revision. Not routed through the policy -- see the module docstring.

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

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

    Both arguments are passed through to the host callable verbatim,
    so content may differ per tenant and per block.

    Raises:
        ComputedBlockUnavailableError: The callable raised. Routed
            through the block's ``on_source_failure`` policy.
        ComputedBlockContractError: The callable returned something
            other than a ``(str, str)`` pair with a non-blank
            revision. Not routed through the policy -- see the module
            docstring.
    """
    try:
        if self._is_async:
            raw: Any = await asyncio.wait_for(
                self._compute(scope, block_id),  # type: ignore[misc]
                timeout=self._timeout,
            )
        else:
            raw = await asyncio.wait_for(
                asyncio.to_thread(self._compute, scope, block_id), timeout=self._timeout
            )
            if inspect.isawaitable(raw):
                # A sync callable that hands back a coroutine (a lambda
                # wrapping an async function). Awaiting it here is safe:
                # a coroutine is not bound to the thread that made it.
                raw = await asyncio.wait_for(raw, timeout=self._timeout)
    except StorageError:
        # Already typed by the host, and more precisely than this
        # adapter could -- NotFoundError for an absent tenant row
        # stays distinguishable from a generic outage.
        raise
    except TimeoutError as exc:
        # A timed-out sync callable still occupies its worker thread
        # in the default executor until it eventually returns; a
        # dedicated bounded executor is a host-side mitigation this
        # adapter does not impose.
        raise ComputedBlockUnavailableError(
            f"computing block {block_id!r} for scope_path {scope.scope_path!r} did "
            f"not return within {self._timeout}s. The block's on_source_failure "
            "policy decides whether this fails the turn or omits the block"
        ) from exc
    except Exception as exc:
        raise ComputedBlockUnavailableError(
            f"computing block {block_id!r} for scope_path {scope.scope_path!r} "
            f"raised {type(exc).__name__}: {exc}. The block's "
            "on_source_failure policy decides whether this fails the turn or "
            "omits the block"
        ) from exc

    # Outside the try on purpose: a shape violation must surface as
    # the TypeError it is, never re-wrapped into the StorageError
    # family the failure policy absorbs.
    content, revision = self._require_pair(raw, scope=scope, block_id=block_id)
    try:
        digest = content_revision(content).split(":", 1)[1]
    except UnicodeEncodeError as exc:
        # A lone surrogate cannot be encoded as UTF-8. Content that
        # cannot be hashed cannot be served, and it is the callable's
        # output that is malformed -- the same deterministic, will-
        # recur-every-call bug _require_pair exists to catch -- so it
        # is a contract error, not a StorageError an omit policy would
        # silently absorb.
        raise ComputedBlockContractError(
            f"the compute callable for block {block_id!r} at scope_path "
            f"{scope.scope_path!r} returned content that could not be encoded as "
            f"UTF-8 ({exc}); prompt content must be valid Unicode text"
        ) from exc
    return BlockRevision(
        content=content,
        revision=revision,
        # created_at / author / message stay unset: a computed value
        # has no editor and no edit time, and inventing them would
        # make framework filler look like recorded provenance.
        content_hash=digest,
    )

ComputedBlockUnavailableError

Bases: StorageError

The host callable raised while computing the block.

Typed as a :class:~symfonic.core.protocols.StorageError so the resolver routes it through the block's on_source_failure policy instead of letting an arbitrary exception from host code escape past it and fail a turn whose policy said to omit the block.

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)

FileBlockSource

FileBlockSource(
    path: str | Path, *, encoding: str = "utf-8"
)

Serves one prompt block from one file on disk.

Satisfies :class:~symfonic.core.prompt.blocks.protocol.BlockSource structurally and stops there -- see the module docstring for why history is not presented.

Parameters:

Name	Type	Description	Default
path	str | Path	The file backing this block. Resolved once at construction so a later chdir cannot change which file a configured source reads.	required
encoding	str	Text encoding of the file. Content is always hashed as UTF-8, so this affects decoding only, never the revision.	'utf-8'

Attributes:

Name	Type	Description
offline_safe	bool	Always True. Every read is a local filesystem read; there is no datastore, no network client and no cache on any path, including :meth:current_revision.
scope_aware	bool	Always False. One file serves the whole deployment.

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

def __init__(self, path: str | Path, *, encoding: str = "utf-8") -> None:
    # ``Path("")`` normalises to ``Path(".")`` -- str(Path("")) is
    # already "." by the time a Path object exists, so the whitespace
    # check above cannot see the empty string a caller passed in
    # (str or Path). Comparing against Path() (also ".") catches an
    # empty string, an empty Path, and a bare "." spelled directly,
    # all of which name the current working directory rather than a
    # file.
    if not str(path).strip() or Path(path) == Path():
        raise ValueError(
            "FileBlockSource requires a path to the file backing the block; "
            "an empty path names nothing to read -- Path('') is the current "
            "directory, so accepting it would defer the mistake to a confusing "
            "IsADirectoryError at resolve time"
        )
    try:
        codecs.lookup(encoding)
    except LookupError as exc:
        raise ValueError(
            f"FileBlockSource received encoding {encoding!r}, which is not a "
            "codec Python recognises. An unvalidated typo here would otherwise "
            "break every read as a raw LookupError instead of failing at this "
            "config site -- fail_closed every turn, or omit silently forever"
        ) from exc
    # Absolute-ised at construction: a source configured at startup
    # must keep reading the same file no matter what the working
    # directory is by the time a turn resolves its blocks.
    self._path = Path(path).expanduser().absolute()
    self._encoding = encoding

encoding property

encoding: str

The text encoding used to decode the file.

path property

path: Path

The absolute path this source reads. Fixed at construction.

current_revision async

current_revision(scope: TenantScope, block_id: str) -> str

Return the current revision id without building a revision.

Not part of :class:BlockSource; offered so a cache-validity check has a cheap path that stays as offline as :meth:load -- the check reads the same local file and hashes it, and reaches no datastore either.

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

async def current_revision(self, scope: TenantScope, block_id: str) -> str:
    """Return the current revision id without building a revision.

    Not part of :class:`BlockSource`; offered so a cache-validity
    check has a cheap path that stays as offline as
    :meth:`load` -- the check reads the same local file and hashes
    it, and reaches no datastore either.
    """
    return content_revision(await asyncio.to_thread(self._read, block_id))

load async

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

Return the current revision of the backing file.

scope and block_id are accepted because the Protocol has no overload that omits them, and ignored because one instance backs one file: the path selects the content. block_id is still used in failure messages so an unreadable file names the block it was serving.

Raises:

Type	Description
BlockFileNotFoundError	The file does not exist.
BlockFileUnavailableError	The file exists but cannot be read as text in :attr:encoding.

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

async def load(self, scope: TenantScope, block_id: str) -> BlockRevision:
    """Return the current revision of the backing file.

    ``scope`` and ``block_id`` are accepted because the Protocol has
    no overload that omits them, and ignored because one instance
    backs one file: the path selects the content. ``block_id`` is
    still used in failure messages so an unreadable file names the
    block it was serving.

    Raises:
        BlockFileNotFoundError: The file does not exist.
        BlockFileUnavailableError: The file exists but cannot be read
            as text in :attr:`encoding`.
    """
    content = await asyncio.to_thread(self._read, block_id)
    revision = content_revision(content)
    return BlockRevision(
        content=content,
        revision=revision,
        # Everything below is left unset on purpose. A file records
        # no author and no message, and its mtime is not a
        # trustworthy creation time (see the module docstring), so
        # inventing values here would make framework filler
        # indistinguishable from provenance the source actually knew.
        content_hash=revision.split(":", 1)[1],
    )

StaticBlockSource

StaticBlockSource(content: str)

Serves one prompt block from one string fixed in config.

Satisfies :class:~symfonic.core.prompt.blocks.protocol.BlockSource structurally and stops there -- see the module docstring for why history is not presented.

Parameters:

Name	Type	Description	Default
content	str	The block body. Hashed once here to produce the constant revision. Rejected when blank: an empty literal renders an empty labelled section, which reads to the model as a boundary section that exists and says nothing, and is in practice a half-finished config rather than an intent.	required

Attributes:

Name	Type	Description
offline_safe	bool	Always True. The content is already in memory before the first turn; no read of it can fail.
scope_aware	bool	Always False. One literal serves the whole deployment.

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

def __init__(self, content: str) -> None:
    if not isinstance(content, str):
        raise TypeError(
            "StaticBlockSource requires the block content as a str; got "
            f"{type(content).__name__}. The value is rendered into the prompt "
            "verbatim, so coercing it here would hide a config mistake behind "
            "a repr()"
        )
    if not content.strip():
        raise ValueError(
            "StaticBlockSource requires non-blank content; a blank literal "
            "renders a labelled block with an empty body, which is a "
            "half-written config rather than a deliberate empty boundary. "
            "Remove the block from the config to omit it"
        )
    self._content = content
    # Computed once, then never recomputed: the revision must be
    # byte-identical on every call so the cached prefix keyed on it
    # is never invalidated by a block that cannot change.
    self._revision = content_revision(content)

content property

content: str

The literal this source serves. Fixed at construction.

revision property

revision: str

The constant revision id -- sha256:<hex> of :attr:content.

current_revision async

current_revision(scope: TenantScope, block_id: str) -> str

Return the constant revision without building a revision object.

Not part of :class:BlockSource; offered for parity with :class:~symfonic.core.prompt.blocks.sources.file.FileBlockSource so a cache-validity check has one shape across adapters. Here it is a field read, so the check costs nothing.

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

async def current_revision(self, scope: TenantScope, block_id: str) -> str:
    """Return the constant revision without building a revision object.

    Not part of :class:`BlockSource`; offered for parity with
    :class:`~symfonic.core.prompt.blocks.sources.file.FileBlockSource`
    so a cache-validity check has one shape across adapters. Here it
    is a field read, so the check costs nothing.
    """
    return self._revision

load async

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

Return the configured literal and its constant revision.

scope and block_id are accepted because the Protocol has no overload that omits them, and ignored because one instance backs one literal.

Never raises: there is no backing system to be unreachable.

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

async def load(self, scope: TenantScope, block_id: str) -> BlockRevision:
    """Return the configured literal and its constant revision.

    ``scope`` and ``block_id`` are accepted because the Protocol has
    no overload that omits them, and ignored because one instance
    backs one literal.

    Never raises: there is no backing system to be unreachable.
    """
    return BlockRevision(
        content=self._content,
        revision=self._revision,
        # created_at / author / message stay unset. A config literal
        # records no author and no edit time -- version control does,
        # and this adapter does not read it. Filling them here would
        # make framework filler indistinguishable from provenance the
        # source actually knew.
        content_hash=self._revision.split(":", 1)[1],
    )

content_revision

content_revision(content: str) -> str

Return the revision id for content -- sha256:<hex>.

A pure function of the text, with no filesystem access at all, so the same content produces the same revision on every host and at every mtime. Exposed rather than inlined so a cache-invalidation check can compute the expected revision for content it already holds, without a second read.

The text is hashed as UTF-8 regardless of the encoding it was read in: the revision identifies the block's content, so re-saving the same words in a different on-disk encoding must not present itself as an edit.

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

def content_revision(content: str) -> str:
    """Return the revision id for ``content`` -- ``sha256:<hex>``.

    A pure function of the text, with no filesystem access at all, so
    the same content produces the same revision on every host and at
    every mtime. Exposed rather than inlined so a cache-invalidation
    check can compute the expected revision for content it already
    holds, without a second read.

    The text is hashed as UTF-8 regardless of the encoding it was read
    in: the revision identifies the block's *content*, so re-saving the
    same words in a different on-disk encoding must not present itself
    as an edit.
    """
    digest = hashlib.sha256(content.encode("utf-8")).hexdigest()
    return f"{REVISION_ALGORITHM}:{digest}"