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.
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/citations.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
The reference list, one citation per line.
Source code in src/symfonic/capabilities/knowledge/citations.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/citations.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
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
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
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(label: str, kind: str, media_type: str, text: str, byte_size: int)
One attachment after extraction, carrying an opaque label.
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(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
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
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.
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/citations.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(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(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(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
|