symfonic.capabilities.memory.scrubbing

scrubbing

Credential scrubbing, applied before a memory exists rather than after.

Two scans, because a secret arrives two ways and only one of them is covered by the shipped scrubber:

• Key names — symfonic.agent.hygiene drops properties keys whose names look like credentials (api_key, authorization). That list is restated here (the capability layer imports nothing outside itself) and pinned to the shipped one by a test, because a drift silently un-redacts every future write.
• Values — the shipped hygiene module explicitly leaves free-form value scanning out of scope, so a bearer token pasted into a sentence is stored verbatim and then recalled into a prompt on some later turn. The value patterns here are the corpus-scrub set (symfonic.tools.corpus_scrub), which already ships in this repository for exactly these shapes.

Placement is the point. A memory is scrubbed before the record is minted, so the write port never receives a credential — there is no window in which a secret is persisted and a later pass has to find it again. The same scrubber runs over the turn before it reaches the extraction model, because a secret handed to a provider is disclosed whether or not it is ever stored.

Redaction replaces rather than deletes: [REDACTED:TOKEN] keeps the sentence readable, which is what makes the surrounding memory still worth recalling, and makes scrubbing idempotent — a redacted text scrubs to itself.

CredentialScrubber

CredentialScrubber(*, value_patterns: Iterable[tuple[str, Pattern[str]]] | None = None, key_parts: Iterable[str] | None = None)

Removes credentials from memory text and from memory properties.

Build a scrubber.

None selects the built-in set; an empty iterable disables that scan. The distinction is deliberate and matches the shipped hygiene contract: switching a scrubber off is something a deployment must say, not something it can fall into by passing an empty config.

Source code in src/symfonic/capabilities/memory/scrubbing.py

def __init__(
    self,
    *,
    value_patterns: Iterable[tuple[str, re.Pattern[str]]] | None = None,
    key_parts: Iterable[str] | None = None,
) -> None:
    """Build a scrubber.

    ``None`` selects the built-in set; an **empty** iterable disables that
    scan. The distinction is deliberate and matches the shipped hygiene
    contract: switching a scrubber off is something a deployment must say,
    not something it can fall into by passing an empty config.
    """
    patterns = (
        CREDENTIAL_VALUE_PATTERNS if value_patterns is None else tuple(value_patterns)
    )
    self._value_patterns = patterns
    parts = (
        DEFAULT_CREDENTIAL_KEY_PARTS if key_parts is None else tuple(key_parts)
    )
    self._key_pattern = (
        re.compile("(?i)(" + "|".join(f"(?:{part})" for part in parts) + ")")
        if parts
        else None
    )

scrub_properties

scrub_properties(properties: Mapping[str, object]) -> tuple[dict[str, object], tuple[str, ...]]

Drop credential-named keys, and scrub the string values that remain.

Shallow, like the shipped scrubber: graph properties are persisted flat, so a nested dict is not a shape any backend writes.

Source code in src/symfonic/capabilities/memory/scrubbing.py

def scrub_properties(
    self, properties: Mapping[str, object]
) -> tuple[dict[str, object], tuple[str, ...]]:
    """Drop credential-named keys, and scrub the string values that remain.

    Shallow, like the shipped scrubber: graph properties are persisted
    flat, so a nested dict is not a shape any backend writes.
    """
    clean: dict[str, object] = {}
    dropped: list[str] = []
    for key, value in properties.items():
        if self._key_pattern is not None and self._key_pattern.search(str(key)):
            dropped.append(key)
            continue
        clean[key] = (
            self.scrub_text(value).text if isinstance(value, str) else value
        )
    return clean, tuple(dropped)

scrub_text

scrub_text(text: str) -> ScrubResult

Replace every credential-shaped value in text.

Source code in src/symfonic/capabilities/memory/scrubbing.py

def scrub_text(self, text: str) -> ScrubResult:
    """Replace every credential-shaped value in ``text``."""
    redactions: list[str] = []
    scrubbed = text
    for label, pattern in self._value_patterns:
        replacement = REDACTION_TEMPLATE.format(label=label)
        scrubbed, count = pattern.subn(replacement, scrubbed)
        redactions.extend([label] * count)
    return ScrubResult(text=scrubbed, redactions=tuple(redactions))

ScrubResult dataclass

ScrubResult(text: str, redactions: tuple[str, ...] = ())

Text with its credentials replaced, and what was replaced.

clean property

clean: bool

Whether the input carried no credential-shaped value.