symfonic.capabilities.knowledge.attachments

attachments

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 dataclass

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

One attachment after extraction, carrying an opaque label.

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.

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,
    )

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),
    )

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"))