block_boundaries¶

Level 2 · Standing context — where a prompt block stops.

prompt_blocks shows what a block is: the tiers, the source adapters, where the rendered region lands. This one shows the three boundaries, which are the parts an adopter is most likely to get wrong because none of them are visible in a happy-path render:

what a delegated child inherits — and what it must not
what happens when a source fails, for each of the three policies
how a block stops rendering once its job is done
Prerequisites: none — static sources, a deliberately broken source, and a host predicate. No model, no network, no database.
Key concepts: capture_block_snapshot, inherit, on_source_failure, last_known_good, render_when, BlockGateError

Run it¶

pip install "symfonic-core[cli]"
symfonic examples add block_boundaries
python -m block_boundaries

Or, from a source checkout (run from the repo root):

python -m examples.block_boundaries

1. What a delegated child inherits¶

parent resolves : ['IDENTITY', 'ONBOARDING']
child snapshot  : ['IDENTITY']

ONBOARDING in the parent's rendered region? True
ONBOARDING in the child's snapshot?         False

ONBOARDING declares inherit=False in the canonical matrix, so the operator does not have to remember. A subagent spawned mid-conversation would otherwise start onboarding a user who is already onboarded — using the parent's own words, which makes it read as intentional.

Where the filter runs matters. capture_block_snapshot() applies it itself, so the snapshot is the child's view. There is no second step a caller could forget, and no way to hand a child the parent's unfiltered blocks by calling the wrong function.

2. What happens when a source fails¶

Same wedged source, three declared policies:

fail_closed       -> BlockResolutionError -- the turn does not run
omit              -> omitted -- the turn continues without this block
last_known_good   -> BlockResolutionError -- the turn does not run

That third line looks wrong and is not. last_known_good serves a revision this process loaded successfully during its own lifetime — it is not a disk cache and never survives a restart. With no good load on record it degrades to fail_closed rather than inventing an empty block. Give it one good load and it holds the line:

healthy source     -> served 1 block(s)
then it breaks     -> served 1 block(s), from_last_known_good=True

The defaults are per tier, not per adapter, because the cost of a missing block differs by what the block is:

BOUNDARIES    platform   -> fail_closed
RULES         operating  -> fail_closed
USER_PROFILE  profile    -> omit
FOCUS         session    -> omit

An agent whose BOUNDARIES block vanished is an agent running without its constraints — worse than an agent that did not run. A missing USER_PROFILE costs personalisation for one turn.

3. How a block stops rendering¶

nothing known yet        -> ['IDENTITY', 'ONBOARDING']
name recorded            -> ['IDENTITY', 'ONBOARDING']
name and role recorded   -> ['IDENTITY']

The block does not render empty; it is absent. The agent stops being told to onboard because the instruction is no longer in the prompt at all, not because it was asked to ignore one.

The gate is checked before the source is read, so a gated-off block costs no I/O — which is why this beats the older trick of a computed source returning an empty string.

And a predicate that raises fails the turn, even on a block declaring on_source_failure="omit":

a raising predicate fails the turn: BlockGateError

BlockGateError is deliberately not a StorageError, so the failure policy cannot absorb it. on_source_failure exists for a source being unavailable — a wedged mount, an outage, something that may clear. A predicate is in-process host code: if it raises once it raises every turn, and an omit-policy block would vanish from every prompt, silently, for as long as the bug lived.

What none of this includes¶

No block-edit tool is registered in the agent's palette on any construction path in this stage. For a delegated child that is enforced rather than merely true: the self-edit request is stripped from every child config the parent builds, and a pre-built child arriving with a block-edit tool is rejected.

Full code¶

"""The three boundaries of a prompt block: delegation, failure, and lifetime.

``examples/prompt_blocks`` shows what a block IS -- the tiers, the source
adapters, where the rendered region lands. This one shows where a block
STOPS, which is the part an adopter is most likely to get wrong because
none of it is visible in a happy-path render:

  1. what a delegated child inherits -- and what it must not
  2. what happens when a source fails, for each of the three policies
  3. how a block stops rendering once its job is done

Runs with no infrastructure: static sources, a deliberately broken
source, and a host predicate. No model, no network, no database.

Usage::

    symfonic examples add block_boundaries
    python -m block_boundaries

    # or, from a checkout
    python -m examples.block_boundaries
"""

from __future__ import annotations

import asyncio

from symfonic.core.prompt.blocks import PromptBlockResolver, PromptBlockSpec
from symfonic.core.prompt.blocks.render import render_blocks
from symfonic.core.prompt.blocks.resolver import BlockGateError, BlockResolutionError
from symfonic.core.prompt.blocks.snapshot import capture_block_snapshot
from symfonic.core.prompt.blocks.sources import StaticBlockSource
from symfonic.core.prompt.blocks.taxonomy import CANONICAL_BLOCKS
from symfonic.core.protocols import StorageError
from symfonic.core.scope import TenantScope

SCOPE = TenantScope.root("tenant", "acme")

IDENTITY = "You are Ada, the operations assistant for Acme Widgets."
ONBOARDING = "This user is new. Ask for their name, then what they work on."


def rule(title: str) -> None:
    print(f"\n{'─' * 66}\n{title}\n{'─' * 66}")


class WedgedSource:
    """A source whose backing store is unavailable.

    Raises ``StorageError`` -- the family the resolver narrows its catch
    to, so it reaches ``on_source_failure`` rather than being mistaken
    for a programming error.
    """

    offline_safe = True
    scope_aware = False

    async def load(self, scope: TenantScope, block_id: str) -> object:
        raise StorageError("mount is wedged")


class SometimesSource(StaticBlockSource):
    """Serves content, then starts failing. For ``last_known_good``."""

    def __init__(self, content: str) -> None:
        super().__init__(content)
        self.healthy = True

    async def load(self, scope: TenantScope, block_id: str) -> object:
        if not self.healthy:
            raise StorageError("upstream went away")
        return await super().load(scope, block_id)


async def section_delegation() -> None:
    rule("1. What a delegated child inherits -- and what it must not")

    specs = (
        PromptBlockSpec(name="IDENTITY", source=StaticBlockSource(IDENTITY)),
        PromptBlockSpec(name="ONBOARDING", source=StaticBlockSource(ONBOARDING)),
    )
    resolver = PromptBlockResolver(specs)

    parent = await resolver.resolve(SCOPE)
    snapshot = await capture_block_snapshot(resolver, SCOPE)

    print(f"  parent resolves : {[b.spec.name for b in parent]}")
    print(f"  child snapshot  : {list(snapshot.block_names)}")
    print()
    print(f"  ONBOARDING in the parent's rendered region? "
          f"{'ONBOARDING' in (render_blocks(parent) or '')}")
    print(f"  ONBOARDING in the child's snapshot?         "
          f"{'ONBOARDING' in (snapshot.l1 or '')}")

    canonical = CANONICAL_BLOCKS["ONBOARDING"]
    print(f"\n  ONBOARDING declares inherit={canonical.inherit} in the canonical")
    print("  matrix, so the operator does not have to remember. A subagent")
    print("  spawned mid-conversation would otherwise start onboarding a user")
    print("  who is already onboarded, using the parent's own words.")

    print("\n  Note WHERE the filter runs: capture_block_snapshot() applies it")
    print("  itself, so the snapshot IS the child's view. There is no second")
    print("  step a caller could forget -- and no way to hand a child the")
    print("  parent's unfiltered blocks by calling the wrong function.")

    print("\n  the snapshot is a point-in-time copy, keyed for cache reuse:")
    # scope_path joins segments with \x1f, which prints invisibly.
    print(f"    scope_path   {snapshot.scope_path!r}")
    print(f"    revision_key {snapshot.revision_key[:2]}...")


async def _resolve_one(spec: PromptBlockSpec) -> str:
    try:
        resolved = await PromptBlockResolver((spec,)).resolve(SCOPE)
        if not resolved:
            return "omitted -- the turn continues without this block"
        return f"served {len(resolved)} block(s)"
    except BlockResolutionError:
        return "BlockResolutionError -- the turn does not run"


async def section_failure() -> None:
    rule("2. What happens when a source fails")

    print("  Same wedged source, three declared policies:\n")
    for policy in ("fail_closed", "omit", "last_known_good"):
        spec = PromptBlockSpec(
            name="RULES", source=WedgedSource(), on_source_failure=policy
        )
        print(f"    {policy:17} -> {await _resolve_one(spec)}")

    print("\n  last_known_good FAILED above, and that is correct: this process")
    print("  had never loaded the block successfully, so there is no known-good")
    print("  revision to serve. It degrades to fail_closed rather than inventing")
    print("  an empty block. Give it one good load first and it holds the line:\n")

    flaky = SometimesSource("Never move money without human approval.")
    spec = PromptBlockSpec(
        name="RULES", source=flaky, on_source_failure="last_known_good"
    )
    resolver = PromptBlockResolver((spec,))
    first = await resolver.resolve(SCOPE)
    print(f"    healthy source     -> served {len(first)} block(s)")
    flaky.healthy = False
    after = await resolver.resolve(SCOPE)
    served = after[0] if after else None
    print(f"    then it breaks     -> served {len(after)} block(s), "
          f"from_last_known_good={getattr(served, 'from_last_known_good', None)}")

    print("\n  The defaults are per TIER, not per adapter, because the cost of")
    print("  a missing block differs by what the block is:")
    for name in ("BOUNDARIES", "RULES", "USER_PROFILE", "FOCUS"):
        row = CANONICAL_BLOCKS[name]
        spec = PromptBlockSpec(
            name=name,
            source="memory" if row.tier in ("profile", "session") else
                   StaticBlockSource("x"),
        )
        print(f"    {name:13} {row.tier:10} -> {spec.on_source_failure}")
    print("\n  An agent whose BOUNDARIES block vanished is an agent running")
    print("  without its constraints, which is worse than an agent that did")
    print("  not run. A missing USER_PROFILE costs personalisation for a turn.")


class OnboardingDone:
    """Host-owned completion state -- the ``render_when`` predicate.

    Core cannot compute this. ``onboarding_checklist`` is prose,
    ``soul_schema`` is a type map, and neither can represent "the user
    declined to say". Only the host knows when the question is over.
    """

    def __init__(self) -> None:
        self.known: set[str] = set()

    def should_render(self, scope: TenantScope, block_id: str) -> bool:
        return not {"name", "role"} <= self.known


async def section_lifetime() -> None:
    rule("3. How a block stops rendering")

    state = OnboardingDone()
    specs = (
        PromptBlockSpec(name="IDENTITY", source=StaticBlockSource(IDENTITY)),
        PromptBlockSpec(
            name="ONBOARDING",
            source=StaticBlockSource(ONBOARDING),
            render_when=state.should_render,
        ),
    )
    resolver = PromptBlockResolver(specs)

    for label, known in (
        ("nothing known yet", set()),
        ("name recorded", {"name"}),
        ("name and role recorded", {"name", "role"}),
    ):
        state.known = set(known)
        blocks = await resolver.resolve(SCOPE)
        names = [b.spec.name for b in blocks]
        print(f"  {label:24} -> {names}")

    print("\n  The block does not render empty; it is ABSENT. The agent stops")
    print("  being told to onboard because the instruction is no longer in the")
    print("  prompt at all, not because it was asked to ignore one.")

    print("\n  The gate is checked BEFORE the source is read, so a gated-off")
    print("  block costs no I/O -- which is why this beats the older trick of")
    print("  a computed source returning an empty string.")

    def broken_gate(scope: TenantScope, block_id: str) -> bool:
        raise RuntimeError("host state store is down")

    spec = PromptBlockSpec(
        name="IDENTITY",
        source=StaticBlockSource(IDENTITY),
        on_source_failure="omit",
        render_when=broken_gate,
    )
    try:
        await PromptBlockResolver((spec,)).resolve(SCOPE)
        print("\n  a raising gate was absorbed -- THIS WOULD BE A BUG")
    except BlockGateError as exc:
        print(f"\n  a raising predicate fails the turn: {type(exc).__name__}")
        print("  even though this block declares on_source_failure='omit'.")
        print("  A gate error is deliberately NOT a StorageError, so the")
        print("  failure policy cannot absorb it: on_source_failure exists for")
        print("  a source being unavailable, and an omit-policy block would")
        print("  otherwise vanish from every prompt, silently, for as long as")
        print("  the host's bug lived.")


async def main() -> None:
    print("Framework WARNING lines below are part of the demonstration: the")
    print("resolver logs when it omits a block or serves a stale one. They")
    print("arrive on stderr, so they may appear before the first section.")
    await section_delegation()
    await section_failure()
    await section_lifetime()

    rule("What none of this includes")
    print("  No block-edit tool is registered in the agent's palette on any")
    print("  construction path in this stage. For a delegated child that is")
    print("  enforced rather than merely true: the self-edit request is")
    print("  stripped from every child config the parent builds, and a")
    print("  pre-built child arriving with a block-edit tool is rejected.")
    print()
    print("  So the boundaries above are the whole story. A block changes")
    print("  when the operator's own system changes it -- a git commit and a")
    print("  deploy, a CMS, an admin UI -- never because the model asked.")


if __name__ == "__main__":
    asyncio.run(main())