symfonic.tools.corpus_scrub

corpus_scrub

symfonic.tools.corpus_scrub -- PII redaction for Phase 12 corpus exports.

Public adopter-facing module for scrubbing real-production episodic data BEFORE it leaves the adopter's environment. Required for the v7.9.0 LLM-extractor flip: the symfonic-side eval harness consumes scrubbed JSONL files; raw production data MUST NOT cross the boundary.

Design contract (v7.8.0):

• Scrubbing happens ADOPTER-SIDE. symfonic-core ships the regex patterns + tokeniser; the adopter runs them locally and uploads only the scrubbed output.
• Opaque-PII tokens are STABLE per-export: the same input string always maps to the same <REDACTED-EMAIL-N> within one corpus so the LLM-extractor can still learn co-occurrence patterns ("user mentions twice in a conversation").
• Drop-record-on-doubt: rows with more than unresolved_threshold unresolved matches (suspicious values that look like PII but don't fit a known pattern) are dropped from the output.
• Amazon-specific: ASINs and marketplace-IDs are STRUCTURAL identifiers (not PII) and are KEPT. Seller-IDs and account-IDs are PII and are REDACTED.

Usage::

from symfonic.tools.corpus_scrub import CorpusScrubber, scrub_jsonl_file

scrubber = CorpusScrubber()
scrub_jsonl_file(
    input_path=Path("dryrun_corpus.jsonl"),
    output_path=Path("dryrun_corpus.scrubbed.jsonl.gz"),
    scrubber=scrubber,
)

Or via the CLI::

python -m symfonic.tools.corpus_scrub \
    --input dryrun_corpus.jsonl \
    --output dryrun_corpus.scrubbed.jsonl.gz

v7.8.0 (Jarvio v7.9.0 corpus-flip ask).

CorpusScrubber

CorpusScrubber(
    *,
    value_credential_patterns: Iterable[
        tuple[str, Pattern[str]]
    ] = DEFAULT_VALUE_CREDENTIAL_PATTERNS,
    pii_patterns: Iterable[
        tuple[str, Pattern[str]]
    ] = DEFAULT_PII_PATTERNS,
    amazon_keep_patterns: Iterable[
        tuple[str, Pattern[str]]
    ] = AMAZON_KEEP_PATTERNS,
    amazon_redact_patterns: Iterable[
        tuple[str, Pattern[str]]
    ] = AMAZON_REDACT_PATTERNS,
    unresolved_threshold: int = 3,
)

Apply redaction rules to a corpus row.

Token-mapping state is instance-scoped. Construct one scrubber per corpus export so co-occurrence tokens (<REDACTED-EMAIL-1>) are stable within the export and not leaked across exports.

Parameters:

Name	Type	Description	Default
value_credential_patterns	Iterable[tuple[str, Pattern[str]]]	Credential-shaped substring patterns applied to VALUES. Default: :data:DEFAULT_VALUE_CREDENTIAL_PATTERNS.	DEFAULT_VALUE_CREDENTIAL_PATTERNS
pii_patterns	Iterable[tuple[str, Pattern[str]]]	Generic-PII patterns. Default: :data:DEFAULT_PII_PATTERNS.	DEFAULT_PII_PATTERNS
amazon_keep_patterns	Iterable[tuple[str, Pattern[str]]]	Amazon structural identifiers to KEEP as-is. Default: :data:AMAZON_KEEP_PATTERNS.	AMAZON_KEEP_PATTERNS
amazon_redact_patterns	Iterable[tuple[str, Pattern[str]]]	Amazon account-id patterns to REDACT. Default: :data:AMAZON_REDACT_PATTERNS.	AMAZON_REDACT_PATTERNS
unresolved_threshold	int	Maximum unresolved substring count allowed per row before scrub_row(drop_on_doubt=True) drops the row. Default 3 per architect spec.	3

Source code in src/symfonic/tools/corpus_scrub/scrubber.py

def __init__(
    self,
    *,
    value_credential_patterns: Iterable[
        tuple[str, re.Pattern[str]]
    ] = DEFAULT_VALUE_CREDENTIAL_PATTERNS,
    pii_patterns: Iterable[
        tuple[str, re.Pattern[str]]
    ] = DEFAULT_PII_PATTERNS,
    amazon_keep_patterns: Iterable[
        tuple[str, re.Pattern[str]]
    ] = AMAZON_KEEP_PATTERNS,
    amazon_redact_patterns: Iterable[
        tuple[str, re.Pattern[str]]
    ] = AMAZON_REDACT_PATTERNS,
    unresolved_threshold: int = 3,
) -> None:
    # Order matters: credentials FIRST (highest sensitivity),
    # then Amazon-redact (account-ids), then generic PII.  Keep-
    # patterns are checked in scrub_string to mask spans before
    # any redaction touches them.
    self._redact_patterns: list[tuple[str, re.Pattern[str]]] = [
        *value_credential_patterns,
        *amazon_redact_patterns,
        *pii_patterns,
    ]
    self._keep_patterns: list[tuple[str, re.Pattern[str]]] = list(
        amazon_keep_patterns,
    )
    self._unresolved_threshold = unresolved_threshold
    # Stable per-label token mapping: ``{label: {raw: token}}``.
    self._token_map: dict[str, dict[str, str]] = {}

scrub_row

scrub_row(
    row: Any, *, drop_on_doubt: bool = True
) -> ScrubResult

Scrub a single corpus row.

When drop_on_doubt is True and the unresolved count exceeds unresolved_threshold, the row is dropped and ScrubResult.scrubbed is None.

Parameters:

Name	Type	Description	Default
row	Any	The JSON-shaped row to scrub. Typically a dict.	required
drop_on_doubt	bool	When True (default), enforce unresolved_threshold. Set False to keep all rows with their original-or-scrubbed content for analysis.	True

Source code in src/symfonic/tools/corpus_scrub/scrubber.py

def scrub_row(
    self, row: Any, *, drop_on_doubt: bool = True,
) -> ScrubResult:
    """Scrub a single corpus row.

    When ``drop_on_doubt`` is True and the unresolved count
    exceeds ``unresolved_threshold``, the row is dropped and
    ``ScrubResult.scrubbed`` is ``None``.

    Args:
        row: The JSON-shaped row to scrub.  Typically a dict.
        drop_on_doubt: When True (default), enforce
            ``unresolved_threshold``.  Set False to keep all rows
            with their original-or-scrubbed content for analysis.
    """
    scrubbed, counts, unresolved = self.scrub_value(row)
    if drop_on_doubt and unresolved > self._unresolved_threshold:
        return ScrubResult(
            scrubbed=None,
            match_counts=counts,
            unresolved=unresolved,
            dropped_reason=(
                f"unresolved={unresolved} > threshold="
                f"{self._unresolved_threshold}"
            ),
        )
    return ScrubResult(
        scrubbed=scrubbed,
        match_counts=counts,
        unresolved=unresolved,
    )

scrub_string

scrub_string(text: str) -> tuple[str, dict[str, int], int]

Redact a single string.

Returns:

Type	Description
tuple[str, dict[str, int], int]	(scrubbed_text, match_counts, unresolved_count).

Algorithm

1. Find all keep-pattern spans; mark them as protected.
2. For each redact pattern, replace matches OUTSIDE the protected spans with stable per-label tokens.
3. Unresolved count is held at 0 in v7.8.0 -- adopters who want heuristic unresolved detection subclass and override.

Source code in src/symfonic/tools/corpus_scrub/scrubber.py

def scrub_string(
    self, text: str,
) -> tuple[str, dict[str, int], int]:
    """Redact a single string.

    Returns:
        ``(scrubbed_text, match_counts, unresolved_count)``.

    Algorithm:
        1. Find all keep-pattern spans; mark them as protected.
        2. For each redact pattern, replace matches OUTSIDE the
           protected spans with stable per-label tokens.
        3. Unresolved count is held at 0 in v7.8.0 -- adopters
           who want heuristic unresolved detection subclass and
           override.
    """
    if not isinstance(text, str) or not text:
        return text, {}, 0

    # Step 1: collect keep-pattern spans.  These spans bypass
    # redaction so ASINs and marketplace-ids stay intact.
    protected: list[tuple[int, int]] = []
    for _label, regex in self._keep_patterns:
        for match in regex.finditer(text):
            protected.append(match.span())
    # Sort + merge overlapping spans for fast O(log n) lookup.
    protected.sort()
    merged: list[tuple[int, int]] = []
    for start, end in protected:
        if merged and start <= merged[-1][1]:
            merged[-1] = (merged[-1][0], max(merged[-1][1], end))
        else:
            merged.append((start, end))

    def _is_protected(start: int, end: int) -> bool:
        return any(start >= ps and end <= pe for ps, pe in merged)

    # Step 2: apply redact patterns in order.  Each pass mutates
    # the working text; offsets must be recomputed per pass.
    match_counts: dict[str, int] = {}
    working = text
    for label, regex in self._redact_patterns:
        # Snapshot the current protected spans against the WORKING
        # text by re-scanning the keep-patterns.  Cheap because
        # the working text is bounded by row size.
        current_protected: list[tuple[int, int]] = []
        for _, krg in self._keep_patterns:
            for km in krg.finditer(working):
                current_protected.append(km.span())
        current_protected.sort()

        # B023 false positive: both closures are consumed synchronously
        # by ``regex.sub`` within this iteration, so late-binding of
        # ``current_protected`` and ``label`` is benign.
        def _protected_now(s: int, e: int, cp=current_protected) -> bool:
            return any(s >= ps and e <= pe for ps, pe in cp)

        def _replace(match: re.Match[str], lbl: str = label) -> str:
            if _protected_now(*match.span()):
                return match.group(0)
            raw = match.group(0)
            match_counts[lbl] = match_counts.get(lbl, 0) + 1
            return self._next_token(lbl, raw)

        working = regex.sub(_replace, working)

    # Step 3: unresolved heuristic -- conservative default of 0.
    unresolved = 0
    return working, match_counts, unresolved

scrub_value

scrub_value(value: Any) -> tuple[Any, dict[str, int], int]

Recursively scrub a JSON-shaped value.

Strings are passed to :meth:scrub_string. Dicts and lists recurse. Other types (int, float, bool, None) pass through.

Source code in src/symfonic/tools/corpus_scrub/scrubber.py

def scrub_value(
    self, value: Any,
) -> tuple[Any, dict[str, int], int]:
    """Recursively scrub a JSON-shaped value.

    Strings are passed to :meth:`scrub_string`.  Dicts and lists
    recurse.  Other types (int, float, bool, None) pass through.
    """
    if isinstance(value, str):
        return self.scrub_string(value)
    if isinstance(value, dict):
        counts: dict[str, int] = {}
        unresolved = 0
        out: dict[str, Any] = {}
        for k, v in value.items():
            sv, sc, su = self.scrub_value(v)
            out[k] = sv
            for lk, lv in sc.items():
                counts[lk] = counts.get(lk, 0) + lv
            unresolved += su
        return out, counts, unresolved
    if isinstance(value, list):
        counts2: dict[str, int] = {}
        unresolved2 = 0
        out_list: list[Any] = []
        for item in value:
            sv2, sc2, su2 = self.scrub_value(item)
            out_list.append(sv2)
            for lk2, lv2 in sc2.items():
                counts2[lk2] = counts2.get(lk2, 0) + lv2
            unresolved2 += su2
        return out_list, counts2, unresolved2
    return value, {}, 0

ScrubResult dataclass

ScrubResult(
    scrubbed: Any | None,
    match_counts: dict[str, int] = dict(),
    unresolved: int = 0,
    dropped_reason: str | None = None,
)

Per-row scrub outcome.

Attributes:

Name	Type	Description
scrubbed	Any | None	The redacted row, or None if dropped.
match_counts	dict[str, int]	Mapping label -> count of matches by label.
unresolved	int	Count of unresolved suspicious substrings (drop threshold check is the caller's responsibility -- the scrubber returns the count but only enforces threshold on scrub_row when drop_on_doubt=True).
dropped_reason	str | None	Human-readable reason when scrubbed is None.

scrub_jsonl_file

scrub_jsonl_file(
    *,
    input_path: Path,
    output_path: Path,
    scrubber: CorpusScrubber | None = None,
    drop_on_doubt: bool = True,
) -> dict[str, Any]

Stream a JSONL file through :class:CorpusScrubber.

Input may be plain .jsonl or gzipped .jsonl.gz. Output extension determines compression (.gz -> gzip). Per architect spec, the canonical output shape is .jsonl.gz so the file is transferable without further compression.

Returns a summary dict

{ "input_rows": int, "output_rows": int, "dropped_rows": int, "match_counts": dict[str, int], }

The summary is also the natural shape for a manifest sidecar.

Source code in src/symfonic/tools/corpus_scrub/scrubber.py

def scrub_jsonl_file(
    *,
    input_path: Path,
    output_path: Path,
    scrubber: CorpusScrubber | None = None,
    drop_on_doubt: bool = True,
) -> dict[str, Any]:
    """Stream a JSONL file through :class:`CorpusScrubber`.

    Input may be plain ``.jsonl`` or gzipped ``.jsonl.gz``.  Output
    extension determines compression (``.gz`` -> gzip).  Per architect
    spec, the canonical output shape is ``.jsonl.gz`` so the file is
    transferable without further compression.

    Returns a summary dict:
        ``{
            "input_rows": int,
            "output_rows": int,
            "dropped_rows": int,
            "match_counts": dict[str, int],
        }``

    The summary is also the natural shape for a manifest sidecar.
    """
    if scrubber is None:
        scrubber = CorpusScrubber()

    input_open = gzip.open if input_path.suffix == ".gz" else open
    output_open = gzip.open if output_path.suffix == ".gz" else open

    input_rows = 0
    output_rows = 0
    dropped_rows = 0
    total_counts: dict[str, int] = {}

    with input_open(input_path, "rt", encoding="utf-8") as f_in, \
            output_open(output_path, "wt", encoding="utf-8") as f_out:
        for line in f_in:
            line = line.strip()
            if not line:
                continue
            input_rows += 1
            row = json.loads(line)
            result = scrubber.scrub_row(row, drop_on_doubt=drop_on_doubt)
            for k, v in result.match_counts.items():
                total_counts[k] = total_counts.get(k, 0) + v
            if result.scrubbed is None:
                dropped_rows += 1
                continue
            f_out.write(json.dumps(result.scrubbed, ensure_ascii=False))
            f_out.write("\n")
            output_rows += 1

    return {
        "input_rows": input_rows,
        "output_rows": output_rows,
        "dropped_rows": dropped_rows,
        "match_counts": total_counts,
    }