`symfonic.core.plugins`¶

plugins ¶

Domain plugin system for symfonic agents.

Exposes the BaseDomainPlugin protocol, the PluginContribution dataclass (7.4+), and the PluginPromptSection adapter. External applications implement BaseDomainPlugin to inject domain-specific behavior (system-prompt content, guardrails, tools) without modifying the core library.

7.4+: plugins MAY ALSO define an optional inject_contributions method returning list[PluginContribution] to emit content at multiple cache positions (cached prefix / volatile suffix). See BaseDomainPlugin docstring and docs/guides/04-plugins-and-domains.md for the migration guide.

BaseDomainPlugin ¶

Bases: Protocol

Protocol for domain-specific agent behavior plugins.

Plugins provide three REQUIRED hooks:

inject_system_prompt: adds domain instructions to the system prompt (single string contribution).
validate_state_transition: guardrails that can block agent actions.
get_domain_tools: MUST return []. Domain tools are passed to the constructor via SymfonicAgent(tools=[...]) — the ToolRegistry freezes when the graph compiles in __init__, and load_plugin() runs after that, so a plugin cannot add tools. A non-empty return raises SymfonicAgentError (see get_domain_tools below). The hook is vestigial for tools and kept only for Protocol compatibility.

Plugins MAY ALSO define ONE OPTIONAL hook (added in 7.4):

inject_contributions: returns list[PluginContribution] so a plugin can emit content at multiple cache positions in one call ("cached" prefix vs. "volatile" suffix). If defined, this method REPLACES inject_system_prompt for prompt rendering — the legacy method is NOT called for plugins that implement the new API. Plugin authors migrating from inject_system_prompt can leave the legacy method as return "" or delete it (the Protocol still considers the class compliant as long as the name exists; an empty stub satisfies that).

Signature::

  async def inject_contributions(
      self,
      state: PromptState,
  ) -> list[PluginContribution]: ...

See symfonic.core.plugins.PluginContribution for the contribution shape and the design note at .claude/docs/2026-05-26-plugin-api-redesign-design.md.

Cache benefit on position="volatile" requires FrameworkConfig.prompt_layer_mode="stratigraphic". On the JIT path, volatile contributions are appended to the flat prompt body with no separate L2 region; the engine emits a one-time WARNING log per agent instance the first time this is observed.

inject_contributions is NOT part of the runtime-checkable Protocol surface — that would break isinstance(plugin, BaseDomainPlugin) for every legacy plugin. Dispatch happens via hasattr at the engine call site.

All hooks are called by the agent at appropriate lifecycle points. Implement this protocol to inject business-specific behavior without modifying the core library.

get_domain_tools ¶

get_domain_tools() -> list[Any]

Soft-deprecated (v8.6.6). MUST return [].

Domain tools are NOT registered through plugins. Pass them to the constructor instead::

SymfonicAgent(tools=[my_tool, ...])

The ToolRegistry freezes when the agent graph compiles inside SymfonicAgent.__init__. load_plugin() runs only after that point, so any tools returned here can no longer be registered: load_plugin() raises SymfonicAgentError on a non-empty return. There is no pre-compile plugin path that collects these tools, so a non-empty return is always an error.

This hook is retained only for plugin-Protocol compatibility (like the inject_system_prompt 7.4 soft-deprecation); it is not removed.

Returns:

Type	Description
`list[Any]`	Always `[]`.

Source code in src/symfonic/core/plugins/base.py

def get_domain_tools(self) -> list[Any]:
    """**Soft-deprecated (v8.6.6). MUST return ``[]``.**

    Domain tools are NOT registered through plugins. Pass them to the
    constructor instead::

        SymfonicAgent(tools=[my_tool, ...])

    The ``ToolRegistry`` freezes when the agent graph compiles inside
    ``SymfonicAgent.__init__``. ``load_plugin()`` runs only after that
    point, so any tools returned here can no longer be registered:
    ``load_plugin()`` raises ``SymfonicAgentError`` on a non-empty return.
    There is no pre-compile plugin path that collects these tools, so a
    non-empty return is *always* an error.

    This hook is retained only for plugin-Protocol compatibility (like the
    ``inject_system_prompt`` 7.4 soft-deprecation); it is not removed.

    Returns:
        Always ``[]``.
    """
    ...

inject_system_prompt `async` ¶

inject_system_prompt(state: PromptState) -> str

Return domain-specific system prompt content.

DEPRECATION NOTICE (7.4, soft): plugins SHOULD migrate to inject_contributions (see the class docstring) for multi-position support and cache hygiene. The legacy method remains supported indefinitely; no DeprecationWarning is emitted. Hard removal is deferred until adoption telemetry exists.

Called during HMS prompt composition. The returned text is appended as a section to the system prompt.

The state mapping is guaranteed to contain at least these stable keys on every render, across both the JIT and stratified prompt-assembly paths (v7.3.x+):

TENANT_ID    str  -- active tenant for this turn
NAMESPACE    str  -- tenant namespace (or sub-tenant id)
DOMAIN_NAME  str  -- FrameworkConfig domain identifier
QUERY        str  -- user query for this turn; empty string
                     when no query is available (never None,
                     never missing)

Additional context keys MAY be present and vary between the JIT and stratified strategies -- for example TOOL_MANIFEST, MEMORY_CONTEXT, SOUL_SCHEMA, CORE_PREFERENCES, ACTIVE_TOOLS, HYDRATED_CONTEXT, REQUIRED_LABELS, ONBOARDING_CHECKLIST, CURRENT_TIMESTAMP, PROSPECTIVE_TASKS, DOMAIN_DESCRIPTION. Plugins MUST use state.get(key, default) for any non-stable key.

QUERY was added to the contract in v7.3.x to unblock dynamic-retrieval plugins (vector search, RAG, query-conditioned HMS scoping); previously plugins had no race-free path to the user query (lifecycle hooks fire past prompt assembly, and plugin instances are shared across concurrent runs).

Parameters:

Name	Type	Description	Default
`state`	`PromptState`	Current prompt state. See the stable-key list above.	required

Returns:

Type	Description
`str`	Domain instructions string (empty string to skip).

Source code in src/symfonic/core/plugins/base.py

async def inject_system_prompt(self, state: PromptState) -> str:
    """Return domain-specific system prompt content.

    DEPRECATION NOTICE (7.4, soft): plugins SHOULD migrate to
    ``inject_contributions`` (see the class docstring) for
    multi-position support and cache hygiene. The legacy method
    remains supported indefinitely; no ``DeprecationWarning`` is
    emitted. Hard removal is deferred until adoption telemetry
    exists.

    Called during HMS prompt composition. The returned text is
    appended as a section to the system prompt.

    The ``state`` mapping is guaranteed to contain at least these
    stable keys on every render, across both the JIT and stratified
    prompt-assembly paths (v7.3.x+):

        TENANT_ID    str  -- active tenant for this turn
        NAMESPACE    str  -- tenant namespace (or sub-tenant id)
        DOMAIN_NAME  str  -- FrameworkConfig domain identifier
        QUERY        str  -- user query for this turn; empty string
                             when no query is available (never None,
                             never missing)

    Additional context keys MAY be present and vary between the JIT
    and stratified strategies -- for example ``TOOL_MANIFEST``,
    ``MEMORY_CONTEXT``, ``SOUL_SCHEMA``, ``CORE_PREFERENCES``,
    ``ACTIVE_TOOLS``, ``HYDRATED_CONTEXT``, ``REQUIRED_LABELS``,
    ``ONBOARDING_CHECKLIST``, ``CURRENT_TIMESTAMP``,
    ``PROSPECTIVE_TASKS``, ``DOMAIN_DESCRIPTION``. Plugins MUST use
    ``state.get(key, default)`` for any non-stable key.

    ``QUERY`` was added to the contract in v7.3.x to unblock
    dynamic-retrieval plugins (vector search, RAG, query-conditioned
    HMS scoping); previously plugins had no race-free path to the
    user query (lifecycle hooks fire past prompt assembly, and
    plugin instances are shared across concurrent runs).

    Args:
        state: Current prompt state. See the stable-key list above.

    Returns:
        Domain instructions string (empty string to skip).
    """
    ...

validate_state_transition `async` ¶

validate_state_transition(
    action: str, context: dict[str, Any]
) -> bool

Validate whether an agent action should be allowed.

Called before tool execution or state changes. Return False to block the action.

Parameters:

Name	Type	Description	Default
`action`	`str`	Name of the action/tool being attempted.	required
`context`	`dict[str, Any]`	Current execution context (tenant, params, etc.)	required

Returns:

Type	Description
`bool`	True to allow, False to block.

Source code in src/symfonic/core/plugins/base.py

async def validate_state_transition(
    self,
    action: str,
    context: dict[str, Any],
) -> bool:
    """Validate whether an agent action should be allowed.

    Called before tool execution or state changes. Return False
    to block the action.

    Args:
        action: Name of the action/tool being attempted.
        context: Current execution context (tenant, params, etc.)

    Returns:
        True to allow, False to block.
    """
    ...

PluginContribution `dataclass` ¶

PluginContribution(
    content: str,
    position: Position = "cached",
    order: int = 100,
    header: str | None = None,
    origin: str | None = None,
    cache_ttl: CacheTtl | None = None,
)

A single section emitted by a plugin into a cache position.

Attributes:

Name	Type	Description
`content`	`str`	Rendered text. Empty string skips the contribution silently (no header, no separator).
`position`	`Position`	`"cached"` (L1 / cached prefix, the default), `"volatile"` (L2 / per-turn uncached suffix), or `"kernel"` (RESERVED in 7.4 — accepted-but-no-op; the dispatcher remaps to `"cached"` with a one-time WARNING).
`order`	`int`	Sort key within a position. LOWER values render first. Default `100`. Negative values are allowed; use them when a plugin needs to render before the default-ordered plugins.
`header`	`str \| None`	`None` (default) — engine auto-injects `### {plugin.name}` on the first contribution per `(plugin, position)` group. `""` — no header at all (plugin owns its scoping). Any non-empty string — used verbatim as the header on this contribution.
`origin`	`str \| None`	Engine-populated diagnostic field carrying the plugin name. Plugins MUST NOT set this; the dispatcher fills it in during dispatch so the auto-header logic and observability tooling can attribute contributions to plugins.

Raises:

Type	Description
`ValueError`	If `position` is not one of `"cached"` / `"volatile"` / `"kernel"`.
`TypeError`	If `content` is not a `str`.

with_origin ¶

with_origin(origin: str) -> PluginContribution

Return a copy carrying the supplied origin (plugin name).

Source code in src/symfonic/core/plugins/contribution.py

def with_origin(self, origin: str) -> PluginContribution:
    """Return a copy carrying the supplied ``origin`` (plugin name)."""
    return replace(self, origin=origin)

PluginPromptSection `dataclass` ¶

PluginPromptSection(
    name: str = "Domain Plugin",
    required: bool = False,
    priority: int = 15,
    cache_breakpoint: bool = True,
    cache_ttl: Literal["5m", "1h"] | None = None,
    _last_contribs: list[PluginContribution] = list(),
    _plugins: list[BaseDomainPlugin] = list(),
    _short_circuit_warned: set[str] = set(),
    _kernel_warned: set[str] = set(),
)

PromptSection that renders content from all loaded domain plugins.

Satisfies the PromptSection protocol so it can be added to a PromptBuilder pipeline or rendered directly from the agent.

Attributes:

Name	Type	Description
`name`	`str`	Section label shown in the composed prompt.
`required`	`bool`	Whether a render failure is fatal (always False here).
`priority`	`int`	Render order — 15 places it after core system context but before memory sections.
`cache_breakpoint`	`bool`	Marks the section for Anthropic prompt caching. Domain rules are stable within a session so caching is safe.

add_plugin ¶

add_plugin(plugin: BaseDomainPlugin) -> None

Register a plugin to contribute its prompt content.

Parameters:

Name	Type	Description	Default
`plugin`	`BaseDomainPlugin`	Any object satisfying BaseDomainPlugin protocol.	required

Source code in src/symfonic/core/plugins/section.py

def add_plugin(self, plugin: BaseDomainPlugin) -> None:
    """Register a plugin to contribute its prompt content.

    Args:
        plugin: Any object satisfying BaseDomainPlugin protocol.
    """
    self._plugins.append(plugin)

render `async` ¶

render(state: PromptState) -> str

Render all plugin contributions into a single string.

Each plugin's output is prefixed with a ### <name> heading. Plugins that return empty strings are silently skipped. Plugins that raise exceptions are logged at DEBUG level and skipped.

7.4+: implemented on top of :meth:render_contributions so the legacy single-string surface and the new contribution surface agree byte-for-byte for plugins that only define inject_system_prompt.

Parameters:

Name	Type	Description	Default
`state`	`PromptState`	Current prompt state forwarded to each plugin.	required

Returns:

Type	Description
`str`	Combined plugin content, or empty string if no plugins produced
`str`	non-empty output.

Source code in src/symfonic/core/plugins/section.py

async def render(self, state: PromptState) -> str:
    """Render all plugin contributions into a single string.

    Each plugin's output is prefixed with a ``### <name>`` heading.
    Plugins that return empty strings are silently skipped.
    Plugins that raise exceptions are logged at DEBUG level and skipped.

    7.4+: implemented on top of :meth:`render_contributions` so the
    legacy single-string surface and the new contribution surface
    agree byte-for-byte for plugins that only define
    ``inject_system_prompt``.

    Args:
        state: Current prompt state forwarded to each plugin.

    Returns:
        Combined plugin content, or empty string if no plugins produced
        non-empty output.
    """
    contribs = await self.render_contributions(state)
    # ``render_contributions`` already cached the list on
    # ``_last_contribs``; no extra bookkeeping here.
    # The legacy ``render`` surface only ever produced "cached"
    # content (it pre-dates the cache-position concept). New API
    # plugins emitting volatile contributions still flow through
    # here because external callers reach for ``render`` — but the
    # engine routes through ``render_contributions`` directly, so
    # external callers seeing a flat string see EVERY position
    # joined together. The legacy ordering ("### {name}\n{body}" per
    # plugin, joined with double-newline) is preserved by
    # ``render_position_group`` for ``header=None`` synthetic
    # contributions.
    cached = render_position_group(contribs, "cached") or ""
    volatile = render_position_group(contribs, "volatile") or ""
    parts = [p for p in (cached, volatile) if p]
    return "\n\n".join(parts)

render_contributions `async` ¶

render_contributions(
    state: PromptState,
) -> list[PluginContribution]

Collect every plugin's contributions into a flat ordered list.

Wraps :func:dispatch_plugin_contributions and threads the section's per-instance latches so observability logs (legacy short-circuit INFO, kernel-position WARNING) fire at most once per (plugin, section-instance) pair.

Parameters:

Name	Type	Description	Default
`state`	`PromptState`	PromptState forwarded to each plugin.	required

Returns:

Type	Description
`list[PluginContribution]`	Ordered list of `PluginContribution` objects with
`list[PluginContribution]`	`origin` populated and `position` normalized
`list[PluginContribution]`	(`"kernel"` remapped to `"cached"`).

Source code in src/symfonic/core/plugins/section.py

async def render_contributions(
    self,
    state: PromptState,
) -> list[PluginContribution]:
    """Collect every plugin's contributions into a flat ordered list.

    Wraps :func:`dispatch_plugin_contributions` and threads the
    section's per-instance latches so observability logs (legacy
    short-circuit INFO, kernel-position WARNING) fire at most once
    per ``(plugin, section-instance)`` pair.

    Args:
        state: PromptState forwarded to each plugin.

    Returns:
        Ordered list of ``PluginContribution`` objects with
        ``origin`` populated and ``position`` normalized
        (``"kernel"`` remapped to ``"cached"``).
    """
    contribs = await dispatch_plugin_contributions(
        self._plugins,
        state,
        short_circuit_warned=self._short_circuit_warned,
        kernel_warned=self._kernel_warned,
    )
    # v7.24.0: stash for the cache-TTL bubble-up rule.  The builder
    # consults ``resolve_cache_ttl`` AFTER ``render(state)`` returns;
    # by then this list reflects the contributions that actually
    # produced the rendered text.
    self._last_contribs = list(contribs)
    return contribs

resolve_cache_ttl ¶

resolve_cache_ttl() -> Literal['5m', '1h'] | None

Compute the effective cache TTL hint for this section.

v7.24.0 precedence rule (architect-fixed):

Section-level :attr:cache_ttl wins when set. One section maps to one wire cache breakpoint, so the section's value is authoritative for the marker on its terminal block.
When section-level is None, the strongest contribution- level TTL from the LAST rendered contribution list bubbles up: "1h" > "5m" > None. Lets adopters declare TTL intent at the contribution granularity without splitting a section.

Callers MUST invoke this AFTER render(state) / render_contributions(state) has populated _last_contribs; otherwise the bubble-up step sees no contributions and falls back to None. This is the contract :class:PromptBuilder relies on.

Source code in src/symfonic/core/plugins/section.py

def resolve_cache_ttl(self) -> Literal["5m", "1h"] | None:
    """Compute the effective cache TTL hint for this section.

    v7.24.0 precedence rule (architect-fixed):

    * Section-level :attr:`cache_ttl` wins when set.  One section
      maps to one wire cache breakpoint, so the section's value is
      authoritative for the marker on its terminal block.
    * When section-level is ``None``, the strongest contribution-
      level TTL from the LAST rendered contribution list bubbles up:
      ``"1h"`` > ``"5m"`` > ``None``.  Lets adopters declare TTL
      intent at the contribution granularity without splitting a
      section.

    Callers MUST invoke this AFTER ``render(state)`` /
    ``render_contributions(state)`` has populated ``_last_contribs``;
    otherwise the bubble-up step sees no contributions and falls
    back to ``None``.  This is the contract :class:`PromptBuilder`
    relies on.
    """
    if self.cache_ttl is not None:
        return self.cache_ttl
    # Bubble-up: 1h beats 5m beats None.
    has_1h = False
    has_5m = False
    for c in self._last_contribs:
        if c.cache_ttl == "1h":
            has_1h = True
            break
        if c.cache_ttl == "5m":
            has_5m = True
    if has_1h:
        return "1h"
    if has_5m:
        return "5m"
    return None

symfonic.core.plugins¶

plugins ¶

BaseDomainPlugin ¶

get_domain_tools ¶

inject_system_prompt async ¶

validate_state_transition async ¶

PluginContribution dataclass ¶

with_origin ¶

PluginPromptSection dataclass ¶

add_plugin ¶

render async ¶

render_contributions async ¶

resolve_cache_ttl ¶

`symfonic.core.plugins`¶

inject_system_prompt `async` ¶

validate_state_transition `async` ¶

PluginContribution `dataclass` ¶

PluginPromptSection `dataclass` ¶

render `async` ¶

render_contributions `async` ¶