symfonic.core.prompt.blocks.sources.computed

computed

ComputedBlockSource -- a prompt block the host computes per call.

ENVIRONMENT is the block this exists for. "You are running in production, region eu-west-1, against the live billing API" is a deployment fact that is known at runtime and stale the moment it is pinned in a config literal. This adapter hands the decision back to the host: it calls a supplied callable with (scope, block_id) and takes back (content, revision).

It is the one built-in adapter that can legitimately serve different content per tenant. That matters more than it looks: without it, every scope-aware path in the block layer -- scope_path keying, the per-tenant branch of the resolver, REQ-SUBAGENT's scope propagation -- would have nothing to exercise it until a database adapter shipped, and a code path with no honest caller is a code path that is wrong by the time it gets one. A host closure over a dict is enough to keep those paths tested for real.

The host owns the revision, and therefore owns the cache

The callable returns the revision alongside the content, rather than this adapter hashing what came back. Hashing here would look safer and would quietly be worse: content that is semantically unchanged but textually noisy -- a timestamp in the rendered text, a dict iterated in a new order -- would hash differently on every turn and re-bill the cached prefix each time. Only the host knows whether its content moved in a way that matters, so only the host can name the revision.

The corollary is a contract the host must honour: the same revision must mean the same content, within a scope. A host that returns a constant revision for content that changes will serve stale text from every cache keyed on it. This adapter cannot detect that -- it never sees a second call's content next to the first -- so it is stated here rather than implied.

Threading

A synchronous callable is run with :func:asyncio.to_thread rather than called inline. Host code behind this interface routinely does blocking I/O -- a DB query, a metadata HTTP call -- and running that on the event loop thread would stall every other in-flight turn in the process, a failure that shows up as unexplained tail latency rather than as an error anyone traces back here. An async def callable is awaited directly, on the loop, as its author intended.

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.

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 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 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 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 = 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.