The attachment bridge: file bytes, extractors, and OCR behind one port.
The extractors themselves stay where they are. What moves here is the part that
was never really theirs: the decision about what is allowed to reach them and
what is allowed to leave. Format parsers are assumed compromise-prone
(AS-ING-5), so this module treats one as a hostile black box behind a
single-method port — bounded on the way in, exception-wrapped on the way out,
and never handed a filesystem path or a credential.
Input bytes and zip declarations are checked before dispatch (AS-ING-3/6).
Raw bytes limit upload size; declared size, ratio, member count, and nesting
bound Office zips. The final text ceiling applies across all attachments.
.. note::
:data:ExtractionLimits.max_total_chars bounds what this bridge emits;
the prompt compiler applies its own, much smaller, learned-content cap
(RenderPolicy.max_learned_chars, 500 by default) and drops any block
above it. A composition root wiring this contribution must raise that cap to
at least the ceiling it sets here, or every attachment set will compile to
nothing but a diagnostic.
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),
)
|
ExtractedAttachment(label: str, kind: str, media_type: str, text: str, byte_size: int)
One attachment after extraction, carrying an opaque label.
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.
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,
)
|
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),
)
|
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"))
|