symfonic.capabilities.knowledge

knowledge

Knowledge, document, attachment, and citation bridges (T3.2.3).

Seven things used to decide independently how retrieved and ingested content reached a model: a knowledge provider formatted its own citation lines, a document store built its own block, extractors returned text nobody bounded, OCR output went wherever the caller put it, multimodal normalisation and provider wire formatting each had their own idea of what a media type was, and a streaming filter stripped citation tags in a different package again.

This capability makes each of them a contribution instead: a declaration that states what it supplies and how far it may be trusted, and never where it goes. Position, budget, cache region, and the untrusted wrapper are decided once, by T3.2.1's compiler, over the whole set. The bridges here decide only what is theirs to decide — what a store returns, what a parser is allowed to see, and what may go back out on the wire.

Three deliverables live here:

• Knowledge bridge — :mod:.retrieval.
• Document and attachment bridge — :mod:.documents, :mod:.attachments, :mod:.multimodal, :mod:.outbound.
• Citation and output bridge — :mod:.citations, :mod:.grounding.

with :mod:.contracts holding the declaration all three emit, :mod:.labels holding the "untrusted name is a label, never an identifier" rule, and :mod:.assembly holding the one-function seam to the prompt compiler.

The T1.2.5 attack-surface contracts are preserved rather than restated: AS-ING-1 becomes a constructor that refuses an authored tier, AS-ING-2 becomes :func:~.labels.safe_label plus a document-id charset, AS-ING-5/6 become the ceilings and the exception wrapper around every parser call, and AS-NET-1/2/3 become :func:~.outbound.check_outbound_url and wire formatters whose only parameter is the content.

AttachmentRef dataclass

AttachmentRef(label: str, kind: str, media_type: str, payload: bytes)

One attachment as it arrived: an untrusted name and untrusted bytes.

AttachmentSource dataclass

AttachmentSource(refs: Sequence[AttachmentRef], extractor: TextExtractor, limits: ExtractionLimits = DEFAULT_LIMITS, on_rejection: Literal['omit', 'raise'] = 'omit', scope_aware: bool = False, offline_safe: bool = True)

A :class:~.contracts.ContextSource over a fixed set of attachments.

select

select(request: ContextRequest) -> IngestSelection

Extract the attachments, and say why each omitted one was omitted.

on_rejection='raise' makes a refusal loud by ending the read; 'omit' — the default, and the right one when four of five files should still reach the model — used to make it silent. It is neither now: the omission and its reason are carried out on the selection, so a composition root can surface "report.pdf: parser failed" the way the prompting capability surfaces a PromptDiagnostic, rather than leaving an operator to guess why a file they attached is absent.

Source code in src/symfonic/capabilities/knowledge/attachments.py

def select(self, request: ContextRequest) -> IngestSelection:
    """Extract the attachments, and say why each omitted one was omitted.

    ``on_rejection='raise'`` makes a refusal loud by ending the read;
    ``'omit'`` — the default, and the right one when four of five files
    should still reach the model — used to make it silent. It is neither
    now: the omission and its reason are carried out on the selection, so a
    composition root can surface "report.pdf: parser failed" the way the
    prompting capability surfaces a ``PromptDiagnostic``, rather than
    leaving an operator to guess why a file they attached is absent.
    """
    blocks: list[str] = []
    labels: list[str] = []
    dropped: list[tuple[str, str]] = []
    used = 0
    for ref in self.refs:
        try:
            extracted = extract_attachment(ref, self.extractor, self.limits)
        except (IngestionRejected, ExtractionFailed) as exc:
            if self.on_rejection == "raise":
                raise
            dropped.append((safe_label(ref.label), f"{type(exc).__name__}: {exc}"))
            continue
        if not extracted.text.strip():
            dropped.append(
                (extracted.label, "the extractor returned nothing but whitespace")
            )
            continue
        block = f"ATTACHMENT [{extracted.label}]\n{extracted.text}"
        cost = len(block) + (len(_BLOCK_SEPARATOR) if blocks else 0)
        if used + cost > self.limits.max_total_chars:
            if self.on_rejection == "raise":
                raise IngestionRejected(
                    f"attachment {extracted.label!r} would take the assembled block "
                    f"past the {self.limits.max_total_chars}-character ceiling; the "
                    "per-attachment cap bounds one file, this one bounds the set."
                )
            dropped.append(
                (
                    extracted.label,
                    "would take the assembled block past the "
                    f"{self.limits.max_total_chars}-character ceiling",
                )
            )
            continue
        blocks.append(block)
        labels.append(f"{extracted.label}:{extracted.byte_size}")
        used += cost
    return IngestSelection(
        text=_BLOCK_SEPARATOR.join(blocks),
        revision="|".join(labels),
        dropped=tuple(dropped),
    )

BridgeContractError

Bases: KnowledgeBridgeError

A declaration is impossible: wrong tier, wrong layer, unusable source.

Raised at declaration time, never at render time. A deployment that declares retrieved content at an authored tier is misconfigured, and the only useful moment to say so is before the first turn.

Citation dataclass

Citation(marker: str, source: str)

One numbered, display-safe reference to a retrieved source.

CitationIndex dataclass

CitationIndex(citations: tuple[Citation, ...] = ())

The grounded set: every source a retrieval actually returned.

from_fragments classmethod

from_fragments(fragments: Iterable[RetrievedFragment]) -> CitationIndex

Number the distinct sources of fragments in ranked order.

Ranked, not received: the numbering has to be a function of the content so two runs over the same retrieval produce the same reference list.

Source code in src/symfonic/capabilities/knowledge/grounding.py

@classmethod
def from_fragments(cls, fragments: Iterable[RetrievedFragment]) -> CitationIndex:
    """Number the distinct sources of ``fragments`` in ranked order.

    Ranked, not received: the numbering has to be a function of the content
    so two runs over the same retrieval produce the same reference list.
    """
    ranked = sorted(fragments, key=lambda f: (-f.score, f.source, f.content))
    seen: list[str] = []
    for fragment in ranked:
        label = safe_source(fragment.source)
        if label and label not in seen:
            seen.append(label)
    return cls(
        tuple(Citation(marker=str(i), source=s) for i, s in enumerate(seen, start=1))
    )

render

render() -> str

The reference list, one citation per line.

Source code in src/symfonic/capabilities/knowledge/grounding.py

def render(self) -> str:
    """The reference list, one citation per line."""
    return "\n".join(f"[{c.marker}] {c.source}" for c in self.citations)

sources

sources() -> frozenset[str]

The grounded source labels a reference may name.

Source code in src/symfonic/capabilities/knowledge/grounding.py

def sources(self) -> frozenset[str]:
    """The grounded source labels a reference may name."""
    return frozenset(citation.source for citation in self.citations)

CitationTagFilter

CitationTagFilter(layer_names: Sequence[str] = DEFAULT_LAYER_NAMES)

Streaming filter that strips internal layer tags across chunk boundaries.

Feed chunks in order and concatenate what comes back; call :meth:flush once at stream end. A partial prefix that never completes is dropped at flush rather than emitted: it was only held back because it matched a known layer name, so emitting it would leak a fragment of an internal annotation.

Source code in src/symfonic/capabilities/knowledge/citations.py

def __init__(self, layer_names: Sequence[str] = DEFAULT_LAYER_NAMES) -> None:
    if not layer_names:
        raise ValueError(
            "a citation filter needs at least one layer name; an empty vocabulary "
            "is a filter that strips nothing, which is better written as no filter."
        )
    self._tags = tuple(f"[{name.lower()}]" for name in layer_names)
    # T3.2.4 finding 3: horizontal whitespace only, never ``\s*``.
    #
    # ``\s*`` is greedy over newlines too, so a tag on its own indented line
    # took the preceding line break with it and welded two lines together:
    # ``"code:\n    [working]    indented"`` became ``"code:    indented"``.
    # That was the most destructive of the three answers one input could
    # produce, and it is a *whole-buffer* defect — under chunked delivery
    # the newline had already been emitted and survived, which is why it is
    # a different finding from the chunk-invariance one and not the same one
    # seen twice.
    self._tag_re = re.compile(
        r"[ \t]*\[(?:" + "|".join(re.escape(name) for name in layer_names) + r")\]",
        re.IGNORECASE,
    )
    self._buffer = ""
    self._last_emitted_was_space = False
    self._pending_seam = False

feed

feed(text: str) -> str

Append text and return the prefix that is safe to emit.

Source code in src/symfonic/capabilities/knowledge/citations.py

def feed(self, text: str) -> str:
    """Append ``text`` and return the prefix that is safe to emit."""
    if not text:
        return ""
    self._buffer += text
    return self._drain(final=False)

flush

flush() -> str

Drain the buffer at stream end, dropping any unfinished tag prefix.

Source code in src/symfonic/capabilities/knowledge/citations.py

def flush(self) -> str:
    """Drain the buffer at stream end, dropping any unfinished tag prefix."""
    return self._drain(final=True)

ContentPart dataclass

ContentPart(part_type: PartType, text: str = '', media_type: str = '', data_b64: str = '')

One provider-neutral piece of a multimodal message.

ContextContribution dataclass

ContextContribution(contribution_id: str, source: ContextSource, capability: str = 'knowledge', layer: ContextLayer = ContextLayer.L2, tier: ContextTier = ContextTier.SESSION, scope: ContextScope = ContextScope.DEPLOYMENT, order: int = 0, inherit: bool = True, pinned: bool = False, requires_hydration: bool = True)

One bridge's declaration of ingested context it contributes.

validate

validate() -> None

Refuse every declaration this capability is not allowed to make.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def validate(self) -> None:
    """Refuse every declaration this capability is not allowed to make."""
    if not self.contribution_id:
        raise BridgeContractError(
            "an ingested-context contribution must declare a non-empty contribution_id."
        )
    if not _ID_CHARSET.match(self.contribution_id):
        raise BridgeContractError(
            f"contribution id {self.contribution_id!r} is outside the permitted charset "
            "[A-Za-z0-9_.-]; ids appear in isolation keys and in the rendered untrusted "
            "delimiter, where a separator or an angle bracket forges a boundary."
        )
    if not callable(getattr(self.source, "read", None)):
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares a source that cannot read: "
            f"{type(self.source).__name__} has no callable read()."
        )
    if self.tier in AUTHORED_TIERS:
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares the authored tier "
            f"{self.tier.value!r}. Retrieved, stored, and extracted content is untrusted "
            "by definition (AS-ING-1) and renders as data; an authored tier renders it "
            "verbatim beside the operator's own instructions."
        )
    if self.layer is ContextLayer.L0:
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares layer L0. L0 is the cached "
            "authored prefix: ingested content placed there is served back to the model "
            "on every later turn of the session, so one poisoned retrieval outlives the "
            "turn that fetched it. Declare L1 (standing) or L2 (per turn)."
        )
    if self.pinned:
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares itself pinned. Pinned content "
            "fails the compile rather than being dropped for budget; ingested content is "
            "evidence, not instruction, and a turn without it is still a correct turn."
        )
    if self.scope is not ContextScope.DEPLOYMENT and not getattr(
        self.source, "scope_aware", False
    ):
        raise BridgeContractError(
            f"contribution {self.contribution_id!r} declares scope={self.scope.value!r} "
            f"but its source {type(self.source).__name__} is not scope_aware: it serves "
            "one value for every tenant. Declare scope='deployment', or supply a source "
            "that keys on the scope path."
        )

ContextContributor

Bases: Protocol

What a knowledge-side capability registers with the composition root.

ContextLayer

Bases: StrEnum

The stratigraphic layer a contribution renders on.

ContextRead dataclass

ContextRead(text: str, revision: str = '', untrusted: bool = True)

What an ingested-context source answered.

untrusted defaults to True — the inverse of the prompt contract's default, and the whole point of a separate value type. In the general contract a source must remember to declare its payload untrusted; here it would have to remember to declare it trusted, which nothing in this capability ever does.

ContextRequest dataclass

ContextRequest(contribution_id: str, scope_path: str = '', turn: int = 0)

What a source is asked for: one contribution, one scope, one turn.

ContextScope

Bases: StrEnum

How widely one contribution's content is shared.

ContextSource

Bases: Protocol

Reads the current ingested content for one contribution and scope.

ContextTier

Bases: StrEnum

Authority tiers, mirroring the prompt contract's vocabulary.

DocumentPolicy dataclass

DocumentPolicy(max_document_chars: int = 20000, max_total_chars: int = 60000)

The ceilings applied to a rendered document block.

DocumentSource dataclass

DocumentSource(store: DocumentStore, document_ids: Sequence[str], policy: DocumentPolicy = DocumentPolicy(), scope_aware: bool = False, offline_safe: bool = False)

A :class:~.contracts.ContextSource backed by a document store.

select

select(request: ContextRequest) -> IngestSelection

Read the documents, and say why each omitted one was omitted.

Four different things make a declared document not appear — the store has no such id, the document is blank, it is over the per-document ceiling, or it would cross the aggregate one — and the rendered text looks identical in all four. The reason channel is what lets a composition root tell an operator which happened, the way the knowledge bridge's :class:~.retrieval.FragmentSelection already does.

Source code in src/symfonic/capabilities/knowledge/documents.py

def select(self, request: ContextRequest) -> IngestSelection:
    """Read the documents, and say why each omitted one was omitted.

    Four different things make a declared document not appear — the store
    has no such id, the document is blank, it is over the per-document
    ceiling, or it would cross the aggregate one — and the rendered text
    looks identical in all four. The reason channel is what lets a
    composition root tell an operator which happened, the way the knowledge
    bridge's :class:`~.retrieval.FragmentSelection` already does.
    """
    blocks: list[str] = []
    revisions: list[str] = []
    dropped: list[tuple[str, str]] = []
    used = 0
    for document_id in self.document_ids:
        validate_document_id(document_id)
        document = self.store.fetch(document_id)
        if document is None:
            dropped.append((document_id, "the store holds no document under this id"))
            continue
        if not document.text.strip():
            dropped.append(
                (document_id, "the document is present but its text is blank")
            )
            continue
        if len(document.text) > self.policy.max_document_chars:
            dropped.append(
                (
                    document_id,
                    f"{len(document.text)} chars exceeds the per-document cap of "
                    f"{self.policy.max_document_chars}; documents are dropped, "
                    "never truncated",
                )
            )
            continue
        block = _render(document)
        cost = len(block) + (len(_BLOCK_SEPARATOR) if blocks else 0)
        if used + cost > self.policy.max_total_chars:
            dropped.append(
                (
                    document_id,
                    "would take the assembled block past the "
                    f"{self.policy.max_total_chars}-char ceiling",
                )
            )
            continue
        blocks.append(block)
        revisions.append(f"{document.document_id}:{document.revision}")
        used += cost
    return IngestSelection(
        text=_BLOCK_SEPARATOR.join(blocks),
        revision="|".join(revisions),
        dropped=tuple(dropped),
    )

DocumentStore

Bases: Protocol

The port a document store adapter satisfies.

None means "not here", not "empty". A store that answered with an empty document for a missing id would make a deleted document indistinguishable from a blank one, and the bridge would render a heading over nothing.

ExtractedAttachment dataclass

ExtractedAttachment(label: str, kind: str, media_type: str, text: str, byte_size: int)

One attachment after extraction, carrying an opaque label.

ExtractionFailed

Bases: KnowledgeBridgeError

A parser could not read one input.

__cause__ carries the underlying library exception. Format parsers are assumed compromise-prone (AS-ING-5); their failure is a typed rejection of that input and never propagates as a raw third-party exception.

ExtractionLimits dataclass

ExtractionLimits(max_bytes: int = 8 * 1024 * 1024, max_uncompressed_bytes: int = 8 * 1024 * 1024, max_expansion_ratio: float = 100.0, max_members: int = 1024, max_nesting_depth: int = 1, max_pages: int = 20, max_text_chars: int = 50000, max_total_chars: int = 120000)

The ceilings applied around a parser call.

FragmentSelection dataclass

FragmentSelection(admitted: tuple[RetrievedFragment, ...] = (), dropped: tuple[tuple[str, str], ...] = ())

What survived selection, and why the rest did not.

IngestSelection dataclass

IngestSelection(text: str = '', revision: str = '', dropped: tuple[tuple[str, str], ...] = ())

What a source emitted, plus a reason for everything it left out.

The knowledge bridge already answers this way — FragmentSelection.dropped carries a (source, reason) pair per refused fragment — and the document and attachment bridges refuse content for the same kinds of reason. Without the channel, an operator asking "why is the handbook not in the prompt?" cannot tell a store that returned None from a document over the per-document ceiling from one evicted by the aggregate cap: the read's text shows the same absence in all three cases and its revision lists only what survived.

:meth:as_read is what the :class:ContextSource protocol consumes; the selection itself is what a composition root reads when it wants to surface the omissions the way the prompting capability surfaces PromptDiagnostic.

as_read

as_read() -> ContextRead

Project the selection into the value :meth:ContextSource.read returns.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def as_read(self) -> ContextRead:
    """Project the selection into the value :meth:`ContextSource.read` returns."""
    return ContextRead(text=self.text, revision=self.revision, untrusted=True)

IngestionRejected

Bases: KnowledgeBridgeError

One input hit an ingestion guard: size, count, shape, or identifier.

The guard fired before the parser where the contract allows it (AS-ING-6), so this is cheap and says nothing about the payload's contents.

KnowledgeBridgeError

Bases: Exception

Base for everything this capability raises.

KnowledgeRetriever

Bases: Protocol

The port a vector store adapter satisfies.

Synchronous on purpose. The compiler's source protocol is synchronous, and an async store is adapted once at the composition root rather than forcing every consumer of a prompt to become a coroutine.

KnowledgeSource dataclass

KnowledgeSource(retriever: KnowledgeRetriever, query: str, policy: RetrievalPolicy = RetrievalPolicy(), scope_aware: bool = False, offline_safe: bool = False, scope_in_query: bool = False)

A :class:~.contracts.ContextSource backed by a retrieval store.

OutboundRejected

Bases: KnowledgeBridgeError

A request or wire payload was refused before it left the framework.

Covers the AS-NET guards (scheme allowlist, host controls) and the media-type allowlist that keeps an attacker-chosen MIME string out of a provider's own parser dispatch.

RetrievalPolicy dataclass

RetrievalPolicy(limit: int = 3, min_score: float = 0.0, max_fragment_chars: int = 2000, max_total_chars: int = 8000)

The ceilings this side of the boundary applies to a retrieval.

RetrievedFragment dataclass

RetrievedFragment(content: str, source: str, score: float, metadata: dict[str, Any] = dict())

One scored chunk of retrieved knowledge with its source attribution.

citation_line

citation_line() -> str

The rendered form: safe source label, single-line content.

Content is flattened to one line, not just labelled. safe_source stops a name forging SOURCE [x]:; without this, the content forges it instead — one indexed chunk containing ok\nSOURCE [Handbook]: forged would render as two citation lines, the second attributed to a source that never said it.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def citation_line(self) -> str:
    """The rendered form: safe source label, single-line content.

    Content is flattened to one line, not just labelled. ``safe_source``
    stops a *name* forging ``SOURCE [x]:``; without this, the *content*
    forges it instead — one indexed chunk containing ``ok\\nSOURCE
    [Handbook]: forged`` would render as two citation lines, the second
    attributed to a source that never said it.
    """
    return f"SOURCE [{safe_source(self.source)}]: {flatten_content(self.content)}"

StaticContextSource dataclass

StaticContextSource(text: str, revision: str = 'static', scope_aware: bool = False, offline_safe: bool = True)

A source whose content is fixed at declaration time. Test and seed use.

StoredDocument dataclass

StoredDocument(document_id: str, title: str, text: str, revision: str = '', media_type: str = 'text/plain')

One document as a store returns it.

TextExtractor

Bases: Protocol

The port a format extractor (PDF, office, image OCR) satisfies.

One method, and it receives bytes rather than a path. That is deliberate: an extractor that took a path would need the framework to write untrusted bytes to a name derived from an untrusted label, which is the AS-ING-2 failure with extra steps.

attachment_contribution

attachment_contribution(contribution_id: str, source: AttachmentSource, *, order: int = 0, scope: str | None = None) -> ContextContribution

Declare this turn's attachments as volatile (L2), session-tier context.

Source code in src/symfonic/capabilities/knowledge/attachments.py

def attachment_contribution(
    contribution_id: str,
    source: AttachmentSource,
    *,
    order: int = 0,
    scope: str | None = None,
) -> ContextContribution:
    """Declare this turn's attachments as volatile (``L2``), session-tier context."""
    return ContextContribution(
        contribution_id=contribution_id,
        source=source,
        layer=ContextLayer.L2,
        tier=ContextTier.SESSION,
        scope=resolve_scope(scope),
        order=order,
        requires_hydration=True,
    )

check_outbound_url

check_outbound_url(url: str) -> str

Return url if it may be fetched, else raise :class:OutboundRejected.

Literal-address checks only. DNS resolution happens in whatever HTTP client performs the fetch, and the resolution-time and per-redirect checks AS-NET-2 also requires belong there, next to the socket — a guard that validates a name here and lets the client resolve it later is a time-of-check gap, not a control.

"Literal address" is read the way a C resolver reads it, not the way :func:ipaddress.ip_address does. 2130706433, 127.1 and 0177.0.0.1 all reach 127.0.0.1 without any DNS involvement, so a guard that only understands dotted-quad hands the loopback interface to anyone who writes the address a different way.

And the host is normalised through IDNA before any of that, because the same argument applies one level down: １２７.0.0.1 is the loopback address written in fullwidth digits, and every check here is a comparison against the ASCII spelling. See :func:_normalise_hostname.

Source code in src/symfonic/capabilities/knowledge/outbound.py

def check_outbound_url(url: str) -> str:
    """Return ``url`` if it may be fetched, else raise :class:`OutboundRejected`.

    Literal-address checks only. DNS resolution happens in whatever HTTP client
    performs the fetch, and the resolution-time and per-redirect checks
    AS-NET-2 also requires belong there, next to the socket — a guard that
    validates a name here and lets the client resolve it later is a
    time-of-check gap, not a control.

    "Literal address" is read the way a C resolver reads it, not the way
    :func:`ipaddress.ip_address` does. ``2130706433``, ``127.1`` and
    ``0177.0.0.1`` all reach ``127.0.0.1`` without any DNS involvement, so a
    guard that only understands dotted-quad hands the loopback interface to
    anyone who writes the address a different way.

    And the host is normalised through IDNA *before* any of that, because the
    same argument applies one level down: ``１２７.0.0.1`` is the loopback
    address written in fullwidth digits, and every check here is a comparison
    against the ASCII spelling. See :func:`_normalise_hostname`.
    """
    parsed = urlparse(url)
    if parsed.scheme.lower() not in ALLOWED_URL_SCHEMES:
        raise OutboundRejected(
            f"outbound scheme {parsed.scheme!r} is not permitted; untrusted-derived "
            f"targets may use {sorted(ALLOWED_URL_SCHEMES)} only (AS-NET-1)."
        )
    hostname = (parsed.hostname or "").strip()
    if not hostname:
        raise OutboundRejected(
            f"outbound url {url!r} names no host; a target the framework cannot "
            "identify is a target it cannot check."
        )
    hostname = _normalise_hostname(hostname)
    # A trailing dot is the DNS root marker: 'localhost.' and '127.0.0.1.' reach
    # exactly what their undotted forms reach, so it is removed before any check
    # rather than allowed to route around all of them. It is removed *after*
    # normalisation because U+FF0E and U+3002 are dots too, and only IDNA knows
    # that.
    hostname = hostname.rstrip(".") or hostname
    if hostname.lower() in _DENIED_HOSTNAMES or hostname.lower().endswith(".localhost"):
        raise OutboundRejected(
            f"outbound host {hostname!r} names the local host; untrusted-derived "
            "targets may not reach the framework's own machine (AS-NET-2)."
        )
    _reject_numeric_literal(hostname)
    _reject_denied_address(hostname)
    return url

contribution_spec

contribution_spec(contribution: ContextContribution) -> Mapping[str, object]

Project a validated declaration into compiler-ready keyword values.

Validation runs first and unconditionally. A spec is the last moment this capability controls, so emitting one for a declaration it would have refused would move the refusal into the compiler, where the error message can no longer explain which bridge got it wrong.

Enum members are emitted as their string values. The prompt contract's layers, tiers, and scopes are StrEnums over the same strings, so the composition root's conversion is total by construction.

Source code in src/symfonic/capabilities/knowledge/assembly.py

def contribution_spec(contribution: ContextContribution) -> Mapping[str, object]:
    """Project a validated declaration into compiler-ready keyword values.

    Validation runs first and unconditionally. A spec is the last moment this
    capability controls, so emitting one for a declaration it would have
    refused would move the refusal into the compiler, where the error message
    can no longer explain which bridge got it wrong.

    Enum members are emitted as their string values. The prompt contract's
    layers, tiers, and scopes are ``StrEnum``s over the same strings, so the
    composition root's conversion is total by construction.
    """
    contribution.validate()
    return MappingProxyType(
        {
            "contribution_id": contribution.contribution_id,
            "source": contribution.source,
            "capability": contribution.capability,
            "layer": contribution.layer.value,
            "tier": contribution.tier.value,
            "scope": contribution.scope.value,
            "order": contribution.order,
            "inherit": contribution.inherit,
            "pinned": contribution.pinned,
            "requires_hydration": contribution.requires_hydration,
        }
    )

document_contribution

document_contribution(contribution_id: str, source: DocumentSource, *, order: int = 0, scope: str | None = None) -> ContextContribution

Declare a document set as standing (L1), session-tier context.

L1 rather than L2: the documents attached to a conversation are stable for its life, and putting them on the volatile layer would push the per-turn boundary above them and forfeit the cached prefix for content that never changes.

Source code in src/symfonic/capabilities/knowledge/documents.py

def document_contribution(
    contribution_id: str,
    source: DocumentSource,
    *,
    order: int = 0,
    scope: str | None = None,
) -> ContextContribution:
    """Declare a document set as standing (``L1``), session-tier context.

    ``L1`` rather than ``L2``: the documents attached to a conversation are
    stable for its life, and putting them on the volatile layer would push the
    per-turn boundary above them and forfeit the cached prefix for content that
    never changes.
    """
    return ContextContribution(
        contribution_id=contribution_id,
        source=source,
        layer=ContextLayer.L1,
        tier=ContextTier.SESSION,
        scope=resolve_scope(scope),
        order=order,
        requires_hydration=True,
    )

drop_ungrounded_citations

drop_ungrounded_citations(text: str, index: CitationIndex) -> tuple[str, tuple[str, ...]]

Remove [[source:X]] references whose X was never retrieved.

Returns the rewritten text and the ungrounded names, each reported once in first-appearance order. When nothing was ungrounded the text is returned byte-identical — a filter that reflowed whitespace on every clean response would make its own no-op observable.

Source code in src/symfonic/capabilities/knowledge/grounding.py

def drop_ungrounded_citations(text: str, index: CitationIndex) -> tuple[str, tuple[str, ...]]:
    """Remove ``[[source:X]]`` references whose ``X`` was never retrieved.

    Returns the rewritten text and the ungrounded names, each reported once in
    first-appearance order. When nothing was ungrounded the text is returned
    byte-identical — a filter that reflowed whitespace on every clean response
    would make its own no-op observable.
    """
    grounded = index.sources()
    dropped: list[str] = []

    def replace(match: re.Match[str]) -> str:
        name = match.group(1).strip()
        if safe_source(name) in grounded:
            return match.group(0)
        if name not in dropped:
            dropped.append(name)
        return ""

    rewritten = SOURCE_REFERENCE.sub(replace, text)
    if not dropped:
        return text, ()
    return _SPACE_RUN.sub(" ", rewritten).strip(), tuple(dropped)

extract_attachment

extract_attachment(ref: AttachmentRef, extractor: TextExtractor, limits: ExtractionLimits = DEFAULT_LIMITS) -> ExtractedAttachment

Extract one attachment's text under the ingestion ceilings.

Raises :class:~.errors.IngestionRejected when a guard fires and :class:~.errors.ExtractionFailed when the parser itself loses. Nothing a third-party parser raises escapes this function untyped.

Source code in src/symfonic/capabilities/knowledge/attachments.py

def extract_attachment(
    ref: AttachmentRef,
    extractor: TextExtractor,
    limits: ExtractionLimits = DEFAULT_LIMITS,
) -> ExtractedAttachment:
    """Extract one attachment's text under the ingestion ceilings.

    Raises :class:`~.errors.IngestionRejected` when a guard fires and
    :class:`~.errors.ExtractionFailed` when the parser itself loses. Nothing a
    third-party parser raises escapes this function untyped.
    """
    if ref.kind not in EXTRACTABLE_KINDS:
        raise IngestionRejected(
            f"attachment kind {ref.kind!r} is not extractable; permitted kinds are "
            f"{sorted(EXTRACTABLE_KINDS)}. An unrecognised container is refused before "
            "dispatch, not handed to a parser to find out."
        )
    if not ref.payload:
        raise IngestionRejected(
            f"attachment {safe_label(ref.label)!r} carries an empty payload; there is "
            "nothing to extract and an empty block would render a heading over nothing."
        )
    if len(ref.payload) > limits.max_bytes:
        raise IngestionRejected(
            f"attachment {safe_label(ref.label)!r} is {len(ref.payload)} bytes, which "
            f"exceeds the {limits.max_bytes}-byte raw-input ceiling checked before "
            "dispatch (AS-ING-6); archive expansion is bounded separately."
        )
    if is_zip_container(ref.media_type, ref.payload):
        try:
            inspect_zip_limits(
                ref.payload,
                max_uncompressed_bytes=limits.max_uncompressed_bytes,
                max_expansion_ratio=limits.max_expansion_ratio,
                max_members=limits.max_members,
                max_nesting_depth=limits.max_nesting_depth,
            )
        except ArchiveLimitExceeded as exc:
            raise IngestionRejected(
                f"attachment {safe_label(ref.label)!r} was rejected before dispatch: {exc}"
            ) from exc

    try:
        raw = extractor.extract(
            ref.kind, ref.media_type, ref.payload, max_pages=limits.max_pages
        )
    except Exception as exc:  # noqa: BLE001 - a parser is a hostile black box
        raise ExtractionFailed(
            f"extraction of attachment {safe_label(ref.label)!r} "
            f"({ref.kind}/{ref.media_type}) failed: {type(exc).__name__}: {exc}"
        ) from exc

    text = normalise_extracted(raw)
    if len(text) > limits.max_text_chars:
        raise IngestionRejected(
            f"attachment {safe_label(ref.label)!r} extracted to {len(text)} characters, "
            f"which exceeds the {limits.max_text_chars}-character ceiling; extracted text "
            "is dropped, never truncated — a cut fact reads as a complete one."
        )

    return ExtractedAttachment(
        label=safe_label(ref.label),
        kind=ref.kind,
        media_type=ref.media_type,
        text=text,
        byte_size=len(ref.payload),
    )

flatten_content

flatten_content(content: str) -> str

Collapse every line-break form in content to a single space.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def flatten_content(content: str) -> str:
    """Collapse every line-break form in ``content`` to a single space."""
    return _LINE_BREAK.sub(" ", content)

knowledge_contribution

knowledge_contribution(contribution_id: str, source: KnowledgeSource, *, order: int = 0, scope: str | None = None) -> ContextContribution

Declare a retrieval as per-turn, session-tier ingested context.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def knowledge_contribution(
    contribution_id: str,
    source: KnowledgeSource,
    *,
    order: int = 0,
    scope: str | None = None,
) -> ContextContribution:
    """Declare a retrieval as per-turn, session-tier ingested context."""
    return ContextContribution(
        contribution_id=contribution_id,
        source=source,
        layer=ContextLayer.L2,
        tier=ContextTier.SESSION,
        scope=resolve_scope(scope),
        order=order,
        requires_hydration=True,
    )

knowledge_sources

knowledge_sources(*, retriever: Any = None, query: str = '', store: Any = None, document_ids: Sequence[str] = (), attachments: Sequence[Any] = (), extractor: Any = None, context: str = '', **policies: Any) -> tuple[Any, ...]

The sources this deployment's knowledge reaches the prompt through.

Every argument is optional and each one adds a source only when it can actually produce something -- a retriever with no query retrieves nothing, and a store with no ids reads nothing. Returning a source that cannot contribute would cost the compiler a budget slot and a delimiter for an empty block, which is the same rule compose follows in governance: leave it out rather than in-but-inert.

Parameters:

Name	Type	Description	Default
retriever	Any	a KnowledgeRetriever for similarity search.	None
query	str	what to retrieve. Required for retriever to be used.	''
store	Any	a DocumentStore to read pinned documents from.	None
document_ids	Sequence[str]	which documents to pin into the prompt.	()
attachments	Sequence[Any]	refs the caller sent with the turn.	()
extractor	Any	turns an attachment into text. Required for attachments.	None
context	str	static text this deployment always wants present.	''
**policies	Any	policy for retrieval, document_policy for the store, limits for extraction -- each forwarded to its own source, and each already carrying a default.	{}

Source code in src/symfonic/capabilities/knowledge/factory.py

def knowledge_sources(
    *,
    retriever: Any = None,
    query: str = "",
    store: Any = None,
    document_ids: Sequence[str] = (),
    attachments: Sequence[Any] = (),
    extractor: Any = None,
    context: str = "",
    **policies: Any,
) -> tuple[Any, ...]:
    """The sources this deployment's knowledge reaches the prompt through.

    Every argument is optional and each one adds a source only when it can
    actually produce something -- a retriever with no query retrieves nothing,
    and a store with no ids reads nothing. Returning a source that cannot
    contribute would cost the compiler a budget slot and a delimiter for an
    empty block, which is the same rule ``compose`` follows in governance:
    leave it out rather than in-but-inert.

    Args:
        retriever: a ``KnowledgeRetriever`` for similarity search.
        query: what to retrieve. Required for ``retriever`` to be used.
        store: a ``DocumentStore`` to read pinned documents from.
        document_ids: which documents to pin into the prompt.
        attachments: refs the caller sent with the turn.
        extractor: turns an attachment into text. Required for ``attachments``.
        context: static text this deployment always wants present.
        **policies: ``policy`` for retrieval, ``document_policy`` for the
            store, ``limits`` for extraction -- each forwarded to its own
            source, and each already carrying a default.
    """
    sources: list[Any] = []

    if retriever is not None and query:
        sources.append(
            _built(KnowledgeSource, retriever=retriever, query=query,
                   policy=policies.get("policy"))
        )
    if store is not None and document_ids:
        sources.append(
            _built(DocumentSource, store=store, document_ids=tuple(document_ids),
                   policy=policies.get("document_policy"))
        )
    if attachments and extractor is not None:
        sources.append(
            _built(AttachmentSource, refs=tuple(attachments), extractor=extractor,
                   limits=policies.get("limits"))
        )
    if context.strip():
        sources.append(StaticContextSource(text=context.strip()))

    return tuple(sources)

layer_index

layer_index(layer: ContextLayer) -> int

Position of layer on the ladder; lower renders earlier.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def layer_index(layer: ContextLayer) -> int:
    """Position of ``layer`` on the ladder; lower renders earlier."""
    return LAYER_LADDER.index(layer)

media_part

media_part(kind: str, media_type: str, payload: bytes) -> ContentPart

Normalise bytes into a base64 part, under the media-type allowlist.

Source code in src/symfonic/capabilities/knowledge/multimodal.py

def media_part(kind: str, media_type: str, payload: bytes) -> ContentPart:
    """Normalise bytes into a base64 part, under the media-type allowlist."""
    if kind not in _MEDIA_KINDS:
        raise OutboundRejected(
            f"part kind {kind!r} has no wire representation; permitted kinds are "
            f"{sorted(_MEDIA_KINDS)}."
        )
    if not payload:
        raise OutboundRejected(f"the {kind} part carries an empty payload.")
    if not media_type:
        raise OutboundRejected(f"the {kind} part declares no media type.")
    allowed = _ALLOWED_MEDIA_TYPES[kind]
    if media_type not in allowed:
        raise OutboundRejected(
            f"{kind} media type {media_type!r} is outside the allowlist "
            f"{sorted(allowed)}. The media type travels to the provider and selects its "
            "decoder, so an attacker-chosen value is an injection into someone else's "
            "parser — and a document's media type comes from the same untrusted "
            "AttachmentRef an image's does."
        )
    return ContentPart(
        part_type="image" if kind == "image" else "document",
        media_type=media_type,
        data_b64=base64.b64encode(payload).decode("ascii"),
    )

normalise_extracted

normalise_extracted(text: str) -> str

Unix line endings, control characters stripped, trailing space removed.

Source code in src/symfonic/capabilities/knowledge/attachments.py

def normalise_extracted(text: str) -> str:
    """Unix line endings, control characters stripped, trailing space removed."""
    unified = text.replace("\r\n", "\n").replace("\r", "\n")
    stripped = _CONTROL.sub("", unified)
    return "\n".join(line.rstrip() for line in stripped.split("\n"))

ordering_key

ordering_key(contribution: ContextContribution) -> tuple[int, int, str]

Total order: layer, then declared order, then id.

Totality is the point — two contributions that tie on layer and order still have a defined relative position, so what the bridge hands the compiler is byte-stable across runs and across upstream dict iteration orders.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def ordering_key(contribution: ContextContribution) -> tuple[int, int, str]:
    """Total order: layer, then declared order, then id.

    Totality is the point — two contributions that tie on layer and order still
    have a defined relative position, so what the bridge hands the compiler is
    byte-stable across runs and across upstream dict iteration orders.
    """
    return (
        layer_index(contribution.layer),
        contribution.order,
        contribution.contribution_id,
    )

render_fragments

render_fragments(selection: FragmentSelection) -> str

Render admitted fragments in the legacy SOURCE [x]: y line format.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def render_fragments(selection: FragmentSelection) -> str:
    """Render admitted fragments in the legacy ``SOURCE [x]: y`` line format."""
    return "\n".join(fragment.citation_line() for fragment in selection.admitted)

resolve_scope

resolve_scope(scope: str | None) -> ContextScope

Turn a caller's scope string into a member, or refuse it in-hierarchy.

ContextScope('tenant-a') raises a bare :class:ValueError, which is the one refusal in this capability a caller catching :class:~.errors.BridgeContractError around declaration building would miss. The bridge factories go through here so that every way of getting a declaration wrong reports the same way.

Source code in src/symfonic/capabilities/knowledge/contracts.py

def resolve_scope(scope: str | None) -> ContextScope:
    """Turn a caller's scope string into a member, or refuse it in-hierarchy.

    ``ContextScope('tenant-a')`` raises a bare :class:`ValueError`, which is the
    one refusal in this capability a caller catching
    :class:`~.errors.BridgeContractError` around declaration building would
    miss. The bridge factories go through here so that every way of getting a
    declaration wrong reports the same way.
    """
    if scope is None:
        return ContextScope.DEPLOYMENT
    try:
        return ContextScope(scope)
    except ValueError as exc:
        raise BridgeContractError(
            f"scope {scope!r} is not a context scope; permitted values are "
            f"{[member.value for member in ContextScope]}. A scope decides how widely "
            "one contribution's content is shared, so an unrecognised one is refused "
            "rather than guessed at."
        ) from exc

safe_label

safe_label(raw: str) -> str

Reduce a filename to an opaque display label.

Path structure is discarded rather than escaped: the basename is taken after splitting on both separators, so a POSIX host still refuses a Windows-shaped traversal. Anything that reduces to "", . or .. becomes :data:FALLBACK_LABEL, because a caller that received an empty label would be tempted to substitute the original.

Source code in src/symfonic/capabilities/knowledge/labels.py

def safe_label(raw: str) -> str:
    """Reduce a filename to an opaque display label.

    Path structure is discarded rather than escaped: the basename is taken
    after splitting on *both* separators, so a POSIX host still refuses a
    Windows-shaped traversal. Anything that reduces to ``""``, ``.`` or ``..``
    becomes :data:`FALLBACK_LABEL`, because a caller that received an empty
    label would be tempted to substitute the original.
    """
    cleaned = _CONTROL.sub("", raw)
    basename = cleaned.replace("\\", "/").rpartition("/")[2]
    basename = basename.strip()
    if basename in ("", ".", ".."):
        return FALLBACK_LABEL
    label = _LABEL_DISALLOWED.sub("_", basename)[:MAX_LABEL_CHARS]
    return label or FALLBACK_LABEL

safe_source

safe_source(raw: str) -> str

Reduce a citation source name to something that cannot forge a delimiter.

Spaces survive — a source name is read by a human — but every bracket-ish character is removed. SOURCE [x]: is a delimiter, and a name allowed to contain ] decides where that delimiter ends.

Source code in src/symfonic/capabilities/knowledge/labels.py

def safe_source(raw: str) -> str:
    """Reduce a citation source name to something that cannot forge a delimiter.

    Spaces survive — a source name is read by a human — but every bracket-ish
    character is removed. ``SOURCE [x]:`` is a delimiter, and a name allowed to
    contain ``]`` decides where that delimiter ends.
    """
    cleaned = _CONTROL.sub("", raw)
    cleaned = _SOURCE_DISALLOWED.sub("", cleaned)
    cleaned = _WHITESPACE_RUN.sub(" ", cleaned).strip()
    return cleaned[:MAX_SOURCE_CHARS]

select_fragments

select_fragments(fragments: Iterable[RetrievedFragment], policy: RetrievalPolicy) -> FragmentSelection

Rank, filter, and cap what the store returned. Never mutates the input.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def select_fragments(
    fragments: Iterable[RetrievedFragment], policy: RetrievalPolicy
) -> FragmentSelection:
    """Rank, filter, and cap what the store returned. Never mutates the input."""
    ranked = sorted(fragments, key=_rank_key)
    admitted: list[RetrievedFragment] = []
    dropped: list[tuple[str, str]] = []
    used = 0

    for fragment in ranked:
        if len(admitted) >= policy.limit:
            dropped.append((fragment.source, f"beyond the {policy.limit}-fragment limit"))
            continue
        if fragment.score < policy.min_score:
            dropped.append(
                (
                    fragment.source,
                    f"score {fragment.score} below the {policy.min_score} floor",
                )
            )
            continue
        if len(fragment.content) > policy.max_fragment_chars:
            dropped.append(
                (
                    fragment.source,
                    f"{len(fragment.content)} chars exceeds the per-fragment cap of "
                    f"{policy.max_fragment_chars}; fragments are dropped, never truncated",
                )
            )
            continue
        line = fragment.citation_line()
        cost = len(line) + (1 if admitted else 0)
        if used + cost > policy.max_total_chars:
            dropped.append(
                (
                    fragment.source,
                    f"would take the block past the {policy.max_total_chars}-char ceiling",
                )
            )
            continue
        admitted.append(fragment)
        used += cost

    return FragmentSelection(admitted=tuple(admitted), dropped=tuple(dropped))

selection_revision

selection_revision(selection: FragmentSelection) -> str

A content-derived revision, so a changed retrieval changes the cache key.

Source code in src/symfonic/capabilities/knowledge/retrieval.py

def selection_revision(selection: FragmentSelection) -> str:
    """A content-derived revision, so a changed retrieval changes the cache key."""
    material = "\x00".join(
        f"{fragment.source}\x01{fragment.content}" for fragment in selection.admitted
    )
    return hashlib.sha256(material.encode("utf-8")).hexdigest()[:16]

text_part

text_part(text: str) -> ContentPart

Normalise a text part. Whitespace-only text is refused, not sent.

Source code in src/symfonic/capabilities/knowledge/multimodal.py

def text_part(text: str) -> ContentPart:
    """Normalise a text part. Whitespace-only text is refused, not sent."""
    stripped = text.strip()
    if not stripped:
        raise OutboundRejected(
            "a text part is empty; an empty block costs a request slot and says nothing."
        )
    return ContentPart(part_type="text", text=stripped)

to_anthropic_blocks

to_anthropic_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]

Render parts as Anthropic content blocks.

Source code in src/symfonic/capabilities/knowledge/multimodal.py

def to_anthropic_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]:
    """Render parts as Anthropic content blocks."""
    blocks: list[dict[str, object]] = []
    for part in parts:
        if part.part_type == "text":
            blocks.append({"type": "text", "text": part.text})
            continue
        blocks.append(
            {
                "type": part.part_type,
                "source": {
                    "type": "base64",
                    "media_type": part.media_type,
                    "data": part.data_b64,
                },
            }
        )
    return blocks

to_openai_blocks

to_openai_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]

Render parts as OpenAI content blocks.

A document part is refused rather than approximated. OpenAI's chat content schema has no document block, and inventing one — a data: image URL carrying a PDF, say — produces a request the provider rejects at best and silently misreads at worst.

Source code in src/symfonic/capabilities/knowledge/multimodal.py

def to_openai_blocks(parts: Sequence[ContentPart]) -> list[dict[str, object]]:
    """Render parts as OpenAI content blocks.

    A document part is refused rather than approximated. OpenAI's chat content
    schema has no document block, and inventing one — a ``data:`` image URL
    carrying a PDF, say — produces a request the provider rejects at best and
    silently misreads at worst.
    """
    blocks: list[dict[str, object]] = []
    for part in parts:
        if part.part_type == "text":
            blocks.append({"type": "text", "text": part.text})
            continue
        if part.part_type == "document":
            raise OutboundRejected(
                "the OpenAI content schema has no document block; extract the document to "
                "text and send it as a text part rather than inventing a wire shape."
            )
        blocks.append(
            {
                "type": "image_url",
                "image_url": {"url": f"data:{part.media_type};base64,{part.data_b64}"},
            }
        )
    return blocks

validate_document_id

validate_document_id(document_id: str) -> str

Refuse any id that could be read as a path or a delimiter.

Source code in src/symfonic/capabilities/knowledge/documents.py

def validate_document_id(document_id: str) -> str:
    """Refuse any id that could be read as a path or a delimiter."""
    if not DOCUMENT_ID_CHARSET.match(document_id):
        raise IngestionRejected(
            f"document id {document_id!r} is outside the permitted charset "
            "[A-Za-z0-9_.:-]{1,128}. A document id is an opaque key: untrusted input "
            "must never be able to select a filesystem path (AS-ING-2)."
        )
    if ".." in document_id:
        raise IngestionRejected(
            f"document id {document_id!r} contains '..'; a traversal sequence is never a "
            "legitimate key, whatever the store does with it."
        )
    return document_id