symfonic.agent.prompts.stratigraphic

stratigraphic

Stratigraphic prompt assembly for prompt_layer_mode='stratigraphic'.

Phase B Commit 4 / 7.0.13 cache-regression fix. Wraps the existing legacy prompt builder output with the L0 Brain Kernel template so the Anthropic provider sees an explicit cache_control breakpoint on a contiguous, session-stable prefix.

Three-layer model:

• L0 Brain Kernel -- brain_v1.txt / brain_v1_jit.txt. 100% static, loaded once per process via importlib.resources. On its own L0 is far below Anthropic's ~1024-token minimum cacheable prefix, so it is NOT given its own breakpoint -- it is the head of the cached region, not a region of its own.
• L1 OS Services -- the session-stable body: domain directives, SOUL schema, required labels, onboarding checklist, tool manifest, plugin contributions. NO timestamp, NO memory context, NO hydrated context, NO active tools, NO prospective tasks.

The onboarding checklist arrives here as part of the extraction directive (extraction_part is spliced into the L1 body), where it lists categories to RECORD when the user volunteers them. It is not a behavioural instruction: nothing tells the agent to ask for those fields, and core has no completion signal that would tell it to stop. The single cache_control breakpoint is placed at the END of L1, so the cached prefix is exactly L0 + L1 -- one contiguous region that clears the 1024-token minimum and contains zero per-turn content. * L2 Task -- per-turn volatile content (timestamp, memory context, hydrated/RAG context, active tools, prospective tasks, extraction section). Appended AFTER the breakpoint and left uncached, so it never forces a cache_creation write priced at 1.25x.

Why one breakpoint and not two: Anthropic caches a prefix. The breakpoint marks the end of the cacheable region; everything before it (across every block) is the cached prefix. Marking L0 separately would either be silently dropped (L0 is sub-minimum) or be redundant with the L1 breakpoint. A single breakpoint at the end of L1 is the correct and cleaner expression of "cache L0+L1, leave L2 hot".

Returned by :func:build_stratigraphic_prompt:

• prompt -- a JSON-encoded list of Anthropic content blocks ready to hand to SystemMessage(content=<list>) after a json.loads. The consolidator in react.py detects this shape and forwards it as a list rather than a flattened string.
• brain_prompt_version -- the BRAIN_PROMPT_VERSION constant string, written into state['brain_prompt_version'] for telemetry.
• system_prompt_hash -- the first 16 hex chars of the SHA-256 of the L1 body. L1 is now session-stable, so warm turns with the same domain + manifest produce a stable hash and Anthropic cache hits can be correlated against it.

StratigraphicPrompt dataclass

StratigraphicPrompt(prompt: str, brain_prompt_version: str, system_prompt_hash: str)

Result of a stratigraphic build.

Attributes:

Name	Type	Description
prompt	str	JSON-encoded list of Anthropic content blocks. Each block is {"type": "text", "text": ...} and the L1 block also carries cache_control={"type": "ephemeral"}. Decodes back to list[dict] via json.loads for downstream consumption (see react._consolidate_messages).
brain_prompt_version	str	Kernel version tag, e.g. "1".
system_prompt_hash	str	16-hex-char SHA-256 prefix of the L1 body.

build_stratigraphic_prompt async

build_stratigraphic_prompt(*, l1_body: str, jit: bool, l2_body: str | None = None, force_last_cache: bool = False, system_prefix_cache_ttl: Literal['5m', '1h'] = '5m') -> StratigraphicPrompt

Assemble L0 (Kernel) + L1 (stable) + optional L2 (per-turn).

The cached prefix is exactly L0 + L1: a single cache_control breakpoint is placed at the end of L1. L2, when supplied, is appended after the breakpoint and left uncached so its per-turn volatility never triggers a 1.25x cache_creation write.

Parameters:

Name	Type	Description	Default
l1_body	str	The session-stable system body -- domain directives, SOUL schema, required labels, onboarding checklist, tool manifest, plugin contributions. MUST NOT contain per-turn content (timestamp, memory context, hydrated context, active tools, prospective tasks); those belong in l2_body.	required
jit	bool	Whether to load the JIT-mode kernel (brain_v1_jit.txt) or the full kernel (brain_v1.txt).	required
l2_body	str | None	The per-turn volatile body. When None or empty no L2 block is emitted and the result is just L0 + L1 with the breakpoint on L1.	None
force_last_cache	bool	Forwarded to PromptBuilder.build_cached. Defaults to False -- the trailing block (L2 when present, else L1) must not auto-promote to a breakpoint on position alone. L1 carries an explicit cache_breakpoint=True so it is cached whether or not L2 follows it.	False

Returns:

Type	Description
StratigraphicPrompt	class:StratigraphicPrompt with the JSON-encoded content
StratigraphicPrompt	blocks, the kernel version tag, and the L1 body hash.

Source code in symfonic/agent/prompts/stratigraphic.py

async def build_stratigraphic_prompt(
    *,
    l1_body: str,
    jit: bool,
    l2_body: str | None = None,
    force_last_cache: bool = False,
    system_prefix_cache_ttl: Literal["5m", "1h"] = "5m",
) -> StratigraphicPrompt:
    """Assemble L0 (Kernel) + L1 (stable) + optional L2 (per-turn).

    The cached prefix is exactly ``L0 + L1``: a single ``cache_control``
    breakpoint is placed at the end of L1.  L2, when supplied, is
    appended after the breakpoint and left uncached so its per-turn
    volatility never triggers a 1.25x ``cache_creation`` write.

    Args:
        l1_body: The session-stable system body -- domain directives,
            SOUL schema, required labels, onboarding checklist, tool
            manifest, plugin contributions.  MUST NOT contain per-turn
            content (timestamp, memory context, hydrated context, active
            tools, prospective tasks); those belong in ``l2_body``.
        jit: Whether to load the JIT-mode kernel
            (``brain_v1_jit.txt``) or the full kernel (``brain_v1.txt``).
        l2_body: The per-turn volatile body.  When ``None`` or empty no
            L2 block is emitted and the result is just ``L0 + L1`` with
            the breakpoint on L1.
        force_last_cache: Forwarded to ``PromptBuilder.build_cached``.
            Defaults to ``False`` -- the trailing block (L2 when present,
            else L1) must not auto-promote to a breakpoint on position
            alone.  L1 carries an explicit ``cache_breakpoint=True`` so
            it is cached whether or not L2 follows it.

    Returns:
        :class:`StratigraphicPrompt` with the JSON-encoded content
        blocks, the kernel version tag, and the L1 body hash.
    """
    kernel = _load_kernel(jit)

    builder = PromptBuilder()
    # L0: invariant kernel.  No cache_breakpoint -- on its own L0 is
    # below Anthropic's ~1024-token minimum cacheable prefix, so a
    # breakpoint here would be silently dropped.  L0 is the HEAD of the
    # cached region, not a region of its own.  ``build_cached`` still
    # marks the first block with cache_control, but the breakpoint that
    # actually defines the cached prefix is the one on L1 below.
    builder.add_section(
        StaticSection(
            name="L0 Brain Kernel",
            priority=10,
            content=kernel,
            cache_breakpoint=False,
        ),
    )
    # L1: session-stable services body.  cache_breakpoint=True places
    # the single real breakpoint at the END of L1, so the cached prefix
    # is exactly L0 + L1 -- one contiguous region that clears the
    # 1024-token minimum and contains zero per-turn content.
    #
    # v7.24.0 §5: ``system_prefix_cache_ttl`` flows from FrameworkConfig
    # into this StaticSection.  ``"5m"`` (default) keeps the wire
    # marker byte-identical to v7.23 (``cache_ttl=None`` on the
    # CacheBlock collapses to the bare ephemeral marker).  ``"1h"``
    # flips it to ``{"type":"ephemeral","ttl":"1h"}`` so the L1 prefix
    # survives idle gaps up to an hour at the cost of a 2x cache_write
    # premium on cold turns.  See
    # ``docs/concepts/cache-tier-tradeoffs.md``.
    _l1_cache_ttl: Literal["5m", "1h"] | None = (
        "1h" if system_prefix_cache_ttl == "1h" else None
    )
    builder.add_section(
        StaticSection(
            name="L1 OS Services",
            priority=20,
            content=l1_body,
            cache_breakpoint=True,
            cache_ttl=_l1_cache_ttl,
        ),
    )
    # L2: per-turn volatile body.  cache_breakpoint=False and
    # force_last_cache=False keep it uncached so its per-turn drift
    # never forces a cache_creation write.
    if l2_body:
        builder.add_section(
            StaticSection(
                name="L2 Task",
                priority=30,
                content=l2_body,
                cache_breakpoint=False,
            ),
        )

    result = await builder.build_cached({}, force_last_cache=force_last_cache)
    return StratigraphicPrompt(
        prompt=result.content,
        brain_prompt_version=BRAIN_PROMPT_VERSION,
        system_prompt_hash=_hash_body(l1_body),
    )

strip_cache_control

strip_cache_control(prompt: str) -> str | None

Return prompt's block list with every cache_control removed.

The mechanism half of T3.1.4 finding 1; the engine owns the policy half (which families may be annotated -- see SymfonicAgent._stratigraphic_annotates_cache). Split that way because "what a block looks like" belongs to whoever builds one, and "whose wire can be described" belongs to whoever knows the provider.

The envelope is preserved: the caller still receives a JSON list of {"type": "text", "text": ...} blocks in the same order, minus the annotation. Flattening instead would make attachments the odd region out, since _to_block already encodes those Anthropic-shaped for unknown.

Returns None when prompt is not a decodable block list, so a caller can degrade deliberately rather than ship markers it meant to remove.

Source code in symfonic/agent/prompts/stratigraphic.py

def strip_cache_control(prompt: str) -> str | None:
    """Return ``prompt``'s block list with every ``cache_control`` removed.

    The mechanism half of T3.1.4 finding 1; the engine owns the policy half
    (which families may be annotated -- see
    ``SymfonicAgent._stratigraphic_annotates_cache``). Split that way because
    "what a block looks like" belongs to whoever builds one, and "whose wire
    can be described" belongs to whoever knows the provider.

    The envelope is preserved: the caller still receives a JSON list of
    ``{"type": "text", "text": ...}`` blocks in the same order, minus the
    annotation. Flattening instead would make attachments the odd region out,
    since ``_to_block`` already encodes those Anthropic-shaped for ``unknown``.

    Returns ``None`` when ``prompt`` is not a decodable block list, so a caller
    can degrade deliberately rather than ship markers it meant to remove.
    """
    try:
        blocks = json.loads(prompt)
        if not isinstance(blocks, list):
            return None
        stripped = [
            {key: value for key, value in block.items() if key != "cache_control"}
            if isinstance(block, dict)
            else block
            for block in blocks
        ]
        return json.dumps(stripped)
    except Exception:
        logger.warning(
            "Stratigraphic cache-marker strip failed; the caller degrades "
            "rather than emitting an annotation this provider cannot describe",
            exc_info=True,
        )
        return None