Detectors & backends

Extension points and the layer internals.

Extension points

wardcat.register_action

register_action(name, fn)

Register (or override) an anonymization action under name.

Source code in src/wardcat/core/actions.py

def register_action(name: str, fn: ActionFn) -> None:
    """Register (or override) an anonymization action under *name*."""
    _ACTIONS[name] = fn

wardcat.ActionContext dataclass

ActionContext(
    salt="",
    tokens=TokenAllocator(),
    locale="en",
    surrogates=_default_surrogates(),
)

Extra context an action may need beyond the span (e.g. the hash salt).

tokens class-attribute instance-attribute

tokens = field(
    default_factory=TokenAllocator,
    repr=False,
    compare=False,
)

Placeholder vault for reversible actions, scoped to a single scan — its context_id is the one stamped into this scan's tokens. Excluded from repr/equality so two contexts with the same salt still compare equal.

locale class-attribute instance-attribute

locale = 'en'

Locale the surrogate action draws names and formats from.

surrogates class-attribute instance-attribute

surrogates = field(
    default_factory=_default_surrogates,
    repr=False,
    compare=False,
)

Surrogate vault for the surrogate action, scoped to a single scan.

wardcat.TokenAllocator

TokenAllocator(context_id=None)

Hands out stable, unique placeholders for one scan — the tokenize vault.

A placeholder is [TYPE_index_contextid] — [EMAIL_1_9f3a2c8b71d4]. The index is per entity type in order of first appearance and restarts at 1 for every scan, so it stays short and readable; the context id is drawn once per allocator and shared by every token it hands out, which is what makes one scan's placeholders distinct from another's. An identical value always gets the same token, so a name repeated three times stays one referent for whatever reads the anonymized text.

That distinctness is the safety property behind :meth:~wardcat.ScanResult.restore: two scans running side by side both hold an "EMAIL number 1", and without the context id their placeholders would be the same string — restoring one request's answer against another's result would silently substitute the wrong person's value. With it there is nothing to match, so the mistake becomes a reported non-substitution instead.

Allocation state is per instance and the :class:Anonymizer <wardcat.core.anonymizer.Anonymizer> builds a fresh one for every apply() call, so concurrent scans never share a counter or an id.

Parameters:

Name	Type	Description	Default
context_id	str | None	the id to stamp into every token. Defaults to a fresh one; pass "" for bare [TYPE_index] placeholders (no cross-scan protection), or a fixed value to make output reproducible in tests. .. warning:: Holds the raw values it has seen for the lifetime of the instance.	None

Source code in src/wardcat/core/actions.py

def __init__(self, context_id: str | None = None) -> None:
    self.context_id = new_context_id() if context_id is None else context_id
    self._tokens: dict[tuple[str, str], str] = {}
    self._counts: dict[str, int] = {}

token_for

token_for(entity_type, text)

Return the placeholder for text, allocating a new one on first sight.

Source code in src/wardcat/core/actions.py

def token_for(self, entity_type: str, text: str) -> str:
    """Return the placeholder for *text*, allocating a new one on first sight."""
    key = (entity_type, text)
    token = self._tokens.get(key)
    if token is None:
        count = self._counts.get(entity_type, 0) + 1
        self._counts[entity_type] = count
        suffix = f"_{self.context_id}" if self.context_id else ""
        token = f"[{entity_type}_{count}{suffix}]"
        self._tokens[key] = token
    return token

Detector interface

wardcat.detectors.base.BaseDetector

Bases: ABC

Interface that all detectors implement.

The engine talks to detectors only through this interface — it never imports a concrete detector. Two optional capabilities are expressed on the base so the engine stays decoupled:

• can_adjudicate — when True the engine routes the other detectors' spans to this detector via the candidates argument (ensemble mode).
• detect_async — a default thread-based implementation is provided; I/O-bound detectors (e.g. an LLM backend) override it with native async.

detect abstractmethod

detect(text, candidates=None)

Scan text and return the spans found.

Parameters:

Name	Type	Description	Default
candidates	list[DetectedSpan] | None	spans found by the other detectors. Only meaningful for adjudicating detectors (can_adjudicate=True); others ignore it.	None

Source code in src/wardcat/detectors/base.py

@abstractmethod
def detect(self, text: str, candidates: list[DetectedSpan] | None = None) -> list[DetectedSpan]:
    """Scan *text* and return the spans found.

    :param candidates: spans found by the other detectors. Only meaningful for
        adjudicating detectors (``can_adjudicate=True``); others ignore it.
    """
    ...

detect_many

detect_many(texts)

Scan several texts; the default is one :meth:detect per text.

A detector whose model runs faster over a batch — SpaCy's nlp.pipe, say — overrides this. The result has one span list per input, in order.

Source code in src/wardcat/detectors/base.py

def detect_many(self, texts: list[str]) -> list[list[DetectedSpan]]:
    """Scan several texts; the default is one :meth:`detect` per text.

    A detector whose model runs faster over a batch — SpaCy's ``nlp.pipe``,
    say — overrides this. The result has one span list per input, in order.
    """
    return [self.detect(text) for text in texts]

detect_async async

detect_async(text, candidates=None)

Async variant of :meth:detect.

Default implementation offloads the synchronous :meth:detect to a thread. I/O-bound detectors should override this with native async I/O.

Source code in src/wardcat/detectors/base.py

async def detect_async(
    self, text: str, candidates: list[DetectedSpan] | None = None
) -> list[DetectedSpan]:
    """Async variant of :meth:`detect`.

    Default implementation offloads the synchronous :meth:`detect` to a
    thread. I/O-bound detectors should override this with native async I/O.
    """
    return await asyncio.to_thread(self.detect, text, candidates)

wardcat.detectors.base.DetectedSpan dataclass

DetectedSpan(
    entity_type, text, start, end, confidence=1.0, source=""
)

A single detected sensitive data span.

entity_type instance-attribute

entity_type

Entity type — e.g. "EMAIL", "CREDIT_CARD".

text instance-attribute

text

Full text copied from the original.

start instance-attribute

start

Start index in the original text (inclusive).

end instance-attribute

end

End index in the original text (exclusive).

confidence class-attribute instance-attribute

confidence = 1.0

Detection confidence in [0.0, 1.0]. Regex/checksum detections are 1.0; NER and LLM detections are 0.85 (model-based, not fully deterministic).

source class-attribute instance-attribute

source = ''

The layer that produced the span. A detector may leave it empty; the engine then stamps :attr:BaseDetector.layer.

LLM backends

The built-in backends are selected with the Backend enum (not user-extensible).

wardcat.Backend

Bases: str, Enum

LLM backend types, as constants — for typo-proof selection.

Pass these to :meth:Wardcat.with_llm instead of bare strings::

from wardcat import Wardcat, Backend

Wardcat(salt="s").with_llm(backend=Backend.OPENAI_COMPATIBLE, model="...")

Each member is its string value (Backend.OLLAMA == "ollama"), so the plain string form is still accepted.

OLLAMA class-attribute instance-attribute

OLLAMA = 'ollama'

Local Ollama service (supports model download).

OPENAI_COMPATIBLE class-attribute instance-attribute

OPENAI_COMPATIBLE = 'openai_compatible'

OpenAI-compatible HTTP API — LM Studio, LocalAI, LiteLLM, …

VLLM class-attribute instance-attribute

VLLM = 'vllm'

vLLM server (OpenAI-compatible API) — native chat, vLLM defaults.

TRANSFORMERS class-attribute instance-attribute

TRANSFORMERS = 'transformers'

In-process HuggingFace Transformers (no HTTP; loads the model locally).

Engine & anonymizer

wardcat.core.engine.DetectionEngine

DetectionEngine(config, detectors, *, build_warnings=())

Merges spans from all detectors, resolves overlaps, applies configured actions, and returns a ScanResult.

Source code in src/wardcat/core/engine.py

def __init__(
    self,
    config: dict[str, Any],
    detectors: list[BaseDetector],
    *,
    build_warnings: Sequence[str] = (),
) -> None:
    self.config = config
    self.detectors = detectors
    # Ensemble adjudication: when enabled and an LLM detector is present,
    # regex/NER spans are passed to the LLM as candidates to verify/relabel/
    # drop (one combined detection + adjudication call). Every regex span
    # (confidence >= _ADJUDICATION_KEEP_CONFIDENCE) is always kept regardless
    # of the LLM — a weak adjudicator that fails to re-detect a real match
    # would otherwise leak it. Only the model layers (NER/LLM) are
    # candidates the LLM may drop/relabel.
    # Detectors are addressed only through BaseDetector — the engine never
    # imports a concrete detector. Adjudicators advertise themselves via the
    # can_adjudicate flag.
    self._adjudicators = [d for d in detectors if d.can_adjudicate]
    self._other_detectors = [d for d in detectors if not d.can_adjudicate]
    self._use_adjudication = bool(
        config.get("llm_detector", {}).get("adjudicate", False)
    ) and bool(self._adjudicators)
    self.salt: str = config.get("salt", "")
    self.entity_config: dict[str, Any] = config.get("entities", {})
    self._max_text_bytes: int = config.get("max_text_bytes", _MAX_TEXT_BYTES)
    self._allowlist: set[str] = set(config.get("allowlist", []))
    # Confidence floor. A span scoring below it is dropped before any action
    # is applied. The default sits above the "uncued" tier the regex layer
    # gives a checksum match with no supporting keyword, and below every
    # other tier, so lowering it trades precision for recall and nothing
    # else changes.
    self._min_confidence: float = float(config.get("min_confidence", 0.8))
    # Strict: a scan that covered less than was configured is an error, not a
    # result with warnings. The guard already refuses to build in that state;
    # this catches what only shows up per scan (an LLM backend going down).
    self._strict: bool = bool(config.get("strict", False))
    # Denylist entries, in configured order, as (entity_type, value or compiled
    # pattern). Compiled once here rather than on every scan. Patterns were
    # screened for catastrophic backtracking when they were configured.
    self._denylist: list[tuple[str, re.Pattern[str]]] = []
    # Literal entries are matched in one pass, whatever their number: a
    # customer list of ten thousand names would otherwise be ten thousand
    # passes over every text. One alternation, longest value first so the
    # longest literal at a position wins, wrapped in a lookahead so every
    # start position is reported — exactly what the entry-by-entry search
    # found, and what overlap resolution expects. Regex entries stay
    # separate: merging them would renumber their groups and let one
    # entry's backtracking reach into another's.
    self._denylist_literals: dict[str, str] = {}
    self._denylist_literal_re: re.Pattern[str] | None = None
    denylist_warnings: list[str] = []
    for entry in config.get("denylist", []):
        entity_type = entry.get("entity_type", "CUSTOM")
        if "pattern" in entry:
            try:
                self._denylist.append((entity_type, re.compile(entry["pattern"])))
            except re.error as exc:
                denylist_warnings.append(
                    f"Denylist pattern {entry['pattern']!r} is not valid regex and was "
                    f"skipped: {exc}"
                )
        elif entry.get("value"):
            value = entry["value"]
            first_type = self._denylist_literals.get(value)
            if first_type is not None:
                if first_type != entity_type:
                    logger.warning(
                        "Denylist value listed under two entity types (%s and %s); "
                        "the first one is used.",
                        first_type,
                        entity_type,
                    )
                continue
            self._denylist_literals[value] = entity_type
    # Every literal that matches at a position is a prefix of the longest one
    # matching there, so the shorter ones are recovered from a prefix table
    # rather than searched for. Overlap resolution then sees the same
    # candidates the entry-by-entry search produced.
    self._denylist_prefixes: dict[str, list[str]] = {}
    if self._denylist_literals:
        ordered = sorted(self._denylist_literals, key=len, reverse=True)
        self._denylist_literal_re = re.compile(
            "(?=(" + "|".join(re.escape(v) for v in ordered) + "))"
        )
        for value in ordered:
            self._denylist_prefixes[value] = [
                value[:k] for k in range(1, len(value)) if value[:k] in self._denylist_literals
            ]
    # Problems found while the guard was built that leave every scan covering
    # less than was configured — an NER model that did not load, say. A log
    # line is read once if at all; the result is what a caller checks, so each
    # result carries them, since each scan really was incomplete.
    self.build_warnings: tuple[str, ...] = (
        *build_warnings,
        *(warning for detector in detectors for warning in detector.build_warnings),
        *denylist_warnings,
    )
    # Value propagation: once any layer detects a value, redact every other
    # whole-token occurrence of that exact value too. Closes the gap where a
    # model-based layer (NER/LLM) reports a repeated value only once.
    # Opt-in — it can over-redact, so short values are skipped and matches
    # must be token-bounded.
    self._propagate: bool = bool(config.get("propagate_matches", False))
    self._propagate_min_len: int = config.get("propagate_min_length", 3)
    # Detection (this class) is kept separate from anonymization (applying the
    # configured action to each span); the Anonymizer owns that stage.
    self._locale: str = config.get("locale", "en")
    self._anonymizer = Anonymizer(self.entity_config, self.salt, self._locale)

    if not self.salt:
        logger.debug(
            "Hash salt is empty — identical values will produce the same hash. "
            "Pass salt=... to Wardcat(...) in production."
        )

scan

scan(text)

Run all detectors, apply actions, and return the result.

Source code in src/wardcat/core/engine.py

def scan(self, text: str) -> ScanResult:
    """Run all detectors, apply actions, and return the result."""
    t_start = time.perf_counter()
    self._check_size(text)

    warnings: list[str] = list(self.build_warnings)
    if self._use_adjudication:
        candidate_spans: list[DetectedSpan] = []
        for detector in self._other_detectors:
            candidate_spans.extend(self._safe_detect(detector, text, warnings))
        raw_spans = [
            s for s in candidate_spans if s.confidence >= _ADJUDICATION_KEEP_CONFIDENCE
        ]
        for adjudicator in self._adjudicators:
            raw_spans.extend(
                self._safe_detect(adjudicator, text, warnings, candidates=candidate_spans)
            )
    else:
        raw_spans = []
        for detector in self.detectors:
            raw_spans.extend(self._safe_detect(detector, text, warnings))

    raw_spans.extend(self._collect_denylist_spans(text))
    spans = self._filter_spans(raw_spans, text)
    # One context id per scan — it is what keeps this scan's reversible
    # placeholders distinct from every other scan's (see TokenAllocator).
    context_id = new_context_id()
    sanitized, violations = self._anonymizer.apply(text, spans, context_id=context_id)

    elapsed_ms = (time.perf_counter() - t_start) * 1000
    logger.info(
        "scan completed: %d violation(s), %d character(s), %.1f ms",
        len(violations),
        len(text),
        elapsed_ms,
    )
    result = ScanResult(
        original_text=text,
        sanitized_text=sanitized,
        violations=violations,
        warnings=warnings,
        context_id=context_id,
        _salt=self.salt,
        _locale=self._locale,
    )
    if self._strict and warnings:
        raise DegradedScanError(warnings, result)
    return result

scan_many

scan_many(texts, *, workers=4)

Scan a batch: batched layers in one pass each, the rest in threads.

The result list matches texts in order and equals what :meth:scan would return for each, apart from context ids. A detector that batches (:meth:~wardcat.detectors.base.BaseDetector.detect_many, which SpaCy NER implements with nlp.pipe) sees the whole batch at once; the others — the LLM layer — run per text in a pool of workers threads. A layer that fails is skipped for the whole batch and recorded on every result. A text that cannot be scanned at all (too large) gets a result with scan_error set and its text unchanged. Adjudication is not batched: with it on, each text goes through :meth:scan.

Source code in src/wardcat/core/engine.py

def scan_many(self, texts: list[str], *, workers: int = 4) -> list[ScanResult]:
    """Scan a batch: batched layers in one pass each, the rest in threads.

    The result list matches *texts* in order and equals what :meth:`scan`
    would return for each, apart from context ids. A detector that batches
    (:meth:`~wardcat.detectors.base.BaseDetector.detect_many`, which SpaCy
    NER implements with ``nlp.pipe``) sees the whole batch at once; the
    others — the LLM layer — run per text in a pool of *workers* threads.
    A layer that fails is skipped for the whole batch and recorded on every
    result. A text that cannot be scanned at all (too large) gets a result
    with ``scan_error`` set and its text unchanged. Adjudication is not
    batched: with it on, each text goes through :meth:`scan`.
    """
    if not texts:
        return []
    if self._use_adjudication:
        return [self._scan_or_error(text) for text in texts]

    results: list[ScanResult | None] = [None] * len(texts)
    todo: list[int] = []
    for i, text in enumerate(texts):
        try:
            self._check_size(text)
        except ValueError as exc:
            results[i] = self._error_result(text, exc)
            continue
        todo.append(i)
    if not todo:
        return results  # type: ignore[return-value]

    warnings: list[str] = list(self.build_warnings)
    batch = [texts[i] for i in todo]
    per_text: list[list[DetectedSpan]] = [[] for _ in todo]
    pooled: list[BaseDetector] = []
    for detector in self.detectors:
        if type(detector).detect_many is BaseDetector.detect_many:
            pooled.append(detector)
            continue
        try:
            found = detector.detect_many(batch)
        except Exception as exc:
            msg = f"{type(detector).__name__} did not run: {exc}"
            logger.warning("scan_many: detector layer skipped — %s", msg)
            warnings.append(msg)
            continue
        for spans, acc in zip(found, per_text, strict=True):
            acc.extend(self._stamp_source(spans, detector))

    # Detectors without a batch path — the LLM layer — still run per text,
    # but in parallel; its own gate bounds what reaches the backend.
    if pooled:
        from concurrent.futures import ThreadPoolExecutor

        failed: dict[int, list[str]] = {}

        def one(k: int) -> list[DetectedSpan]:
            local: list[str] = []
            spans: list[DetectedSpan] = []
            for detector in pooled:
                spans.extend(self._safe_detect(detector, batch[k], local))
            if local:
                failed[k] = local
            return spans

        with ThreadPoolExecutor(max_workers=max(1, workers)) as pool:
            for k, spans in enumerate(pool.map(one, range(len(batch)))):
                per_text[k].extend(spans)
    else:
        failed = {}

    for k, i in enumerate(todo):
        text = texts[i]
        item_warnings = warnings + failed.get(k, [])
        # The per-item tail is isolated as scan_batch always was: an error
        # here files that item under scan_error and the rest go on.
        try:
            raw = per_text[k] + self._collect_denylist_spans(text)
            spans = self._filter_spans(raw, text)
            context_id = new_context_id()
            sanitized, violations = self._anonymizer.apply(text, spans, context_id=context_id)
        except Exception as exc:
            result = self._error_result(text, exc)
            result.warnings = item_warnings
            results[i] = result
            continue
        result = ScanResult(
            original_text=text,
            sanitized_text=sanitized,
            violations=violations,
            warnings=item_warnings,
            context_id=context_id,
            _salt=self.salt,
            _locale=self._locale,
        )
        if self._strict and item_warnings:
            raise DegradedScanError(item_warnings, result)
        results[i] = result
    return results  # type: ignore[return-value]

scan_async async

scan_async(text)

Async variant — uses native async for I/O-bound detectors (LLM backend).

CPU-bound detectors (regex, NER) run via asyncio.to_thread; the LLM detector uses its own detect_async() method with a native httpx.AsyncClient.

Source code in src/wardcat/core/engine.py

async def scan_async(self, text: str) -> ScanResult:
    """Async variant — uses native async for I/O-bound detectors (LLM backend).

    CPU-bound detectors (regex, NER) run via ``asyncio.to_thread``;
    the LLM detector uses its own ``detect_async()`` method with a
    native ``httpx.AsyncClient``.
    """
    t_start = time.perf_counter()
    self._check_size(text)

    warnings: list[str] = list(self.build_warnings)
    if self._use_adjudication:
        cand_results = await asyncio.gather(
            *(self._safe_detect_async(d, text) for d in self._other_detectors)
        )
        candidate_spans: list[DetectedSpan] = [s for spans, _ in cand_results for s in spans]
        warnings.extend(w for _, w in cand_results if w)
        raw_spans = [
            s for s in candidate_spans if s.confidence >= _ADJUDICATION_KEEP_CONFIDENCE
        ]
        adj_results = await asyncio.gather(
            *(
                self._safe_detect_async(d, text, candidates=candidate_spans)
                for d in self._adjudicators
            )
        )
        for spans, w in adj_results:
            raw_spans.extend(spans)
            if w:
                warnings.append(w)
    else:
        results = await asyncio.gather(
            *(self._safe_detect_async(d, text) for d in self.detectors)
        )
        raw_spans = [s for spans, _ in results for s in spans]
        warnings.extend(w for _, w in results if w)
    raw_spans.extend(self._collect_denylist_spans(text))
    spans = self._filter_spans(raw_spans, text)
    # One context id per scan — it is what keeps this scan's reversible
    # placeholders distinct from every other scan's (see TokenAllocator).
    context_id = new_context_id()
    sanitized, violations = self._anonymizer.apply(text, spans, context_id=context_id)

    elapsed_ms = (time.perf_counter() - t_start) * 1000
    logger.info(
        "scan_async completed: %d violation(s), %d character(s), %.1f ms",
        len(violations),
        len(text),
        elapsed_ms,
    )
    result = ScanResult(
        original_text=text,
        sanitized_text=sanitized,
        violations=violations,
        warnings=warnings,
        context_id=context_id,
        _salt=self.salt,
        _locale=self._locale,
    )
    if self._strict and warnings:
        raise DegradedScanError(warnings, result)
    return result

wardcat.core.anonymizer.Anonymizer

Anonymizer(entity_config, salt='', locale='en')

Applies configured actions to detected spans and rebuilds the text.

Source code in src/wardcat/core/anonymizer.py

def __init__(self, entity_config: dict[str, Any], salt: str = "", locale: str = "en") -> None:
    self._entity_config = entity_config
    self._salt = salt
    self._locale = locale

apply

apply(text, spans, *, context_id=None)

Return (sanitized_text, violations) for spans (already filtered).

spans must be sorted/non-overlapping (the engine guarantees this).

Parameters:

Name	Type	Description	Default
context_id	str | None	stamped into reversible (tokenize) placeholders so this call's tokens cannot collide with another scan's. Defaults to a fresh id; pass one to record it alongside the result, or "" for bare [TYPE_index] placeholders.	None

Source code in src/wardcat/core/anonymizer.py

def apply(
    self,
    text: str,
    spans: list[DetectedSpan],
    *,
    context_id: str | None = None,
) -> tuple[str, list[Violation]]:
    """Return ``(sanitized_text, violations)`` for *spans* (already filtered).

    *spans* must be sorted/non-overlapping (the engine guarantees this).

    :param context_id: stamped into reversible (``tokenize``) placeholders so
        this call's tokens cannot collide with another scan's. Defaults to a
        fresh id; pass one to record it alongside the result, or ``""`` for
        bare ``[TYPE_index]`` placeholders.
    """
    # A fresh context per call: reversible actions allocate placeholders in it,
    # so state must not leak between scans — one Anonymizer instance is shared
    # by every scan on a guard, including the concurrent ones in scan_batch.
    ctx = ActionContext(
        salt=self._salt,
        tokens=TokenAllocator(context_id),
        locale=self._locale,
        surrogates=SurrogateAllocator(self._salt, self._locale),
    )
    violations: list[Violation] = []
    sanitized = text
    offset = 0

    for span in spans:
        action_name = self._entity_config.get(span.entity_type, {}).get("action", "warn")
        try:
            replacement = get_action(action_name)(span, ctx)
        except ActionFallback as fallback:
            # The action declined this span; the one it named is applied
            # and recorded, so the violation says what really happened.
            action_name = fallback.action
            replacement = get_action(action_name)(span, ctx)

        # Where this span sits in the output: shifted by every replacement
        # before it, and as wide as what now stands there.
        adj_start = span.start + offset
        if replacement is not None:
            adj_end = span.end + offset
            sanitized = sanitized[:adj_start] + replacement + sanitized[adj_end:]
            offset += len(replacement) - (span.end - span.start)
            sanitized_end = adj_start + len(replacement)
        else:
            sanitized_end = adj_start + (span.end - span.start)

        violations.append(
            Violation(
                entity_type=span.entity_type,
                original=span.text,
                start=span.start,
                end=span.end,
                action=action_name,
                replacement=replacement,
                confidence=span.confidence,
                source=span.source,
                sanitized_start=adj_start,
                sanitized_end=sanitized_end,
            )
        )

    return sanitized, violations