symfonic.agent.prompts.ports

ports

The declared port of the prompt-assembly capability.

This module exists because the matrix is silent here, not because it is permissive. symfonic.agent.prompts is ruled capability (T1.2.4 archcheck.toml) and so is symfonic.capabilities.prompting, and check_boundaries returns nothing for a same-layer edge — "same-layer structure is governed by package-level rules, not the matrix" (:mod:symfonic.devtools.archcheck.boundaries). So the one cross-package import, symfonic.capabilities.prompting.hms reaching :class:~symfonic.agent.prompts.system_prompt.HMSSystemPromptSection, is not licensed by a cell; it is out of the checker's reach entirely.

T1.2.1's dependency-matrix.md says what the self cell does not mean: "capability → capability cross-talk is not an import edge at all: collaboration goes through ports declared to the compiler and wired via the plan (LAY-ADR §3.4). The self cell does not license importing another capability package's internals." system_prompt is an internal module of a different capability package, so that note excludes exactly the edge above.

The package-level rule the matrix defers to is therefore written here and enforced by test_the_prompting_capability_reaches_agent_prompts_only_through_a_port: importers outside symfonic.agent name this module and nothing else under symfonic.agent.prompts. Nothing is defined locally; each name is re-exported from its owning module, so there is still one implementation.

HMSSystemPromptSection dataclass

HMSSystemPromptSection(name: str = 'HMS System Prompt', required: bool = False, priority: int = 5, cache_breakpoint: bool = True, template_path: Path | None = None)

PromptSection that renders the HMS-aware system prompt template.

Substitutes all {{VARIABLE}} placeholders from the PromptState context. Raises ValueError if any placeholder remains unresolved after rendering.

Attributes:

Name	Type	Description
name	str	Section display name used by PromptBuilder.
required	bool	Whether a render failure should abort the build.
priority	int	Lower numbers appear earlier in the assembled prompt.
template_path	Path | None	Optional override path (file or directory) for the hms_system.txt template. Defaults to the bundled template.

estimate_tokens staticmethod

estimate_tokens(text: str) -> int

Estimate token count using a conservative 3-chars-per-token heuristic.

The len // 4 heuristic undercounts by 20-30 % on structured / markdown text. len // 3 is closer to observed tokeniser output for prompts that contain headings, bullet lists, and JSON blocks.

Parameters:

Name	Type	Description	Default
text	str	The text to estimate token count for.	required

Returns:

Type	Description
int	Estimated number of tokens (conservative upper bound).

Source code in src/symfonic/agent/prompts/system_prompt.py

@staticmethod
def estimate_tokens(text: str) -> int:
    """Estimate token count using a conservative 3-chars-per-token heuristic.

    The ``len // 4`` heuristic undercounts by 20-30 % on structured /
    markdown text.  ``len // 3`` is closer to observed tokeniser output
    for prompts that contain headings, bullet lists, and JSON blocks.

    Args:
        text: The text to estimate token count for.

    Returns:
        Estimated number of tokens (conservative upper bound).
    """
    return len(text) // 3

render async

render(state: PromptState) -> str

Render the HMS system prompt by substituting placeholders.

Parameters:

Name	Type	Description	Default
state	PromptState	Mapping that must contain values for every {{VARIABLE}} placeholder in the template.	required

Returns:

Type	Description
str	The fully substituted prompt text.

Raises:

Type	Description
SecurityScopeError	If TENANT_ID is missing or empty.
ValueError	If any {{VARIABLE}} placeholder remains after substitution (i.e. the state is missing required keys).

Source code in src/symfonic/agent/prompts/system_prompt.py

async def render(self, state: PromptState) -> str:
    """Render the HMS system prompt by substituting placeholders.

    Args:
        state: Mapping that must contain values for every
            ``{{VARIABLE}}`` placeholder in the template.

    Returns:
        The fully substituted prompt text.

    Raises:
        SecurityScopeError: If ``TENANT_ID`` is missing or empty.
        ValueError: If any ``{{VARIABLE}}`` placeholder remains after
            substitution (i.e. the state is missing required keys).
    """
    return self.render_sync(state)

render_sync

render_sync(state: PromptState) -> str

The whole of :meth:render, without the coroutine.

Extracted for #21: a prompting source's read is synchronous by contract, and rendering never awaited anything -- render is a coroutine only because PromptSection is. Splitting it keeps one implementation of the placeholder rules, which is the point of porting rather than rewriting: the upper/lower key fallback is invisible until an adopter's state happens to use the other case.

Source code in src/symfonic/agent/prompts/system_prompt.py

def render_sync(self, state: PromptState) -> str:
    """The whole of :meth:`render`, without the coroutine.

    Extracted for #21: a prompting source's ``read`` is synchronous by
    contract, and rendering never awaited anything -- ``render`` is a
    coroutine only because ``PromptSection`` is. Splitting it keeps *one*
    implementation of the placeholder rules, which is the point of porting
    rather than rewriting: the upper/lower key fallback is invisible until
    an adopter's state happens to use the other case.
    """
    tenant_id: Any = state.get("TENANT_ID") or state.get("tenant_id")
    if not tenant_id or not str(tenant_id).strip():
        raise SecurityScopeError(
            "tenant_id is required for HMS prompt rendering"
        )

    template = self._get_template()
    rendered = _substitute(template, state)
    remaining = _PLACEHOLDER_RE.findall(rendered)
    if remaining:
        raise ValueError(
            f"Unresolved HMS system prompt placeholders: {remaining}"
        )
    return rendered

validate_budget

validate_budget(rendered: str, max_tokens: int = 1000) -> tuple[bool, int]

Check whether rendered fits within the token budget.

Does NOT raise — callers decide how to handle over-budget results.

Parameters:

Name	Type	Description	Default
rendered	str	The already-rendered prompt text.	required
max_tokens	int	Maximum allowed token count.	1000

Returns:

Type	Description
tuple[bool, int]	Tuple of (within_budget, estimated_tokens).

Source code in src/symfonic/agent/prompts/system_prompt.py

def validate_budget(
    self, rendered: str, max_tokens: int = 1000
) -> tuple[bool, int]:
    """Check whether *rendered* fits within the token budget.

    Does NOT raise — callers decide how to handle over-budget results.

    Args:
        rendered: The already-rendered prompt text.
        max_tokens: Maximum allowed token count.

    Returns:
        Tuple of ``(within_budget, estimated_tokens)``.
    """
    estimated = self.estimate_tokens(rendered)
    return estimated <= max_tokens, estimated

MemoryExtractionSection dataclass

MemoryExtractionSection(name: str = 'Memory Extraction Directives', required: bool = False, priority: int = 90, cache_breakpoint: bool = False, template_path: Path | None = None, template_name: str = _EXTRACTION_TEMPLATE)

PromptSection that renders the memory extraction directives.

Substitutes {{TENANT_ID}} from the PromptState context. Used as a post-response instruction to guide structured memory mutations.

Attributes:

Name	Type	Description
name	str	Section display name used by PromptBuilder.
required	bool	Whether a render failure should abort the build.
priority	int	Lower numbers appear earlier in the assembled prompt.
template_path	Path | None	Optional override path for extraction.txt.
template_name	str	Bundled template to load when template_path is unset (v9.2.0 #25.4c): "extraction" (default, Anthropic delimiter) or "extraction_openai" (explicit directive for OpenAI-family models). Ignored when template_path points at a file.

render async

render(state: PromptState) -> str

Render the extraction directives with state substitution.

Parameters:

Name	Type	Description	Default
state	PromptState	Must contain tenant_id or TENANT_ID. May also contain DOMAIN_NAME, REQUIRED_LABELS, ONBOARDING_CHECKLIST, SOUL_SCHEMA, and known_entities (a list of {id, label, type} dicts) from the domain template / hydration pipeline.	required

Returns:

Type	Description
str	Rendered extraction directive string, or empty string if no
str	tenant is present in state. Byte-for-byte compatible with
str	pre-7.4 behaviour: the KNOWN ENTITIES block is interleaved
str	at the {{KNOWN_ENTITIES_SECTION}} position in the
str	template, exactly where pre-split callers expected it.

Source code in src/symfonic/agent/prompts/extraction_prompt.py

async def render(self, state: PromptState) -> str:
    """Render the extraction directives with state substitution.

    Args:
        state: Must contain ``tenant_id`` or ``TENANT_ID``.
            May also contain ``DOMAIN_NAME``, ``REQUIRED_LABELS``,
            ``ONBOARDING_CHECKLIST``, ``SOUL_SCHEMA``, and
            ``known_entities`` (a list of ``{id, label, type}``
            dicts) from the domain template / hydration pipeline.

    Returns:
        Rendered extraction directive string, or empty string if no
        tenant is present in state.  Byte-for-byte compatible with
        pre-7.4 behaviour: the KNOWN ENTITIES block is interleaved
        at the ``{{KNOWN_ENTITIES_SECTION}}`` position in the
        template, exactly where pre-split callers expected it.
    """
    stable_part, volatile_part = await self.render_split(
        state, _legacy=True,
    )
    if not stable_part and not volatile_part:
        return ""
    # Legacy mode: the volatile block was interleaved at the
    # ``{{KNOWN_ENTITIES_SECTION}}`` placeholder position. The
    # ``_legacy=True`` path above performs that interleaving inline
    # and returns the joined string in ``stable_part``, leaving
    # ``volatile_part`` empty -- so the legacy public output is a
    # single byte-for-byte identical string.
    return stable_part

render_split async

render_split(state: PromptState, *, _legacy: bool = False) -> tuple[str, str]

Render the extraction directives as a stable/volatile pair.

v7.4 Item 15: the extraction protocol's stable surface (schema, label rules, few-shot example, domain directives) lives in L1 (cached prefix) while the per-turn volatile KNOWN ENTITIES block lives in L2. This method returns the two strings separately so the engine can route them to different cache layers.

Parameters:

Name	Type	Description	Default
state	PromptState	Same contract as :meth:render.	required
_legacy	bool	INTERNAL. When True, the volatile KNOWN ENTITIES block is interleaved into the stable string at the {{KNOWN_ENTITIES_SECTION}} placeholder position, exactly matching pre-split byte output for legacy callers. volatile_part is then returned as "". Public callers should NOT set this -- use the two-tuple shape and route each part independently.	False

Returns:

Type	Description
str	(stable_part, volatile_part). Either or both may be
str	empty. When no tenant is in state, both are empty.

Source code in src/symfonic/agent/prompts/extraction_prompt.py

async def render_split(
    self,
    state: PromptState,
    *,
    _legacy: bool = False,
) -> tuple[str, str]:
    """Render the extraction directives as a stable/volatile pair.

    v7.4 Item 15: the extraction protocol's stable surface (schema,
    label rules, few-shot example, domain directives) lives in L1
    (cached prefix) while the per-turn volatile KNOWN ENTITIES
    block lives in L2.  This method returns the two strings
    separately so the engine can route them to different cache
    layers.

    Args:
        state: Same contract as :meth:`render`.
        _legacy: INTERNAL.  When ``True``, the volatile KNOWN
            ENTITIES block is interleaved into the stable string at
            the ``{{KNOWN_ENTITIES_SECTION}}`` placeholder
            position, exactly matching pre-split byte output for
            legacy callers.  ``volatile_part`` is then returned as
            ``""``.  Public callers should NOT set this -- use the
            two-tuple shape and route each part independently.

    Returns:
        ``(stable_part, volatile_part)``.  Either or both may be
        empty.  When no tenant is in state, both are empty.
    """
    tenant_id: str | None = state.get("TENANT_ID") or state.get("tenant_id")  # type: ignore[assignment]
    if not tenant_id:
        return "", ""

    template = self._get_template()
    rendered = template.replace("{{TENANT_ID}}", str(tenant_id))

    # Render the optional KNOWN ENTITIES directive.  An empty list
    # collapses to "" so we don't emit a header with no body.
    known_entities = state.get("known_entities") or state.get("KNOWN_ENTITIES")  # type: ignore[assignment]
    known_section = _render_known_entities_section(
        known_entities if isinstance(known_entities, list) else None,
    )

    if _legacy:
        # Interleave at the placeholder position so legacy callers
        # see the pre-7.4 byte layout.
        rendered = rendered.replace(
            "{{KNOWN_ENTITIES_SECTION}}", known_section,
        )
        volatile_part = ""
    else:
        # Stable part: drop the placeholder entirely so the L1 body
        # contains zero per-turn content.  Volatile part: the
        # rendered KNOWN ENTITIES directive (with its leading
        # newline preserved so it can be appended cleanly after
        # other L2 content).
        rendered = rendered.replace("{{KNOWN_ENTITIES_SECTION}}", "")
        volatile_part = known_section.lstrip("\n") if known_section else ""

    # Substitute domain-specific placeholders with defaults.
    _domain_defaults: dict[str, str] = {
        "DOMAIN_NAME": "default",
        "REQUIRED_LABELS": "(none)",
        "ONBOARDING_CHECKLIST": "  (none)",
        "SOUL_SCHEMA": "  (none)",
    }
    for key, default in _domain_defaults.items():
        value = str(state.get(key, default))  # type: ignore[arg-type]
        rendered = rendered.replace(f"{{{{{key}}}}}", value)

    # Validate no other placeholders remain in the stable body.
    remaining = _PLACEHOLDER_RE.findall(rendered)
    if remaining:
        raise ValueError(
            f"Unresolved extraction prompt placeholders: {remaining}"
        )
    return rendered, volatile_part