symfonic.capabilities.knowledge.documents

documents

The document bridge: a document store, adapted into one context declaration.

The guard that matters here is small and easy to lose in a rewrite: a document id is a key, and AS-ING-2 says untrusted input must not select a filesystem path. A store keyed by open(f"docs/{document_id}") and an id of ../../etc/passwd is the whole vulnerability, so the id is validated against a charset before the store sees it. This package performs no filesystem access itself; the check exists to protect stores that do.

Titles get the same treatment as citation sources: they arrive with the content, they are interpolated into a rendered delimiter, so they are reduced to labels rather than trusted.

.. note:: :class:DocumentPolicy's ceilings bound 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 rather than truncating. The two are deliberately independent — the bridge does not import the prompting capability — which means a composition root wiring this contribution must raise max_learned_chars to at least the ceiling it sets here. Leave it at its default and a real document compiles to nothing but a diagnostic. See tests/capabilities/knowledge/ test_prompting_seam.py::TestRenderPolicyAlignment.

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.

StoredDocument dataclass

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

One document as a store returns it.

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

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