Module adcp.types.error_narrowing

Narrow pydantic discriminated-union ValidationErrors to the variant the user actually intended.

Background (Stability AI Emma backend test, verdict 5/10): when an adopter constructs a CreativeManifest whose assets value matches one variant of a discriminated union but is missing fields required by THAT variant, pydantic 2 reports validation errors for EVERY variant in the union (13+ for asset content types). The error dump runs 60+ lines and obscures the actual problem (a single missing field on the variant the user picked).

This module exposes :func:narrow_union_errors() which post-processes ValidationError.errors() to keep only the errors from the "closest fit" variant — the one whose discriminator matched OR the one with the fewest non-discriminator errors. The result is a focused error pointing at the user's actual mistake.

Used by:

• :func:create_tool_caller() — narrows wire-side INVALID_REQUEST errors automatically.
• Adopter code via :func:narrow_validation_error (manual) — for adopters who construct typed models in their platform method bodies and want the same friendlier error UX.

Functions

def narrow_union_errors(errors: Any) ‑> list[typing.Any]

Expand source code

def narrow_union_errors(
    errors: Any,
) -> list[Any]:
    """Return a focused subset of ``errors`` for discriminated-union
    failures.

    For each (parent_loc) where multiple variant errors exist, pick
    the "best fit" variant by:

    1. **Discriminator match**: variants with no ``literal_error``,
       ``union_tag_not_found``, or ``union_tag_invalid`` had their
       discriminator value match the user's input. Keep ONLY their
       errors.
    2. **Fewest non-discriminator errors**: if no clear discriminator
       winner, the variant with the smallest count of non-literal
       errors is the closest fit. Keep ONLY its errors.

    Errors that aren't part of a union failure (no variant in their
    ``loc``) pass through unchanged. The function never returns an
    empty list when the input is non-empty — the worst case falls
    back to the input.

    **Edge case** (residual): pydantic's ``literal_error`` type fires
    on ANY ``Literal[...]`` field mismatch, not just the discriminator.
    A user input that hits a non-discriminator literal mismatch on the
    matched variant (e.g., correct ``asset_type`` but wrong
    ``codec``) will eliminate the matched variant from step 1 and the
    fallback may pick a wrong variant. The narrowing reduces noise
    even in this case but may surface the wrong variant's errors.
    Resolving this requires knowing the discriminator field name,
    which the heuristic doesn't have access to. Schema-level fix
    (``Annotated[Union[...], Field(discriminator=...)]``) avoids the
    issue entirely; tracked as a follow-up.

    Mirrors the JS-side ``narrowUnionValidationErrors`` (when ported).
    """
    if not errors:
        return []

    errors_list = list(errors)
    # DoS guard: don't process pathologically-large inputs. Below the
    # cap, narrowing helps. Above it, we're either in a hostile
    # request or a legitimately massive schema; either way, the
    # narrowing UX win doesn't justify the CPU.
    if len(errors_list) > _MAX_NARROW_INPUT_SIZE:
        return errors_list

    # Bucket errors by (prefix_before_variant) — every error sharing
    # the same prefix is contending for the same logical slot, and
    # different errors in the same bucket are different variants of
    # the same union.
    buckets: dict[tuple[Any, ...], list[tuple[str, Any]]] = {}
    passthrough: list[Any] = []
    asset_variant_prefixes: set[tuple[Any, ...]] = set()

    for err in errors_list:
        loc = tuple(err.get("loc", ()))
        if "AssetVariant" in loc:
            asset_variant_prefixes.add(loc[: loc.index("AssetVariant")])

    for err in errors_list:
        loc = tuple(err.get("loc", ()))
        if "Assets" in loc and loc[: loc.index("Assets")] in asset_variant_prefixes:
            continue
        split = _split_at_variant(loc)
        if split is None:
            passthrough.append(err)
            continue
        prefix, variant, _suffix = split
        buckets.setdefault(prefix, []).append((variant, err))

    if not buckets:
        return errors_list

    # Defensive copy of the dicts we're about to surface — the caller
    # might mutate the returned list and we don't want that to leak
    # back into the input. ``dict(err)`` is shallow which is fine:
    # ``loc`` is a tuple (immutable), and other values are scalars or
    # nested dicts pydantic doesn't share across errors.
    narrowed: list[Any] = [dict(err) for err in passthrough]
    for _prefix, variant_errors in buckets.items():
        # Group by variant name within this bucket.
        per_variant: dict[str, list[Any]] = {}
        for variant, err in variant_errors:
            per_variant.setdefault(variant, []).append(err)

        if len(per_variant) <= 1:
            # Only one variant in this bucket — no narrowing needed.
            for errs in per_variant.values():
                narrowed.extend(dict(e) for e in errs)
            continue

        winner = _pick_winning_variant(per_variant)
        if winner is None:
            # Couldn't disambiguate; fall back to all variants for
            # this bucket so the adopter doesn't lose information.
            for errs in per_variant.values():
                narrowed.extend(dict(e) for e in errs)
            continue
        narrowed.extend(dict(e) for e in per_variant[winner])

    return narrowed

Return a focused subset of errors for discriminated-union failures.

For each (parent_loc) where multiple variant errors exist, pick the "best fit" variant by:

1. Discriminator match: variants with no literal_error, union_tag_not_found, or union_tag_invalid had their discriminator value match the user's input. Keep ONLY their errors.
2. Fewest non-discriminator errors: if no clear discriminator winner, the variant with the smallest count of non-literal errors is the closest fit. Keep ONLY its errors.

Errors that aren't part of a union failure (no variant in their loc) pass through unchanged. The function never returns an empty list when the input is non-empty — the worst case falls back to the input.

Edge case (residual): pydantic's literal_error type fires on ANY Literal[…] field mismatch, not just the discriminator. A user input that hits a non-discriminator literal mismatch on the matched variant (e.g., correct asset_type but wrong codec) will eliminate the matched variant from step 1 and the fallback may pick a wrong variant. The narrowing reduces noise even in this case but may surface the wrong variant's errors. Resolving this requires knowing the discriminator field name, which the heuristic doesn't have access to. Schema-level fix (Annotated[Union[...], Field(discriminator=...)]) avoids the issue entirely; tracked as a follow-up.

Mirrors the JS-side narrowUnionValidationErrors (when ported).