refactor(parser): resolve immutable release candidates

backend/src/module/parser/analyser/tokenizer/__init__.py

@@ -1,13 +1,17 @@
 """Public APIs for generic and legacy resource-title parsing."""
 
 from .compat import tokenize_title
-from .parser import parse_release_title
+from .parser import parse_release_title, parse_release_title_with_trace
 from .result import MediaType, ParsedRelease, ReleaseKind
+from .trace import ParseOutcome, ParseTrace
 
 __all__ = [
     "MediaType",
+    "ParseOutcome",
+    "ParseTrace",
     "ParsedRelease",
     "ReleaseKind",
     "parse_release_title",
+    "parse_release_title_with_trace",
     "tokenize_title",
 ]

backend/src/module/parser/analyser/tokenizer/candidate.py

@@ -0,0 +1,240 @@
+"""Immutable evidence and candidate models used by the title resolver.
+
+Regex rules should only describe what they observed.  They must not mutate the
+parse result or remove text from the title.  Semantic candidates group one or
+more observations into an interpretation that can be resolved deterministically.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from enum import StrEnum
+
+from .result import MediaType, ReleaseKind
+
+
+class ClaimField(StrEnum):
+    """Scalar and repeatable fields a candidate can contribute."""
+
+    GROUP = "group"
+    SEASON = "season"
+    SEASON_RAW = "season_raw"
+    EPISODE = "episode"
+    EPISODE_END = "episode_end"
+    EPISODE_TITLE = "episode_title"
+    MEDIA_TYPE = "media_type"
+    RELEASE_KIND = "release_kind"
+    RESOLUTION = "resolution"
+    SOURCE = "source"
+    SUBTITLE = "subtitle"
+    CODECS = "codecs"
+    AUDIO = "audio"
+    CONTAINER = "container"
+    VERSION = "version"
+    YEAR = "year"
+    TAGS = "tags"
+
+
+class ShadowedSpanPolicy(StrEnum):
+    """Whether valid-but-shadowed metadata remains excluded from the title."""
+
+    EXCLUDE = "exclude"
+    KEEP = "keep"
+
+
+class OverlapPolicy(StrEnum):
+    """Whether a candidate may share source characters with another winner."""
+
+    EXCLUSIVE = "exclusive"
+    SHARED = "shared"
+
+
+class DecisionStatus(StrEnum):
+    """Outcome of resolving one semantic candidate."""
+
+    SELECTED = "selected"
+    SHADOWED = "shadowed"
+    REJECTED_AS_TITLE = "rejected_as_title"
+    REJECTED_CONFLICT = "rejected_conflict"
+
+
+@dataclass(frozen=True, slots=True, order=True)
+class Span:
+    """Half-open character range using coordinates local to one segment."""
+
+    segment: int
+    start: int
+    end: int
+
+    def __post_init__(self) -> None:
+        if self.segment < 0:
+            raise ValueError("segment must be non-negative")
+        if self.start < 0:
+            raise ValueError("span start must be non-negative")
+        if self.end <= self.start:
+            raise ValueError("span end must be greater than start")
+
+    @property
+    def length(self) -> int:
+        return self.end - self.start
+
+    def overlaps(self, other: Span) -> bool:
+        return (
+            self.segment == other.segment
+            and self.start < other.end
+            and other.start < self.end
+        )
+
+    def contains(self, other: Span) -> bool:
+        return (
+            self.segment == other.segment
+            and self.start <= other.start
+            and self.end >= other.end
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class Observation:
+    """A direct, interpretation-free match produced by a tokenizer rule."""
+
+    id: str
+    rule_id: str
+    kind: str
+    span: Span
+    text: str
+    captures: tuple[str | None, ...] = ()
+
+    def __post_init__(self) -> None:
+        if not self.id:
+            raise ValueError("observation id must not be empty")
+        if not self.rule_id:
+            raise ValueError("observation rule_id must not be empty")
+        if not self.kind:
+            raise ValueError("observation kind must not be empty")
+
+    @property
+    def sort_key(self) -> tuple[int, int, int, str, str]:
+        return (
+            self.span.segment,
+            self.span.start,
+            self.span.end,
+            self.rule_id,
+            self.id,
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class Claims:
+    """Typed semantic fields proposed by a candidate.
+
+    ``None`` means that the candidate makes no claim for a scalar field.  Empty
+    tuples likewise mean no contribution to repeatable fields.
+    """
+
+    group: str | None = None
+    season: int | None = None
+    season_raw: str | None = None
+    episode: int | float | None = None
+    episode_end: int | float | None = None
+    episode_title: str | None = None
+    media_type: MediaType | None = None
+    release_kind: ReleaseKind | None = None
+    resolution: str | None = None
+    source: str | None = None
+    subtitle: str | None = None
+    codecs: tuple[str, ...] = ()
+    audio: tuple[str, ...] = ()
+    container: str | None = None
+    version: int | None = None
+    year: int | None = None
+    tags: tuple[str, ...] = ()
+
+    def scalar_items(self) -> tuple[tuple[ClaimField, object], ...]:
+        values: tuple[tuple[ClaimField, object | None], ...] = (
+            (ClaimField.GROUP, self.group),
+            (ClaimField.SEASON, self.season),
+            (ClaimField.SEASON_RAW, self.season_raw),
+            (ClaimField.EPISODE, self.episode),
+            (ClaimField.EPISODE_END, self.episode_end),
+            (ClaimField.EPISODE_TITLE, self.episode_title),
+            (ClaimField.MEDIA_TYPE, self.media_type),
+            (ClaimField.RELEASE_KIND, self.release_kind),
+            (ClaimField.RESOLUTION, self.resolution),
+            (ClaimField.SOURCE, self.source),
+            (ClaimField.SUBTITLE, self.subtitle),
+            (ClaimField.CONTAINER, self.container),
+            (ClaimField.VERSION, self.version),
+            (ClaimField.YEAR, self.year),
+        )
+        return tuple((field, value) for field, value in values if value is not None)
+
+    def repeatable_items(self) -> tuple[tuple[ClaimField, tuple[str, ...]], ...]:
+        return (
+            (ClaimField.CODECS, self.codecs),
+            (ClaimField.AUDIO, self.audio),
+            (ClaimField.TAGS, self.tags),
+        )
+
+    @property
+    def is_empty(self) -> bool:
+        return not self.scalar_items() and not any(
+            values for _, values in self.repeatable_items()
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class Candidate:
+    """One semantic interpretation offered to the resolver.
+
+    ``conflict_tags`` and ``blocks`` express cross-field incompatibilities.  A
+    strong movie marker, for example, can block the ``ambiguous_episode`` tag
+    carried by a trailing bare number without blocking an explicit SxxExx.
+    """
+
+    id: str
+    rule_id: str
+    spans: tuple[Span, ...]
+    claims: Claims = Claims()
+    priority: int = 0
+    specificity: int = 0
+    observation_ids: tuple[str, ...] = ()
+    evidence: tuple[str, ...] = ()
+    conflict_tags: frozenset[str] = frozenset()
+    blocks: frozenset[str] = frozenset()
+    preserve_as_title_on_conflict: bool = False
+    shadowed_span_policy: ShadowedSpanPolicy = ShadowedSpanPolicy.EXCLUDE
+    shadowed_spans: tuple[Span, ...] | None = None
+    overlap_policy: OverlapPolicy = OverlapPolicy.EXCLUSIVE
+
+    def __post_init__(self) -> None:
+        if not self.id:
+            raise ValueError("candidate id must not be empty")
+        if not self.rule_id:
+            raise ValueError("candidate rule_id must not be empty")
+        canonical_spans = tuple(sorted(set(self.spans)))
+        object.__setattr__(self, "spans", canonical_spans)
+        if self.shadowed_spans is not None:
+            canonical_shadowed = tuple(sorted(set(self.shadowed_spans)))
+            if not set(canonical_shadowed).issubset(canonical_spans):
+                raise ValueError("shadowed_spans must be a subset of spans")
+            object.__setattr__(self, "shadowed_spans", canonical_shadowed)
+
+    @property
+    def sort_key(self) -> tuple[int, int, tuple[Span, ...], str, str]:
+        """Stable ordering key; it never depends on collection insertion order."""
+        return (-self.priority, -self.specificity, self.spans, self.rule_id, self.id)
+
+
+@dataclass(frozen=True, slots=True)
+class Decision:
+    """Auditable outcome for one candidate."""
+
+    candidate_id: str
+    status: DecisionStatus
+    reason: str
+    won_fields: tuple[ClaimField, ...] = ()
+    excluded_spans: tuple[Span, ...] = ()
+
+    @property
+    def exclude_from_title(self) -> bool:
+        return bool(self.excluded_spans)

backend/src/module/parser/analyser/tokenizer/resolver.py

@@ -0,0 +1,339 @@
+"""Deterministic conflict resolution for tokenizer candidates."""
+
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from itertools import combinations
+from typing import Iterable
+
+from .candidate import (
+    Candidate,
+    ClaimField,
+    Claims,
+    Decision,
+    DecisionStatus,
+    OverlapPolicy,
+    ShadowedSpanPolicy,
+    Span,
+)
+from .result import MediaType, ReleaseKind
+
+
+@dataclass(frozen=True, slots=True)
+class Resolution:
+    """Resolved semantic values plus the decisions that produced them."""
+
+    claims: Claims
+    decisions: tuple[Decision, ...]
+    excluded_spans: tuple[Span, ...]
+    evidence: tuple[str, ...]
+    warnings: tuple[ResolutionWarning, ...] = ()
+
+    @property
+    def selected_candidate_ids(self) -> tuple[str, ...]:
+        return tuple(
+            decision.candidate_id
+            for decision in self.decisions
+            if decision.status is DecisionStatus.SELECTED
+        )
+
+    def decision_for(self, candidate_id: str) -> Decision:
+        for decision in self.decisions:
+            if decision.candidate_id == candidate_id:
+                return decision
+        raise KeyError(candidate_id)
+
+
+@dataclass(frozen=True, slots=True, order=True)
+class ResolutionWarning:
+    """Non-fatal ambiguity surfaced instead of hidden by a stable tie-break."""
+
+    field: ClaimField
+    candidate_ids: tuple[str, str]
+    reason: str = "equal-rank-conflicting-claims"
+
+
+def resolve_candidates(
+    candidates: Iterable[Candidate], *, collect_warnings: bool = True
+) -> Resolution:
+    """Resolve candidates without depending on their input iteration order.
+
+    Higher priority and specificity win.  Remaining ties are settled by source
+    coordinates, rule id, and candidate id.  Rules should encode intentional
+    precedence in ``priority`` rather than relying on declaration order.
+    """
+
+    ordered = tuple(sorted(candidates, key=lambda candidate: candidate.sort_key))
+    _ensure_unique_ids(ordered)
+
+    selected: list[Candidate] = []
+    scalar_values: dict[ClaimField, object] = {}
+    repeatable_values: dict[ClaimField, list[str]] = {
+        ClaimField.CODECS: [],
+        ClaimField.AUDIO: [],
+        ClaimField.TAGS: [],
+    }
+    decisions: list[Decision] = []
+
+    for candidate in ordered:
+        explicit_blocker = _find_explicit_blocker(candidate, selected)
+        if explicit_blocker is not None:
+            as_title = candidate.preserve_as_title_on_conflict
+            decisions.append(
+                Decision(
+                    candidate_id=candidate.id,
+                    status=(
+                        DecisionStatus.REJECTED_AS_TITLE
+                        if as_title
+                        else DecisionStatus.REJECTED_CONFLICT
+                    ),
+                    reason=f"blocked-by:{explicit_blocker.id}",
+                )
+            )
+            continue
+
+        overlap_blocker = _find_overlap_blocker(candidate, selected)
+        if overlap_blocker is not None:
+            as_title = candidate.preserve_as_title_on_conflict
+            decisions.append(
+                Decision(
+                    candidate_id=candidate.id,
+                    status=(
+                        DecisionStatus.REJECTED_AS_TITLE
+                        if as_title
+                        else DecisionStatus.REJECTED_CONFLICT
+                    ),
+                    reason=f"overlaps:{overlap_blocker.id}",
+                )
+            )
+            continue
+
+        scalar_items = candidate.claims.scalar_items()
+        occupied = tuple(field for field, _ in scalar_items if field in scalar_values)
+        if occupied:
+            shadowed_excluded_spans: tuple[Span, ...] = ()
+            if candidate.shadowed_span_policy is ShadowedSpanPolicy.EXCLUDE:
+                shadowed_excluded_spans = (
+                    candidate.shadowed_spans
+                    if candidate.shadowed_spans is not None
+                    else candidate.spans
+                )
+            decisions.append(
+                Decision(
+                    candidate_id=candidate.id,
+                    status=DecisionStatus.SHADOWED,
+                    reason="shadowed-fields:"
+                    + ",".join(field.value for field in occupied),
+                    excluded_spans=shadowed_excluded_spans,
+                )
+            )
+            continue
+
+        won_fields: list[ClaimField] = []
+        for field, value in scalar_items:
+            if field not in scalar_values:
+                scalar_values[field] = value
+                won_fields.append(field)
+
+        for field, values in candidate.claims.repeatable_items():
+            target = repeatable_values[field]
+            before = len(target)
+            for value in values:
+                _append_unique_casefold(target, value)
+            if len(target) > before:
+                won_fields.append(field)
+
+        selected.append(candidate)
+        decisions.append(
+            Decision(
+                candidate_id=candidate.id,
+                status=DecisionStatus.SELECTED,
+                reason="selected",
+                won_fields=tuple(won_fields),
+                excluded_spans=candidate.spans,
+            )
+        )
+
+    decisions_tuple = tuple(decisions)
+    warnings = (
+        _find_equal_rank_conflicts(ordered, decisions_tuple) if collect_warnings else ()
+    )
+    excluded_spans = _merge_spans(
+        span for decision in decisions_tuple for span in decision.excluded_spans
+    )
+    evidence = _collect_evidence(ordered, decisions_tuple)
+
+    return Resolution(
+        claims=_build_claims(scalar_values, repeatable_values),
+        decisions=decisions_tuple,
+        excluded_spans=excluded_spans,
+        evidence=evidence,
+        warnings=warnings,
+    )
+
+
+def _ensure_unique_ids(candidates: tuple[Candidate, ...]) -> None:
+    seen: set[str] = set()
+    for candidate in candidates:
+        if candidate.id in seen:
+            raise ValueError(f"duplicate candidate id: {candidate.id}")
+        seen.add(candidate.id)
+
+
+def _find_equal_rank_conflicts(
+    candidates: tuple[Candidate, ...],
+    decisions: tuple[Decision, ...],
+) -> tuple[ResolutionWarning, ...]:
+    selected_ids = {
+        decision.candidate_id
+        for decision in decisions
+        if decision.status is DecisionStatus.SELECTED
+    }
+    warnings: set[ResolutionWarning] = set()
+    for left, right in combinations(candidates, 2):
+        if (
+            left.priority != right.priority
+            or left.specificity != right.specificity
+            or not ({left.id, right.id} & selected_ids)
+        ):
+            continue
+        left_values = dict(left.claims.scalar_items())
+        right_values = dict(right.claims.scalar_items())
+        for field in left_values.keys() & right_values.keys():
+            if _warning_values_equal(field, left_values[field], right_values[field]):
+                continue
+            first_id, second_id = sorted((left.id, right.id))
+            warnings.add(
+                ResolutionWarning(
+                    field=field,
+                    candidate_ids=(first_id, second_id),
+                )
+            )
+    return tuple(sorted(warnings))
+
+
+def _warning_values_equal(field: ClaimField, left: object, right: object) -> bool:
+    # ``season_raw`` records the spelling that supplied the semantic season;
+    # different spellings are provenance, not contradictory values.
+    if field is ClaimField.SEASON_RAW:
+        return True
+    if isinstance(left, str) and isinstance(right, str):
+        left_normalized = left.casefold()
+        right_normalized = right.casefold()
+        if field is ClaimField.SOURCE:
+            left_normalized = re.sub(r"[-_.\s]+", "", left_normalized)
+            right_normalized = re.sub(r"[-_.\s]+", "", right_normalized)
+        return left_normalized == right_normalized
+    return left == right
+
+
+def _find_explicit_blocker(
+    candidate: Candidate, selected: list[Candidate]
+) -> Candidate | None:
+    for winner in selected:
+        if (winner.blocks & candidate.conflict_tags) or (
+            candidate.blocks & winner.conflict_tags
+        ):
+            return winner
+    return None
+
+
+def _find_overlap_blocker(
+    candidate: Candidate, selected: list[Candidate]
+) -> Candidate | None:
+    if candidate.overlap_policy is OverlapPolicy.SHARED:
+        return None
+    for winner in selected:
+        if winner.overlap_policy is OverlapPolicy.SHARED:
+            continue
+        if any(
+            span.overlaps(winner_span)
+            for span in candidate.spans
+            for winner_span in winner.spans
+        ):
+            return winner
+    return None
+
+
+def _append_unique_casefold(values: list[str], value: str) -> None:
+    folded = value.casefold()
+    if folded not in {existing.casefold() for existing in values}:
+        values.append(value)
+
+
+def _build_claims(
+    scalar_values: dict[ClaimField, object],
+    repeatable_values: dict[ClaimField, list[str]],
+) -> Claims:
+    return Claims(
+        group=_scalar(scalar_values, ClaimField.GROUP, str),
+        season=_scalar(scalar_values, ClaimField.SEASON, int),
+        season_raw=_scalar(scalar_values, ClaimField.SEASON_RAW, str),
+        episode=_number_scalar(scalar_values, ClaimField.EPISODE),
+        episode_end=_number_scalar(scalar_values, ClaimField.EPISODE_END),
+        episode_title=_scalar(scalar_values, ClaimField.EPISODE_TITLE, str),
+        media_type=_scalar(scalar_values, ClaimField.MEDIA_TYPE, MediaType),
+        release_kind=_scalar(scalar_values, ClaimField.RELEASE_KIND, ReleaseKind),
+        resolution=_scalar(scalar_values, ClaimField.RESOLUTION, str),
+        source=_scalar(scalar_values, ClaimField.SOURCE, str),
+        subtitle=_scalar(scalar_values, ClaimField.SUBTITLE, str),
+        codecs=tuple(repeatable_values[ClaimField.CODECS]),
+        audio=tuple(repeatable_values[ClaimField.AUDIO]),
+        container=_scalar(scalar_values, ClaimField.CONTAINER, str),
+        version=_scalar(scalar_values, ClaimField.VERSION, int),
+        year=_scalar(scalar_values, ClaimField.YEAR, int),
+        tags=tuple(repeatable_values[ClaimField.TAGS]),
+    )
+
+
+def _scalar[T](
+    values: dict[ClaimField, object], field: ClaimField, expected: type[T]
+) -> T | None:
+    value = values.get(field)
+    if value is None:
+        return None
+    if expected is not object and not isinstance(value, expected):
+        raise TypeError(f"invalid value for {field.value}: {value!r}")
+    return value  # type: ignore[return-value]
+
+
+def _number_scalar(
+    values: dict[ClaimField, object], field: ClaimField
+) -> int | float | None:
+    value = values.get(field)
+    if value is None:
+        return None
+    if not isinstance(value, (int, float)):
+        raise TypeError(f"invalid value for {field.value}: {value!r}")
+    return value
+
+
+def _merge_spans(spans: Iterable[Span]) -> tuple[Span, ...]:
+    ordered = sorted(set(spans))
+    if not ordered:
+        return ()
+
+    merged: list[Span] = [ordered[0]]
+    for span in ordered[1:]:
+        previous = merged[-1]
+        if span.segment == previous.segment and span.start <= previous.end:
+            merged[-1] = Span(
+                previous.segment, previous.start, max(previous.end, span.end)
+            )
+        else:
+            merged.append(span)
+    return tuple(merged)
+
+
+def _collect_evidence(
+    candidates: tuple[Candidate, ...], decisions: tuple[Decision, ...]
+) -> tuple[str, ...]:
+    evidence: list[str] = []
+    for candidate, decision in zip(candidates, decisions, strict=True):
+        if decision.status is not DecisionStatus.SELECTED:
+            continue
+        for item in candidate.evidence:
+            if item not in evidence:
+                evidence.append(item)
+    return tuple(evidence)

backend/src/module/parser/analyser/tokenizer/trace.py

@@ -0,0 +1,108 @@
+"""Immutable diagnostics for explaining release-title parser decisions."""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from .candidate import Candidate, Claims, Decision, DecisionStatus, Observation, Span
+from .resolver import Resolution, ResolutionWarning
+from .result import ParsedRelease
+
+
+@dataclass(frozen=True, slots=True)
+class TraceSegment:
+    """Segment snapshot with explicit normalized-text coordinates."""
+
+    index: int
+    text: str
+    outer_start: int
+    outer_end: int
+    content_start: int
+    content_end: int
+    enclosure: str | None = None
+
+    def __post_init__(self) -> None:
+        if self.index < 0:
+            raise ValueError("segment index must be non-negative")
+        if not (0 <= self.outer_start <= self.content_start):
+            raise ValueError("invalid segment start coordinates")
+        if not (self.content_start <= self.content_end <= self.outer_end):
+            raise ValueError("invalid segment end coordinates")
+
+
+@dataclass(frozen=True, slots=True)
+class ParseTrace:
+    """All observations and resolver decisions for one parser invocation."""
+
+    raw: str
+    normalized: str
+    segments: tuple[TraceSegment, ...] = ()
+    observations: tuple[Observation, ...] = ()
+    candidates: tuple[Candidate, ...] = ()
+    claims: Claims = Claims()
+    decisions: tuple[Decision, ...] = ()
+    excluded_spans: tuple[Span, ...] = ()
+    residuals: tuple[str, ...] = ()
+    warnings: tuple[ResolutionWarning, ...] = ()
+
+    def __post_init__(self) -> None:
+        object.__setattr__(
+            self, "segments", tuple(sorted(self.segments, key=lambda item: item.index))
+        )
+        object.__setattr__(
+            self,
+            "observations",
+            tuple(sorted(self.observations, key=lambda item: item.sort_key)),
+        )
+        object.__setattr__(
+            self,
+            "candidates",
+            tuple(sorted(self.candidates, key=lambda item: item.sort_key)),
+        )
+
+    @classmethod
+    def from_resolution(
+        cls,
+        *,
+        raw: str,
+        normalized: str,
+        resolution: Resolution,
+        segments: tuple[TraceSegment, ...] = (),
+        observations: tuple[Observation, ...] = (),
+        candidates: tuple[Candidate, ...] = (),
+        residuals: tuple[str, ...] = (),
+    ) -> ParseTrace:
+        return cls(
+            raw=raw,
+            normalized=normalized,
+            segments=segments,
+            observations=observations,
+            candidates=candidates,
+            claims=resolution.claims,
+            decisions=resolution.decisions,
+            excluded_spans=resolution.excluded_spans,
+            residuals=residuals,
+            warnings=resolution.warnings,
+        )
+
+    def decision_for(self, candidate_id: str) -> Decision:
+        for decision in self.decisions:
+            if decision.candidate_id == candidate_id:
+                return decision
+        raise KeyError(candidate_id)
+
+    @property
+    def rejected_as_title(self) -> tuple[str, ...]:
+        return tuple(
+            decision.candidate_id
+            for decision in self.decisions
+            if decision.status is DecisionStatus.REJECTED_AS_TITLE
+        )
+
+
+@dataclass(frozen=True, slots=True)
+class ParseOutcome:
+    """Optional parse result paired with its complete diagnostic trace."""
+
+    result: ParsedRelease | None
+    trace: ParseTrace

backend/src/test/test_release_parser.py

@@ -113,6 +113,52 @@ def test_episode_range_is_a_range_release(raw: str) -> None:
     assert parsed.release_kind is ReleaseKind.RANGE
 
 
+@pytest.mark.parametrize(
+    ("raw", "media_type", "season", "episode_end"),
+    (
+        (
+            "[Group] Anime Title S02E01-E12 [1080p]",
+            MediaType.EPISODE,
+            2,
+            12,
+        ),
+        (
+            "[Group] Anime Title Season 2 Episodes 01-12 [1080p]",
+            MediaType.EPISODE,
+            2,
+            12,
+        ),
+        (
+            "[Group] Anime Title OVA01-02 [1080p]",
+            MediaType.OVA,
+            None,
+            2,
+        ),
+        (
+            "[Group] Anime Title OAD01-OAD03 [1080p]",
+            MediaType.OAD,
+            None,
+            3,
+        ),
+    ),
+)
+def test_compound_range_forms_do_not_degrade_to_single_episode(
+    raw: str,
+    media_type: MediaType,
+    season: int | None,
+    episode_end: int,
+) -> None:
+    parsed = parse_release_title(raw)
+
+    assert parsed is not None
+    assert parsed.media_type is media_type
+    assert parsed.release_kind is ReleaseKind.RANGE
+    assert parsed.season == season
+    assert parsed.episode == 1
+    assert parsed.episode_end == episode_end
+    assert raw_parser(raw) is None
+
+
 def test_complete_batch_without_episode_number() -> None:
     parsed = parse_release_title(
         "[VCB-Studio] Violet Evergarden Complete Batch [BDRip 1080p]"
@@ -262,6 +308,36 @@ def test_movie_word_in_group_does_not_mark_episode_as_movie() -> None:
     assert parsed.media_type is MediaType.EPISODE
 
 
+def test_rejected_structural_marker_inside_group_does_not_infer_media() -> None:
+    parsed = parse_release_title("[S02E05-Fansub] Plain Anime Title [1080p]")
+
+    assert parsed is not None
+    assert parsed.group == "S02E05-Fansub"
+    assert parsed.episode is None
+    assert parsed.media_type is MediaType.UNKNOWN
+
+
+def test_shadowed_episode_title_candidate_does_not_remove_title_text() -> None:
+    raw = "[Group] Anime S01E01 Season 2 Episode 3 - Another Title " "[1080p]"
+
+    parsed = parse_release_title(raw)
+
+    assert parsed is not None
+    assert parsed.season == 1
+    assert parsed.episode == 1
+    assert "Another Title" in _title_text(parsed)
+    assert "episode-title" not in parsed.evidence
+
+
+def test_resolved_evidence_excludes_shadowed_special_candidate() -> None:
+    parsed = parse_release_title("[Group] Anime OVA01 OAD02 [1080p]")
+
+    assert parsed is not None
+    assert parsed.media_type is MediaType.OVA
+    assert "ova" in parsed.evidence
+    assert "oad" not in parsed.evidence
+
+
 def test_movie_word_in_regular_series_title_does_not_override_episode() -> None:
     parsed = parse_release_title(
         "[SubGroup] The Movie Star Next Door - 03 [WEB-DL 1080p]"

backend/src/test/test_release_parser_golden.py

@@ -0,0 +1,160 @@
+"""Golden-corpus regression tests for release-title parsing.
+
+The fixture snapshots only the public generic result and the legacy adapter.
+Parser evidence and trace data are deliberately excluded so internal rule and
+resolver changes do not create meaningless snapshot churn.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import asdict, fields
+from enum import Enum
+from pathlib import Path
+from typing import Any
+
+import pytest
+
+from module.parser.analyser.raw_parser import raw_parser
+from module.parser.analyser.tokenizer import (
+    parse_release_title,
+    parse_release_title_with_trace,
+)
+
+_CORPUS_PATH = Path(__file__).parent / "fixtures" / "release_title_golden.json"
+_CORPUS_SIZE = 71
+
+_GENERIC_DEFAULTS: dict[str, Any] = {
+    "title_en": None,
+    "title_zh": None,
+    "title_jp": None,
+    "group": None,
+    "season": None,
+    "season_raw": None,
+    "episode": None,
+    "episode_end": None,
+    "episode_title": None,
+    "media_type": "unknown",
+    "release_kind": "single",
+    "resolution": None,
+    "source": None,
+    "subtitle": None,
+    "codecs": [],
+    "audio": [],
+    "container": None,
+    "version": None,
+    "year": None,
+    "tags": [],
+}
+
+
+def _load_corpus() -> dict[str, Any]:
+    with _CORPUS_PATH.open(encoding="utf-8") as corpus_file:
+        return json.load(corpus_file)
+
+
+_CORPUS = _load_corpus()
+_CASES = _CORPUS["cases"]
+
+
+def _case_params() -> list[Any]:
+    return [pytest.param(case, id=case["id"]) for case in _CASES]
+
+
+def _json_value(value: Any) -> Any:
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, tuple):
+        return list(value)
+    return value
+
+
+def _generic_snapshot(parsed: Any) -> dict[str, Any]:
+    return {
+        field_name: _json_value(getattr(parsed, field_name))
+        for field_name in _GENERIC_DEFAULTS
+    }
+
+
+def _expected_generic(case: dict[str, Any]) -> dict[str, Any] | None:
+    overrides = case["expected"]
+    if overrides is None:
+        return None
+    return {**_GENERIC_DEFAULTS, **overrides}
+
+
+def test_golden_corpus_integrity() -> None:
+    assert _CORPUS["schema_version"] == 1
+    assert len(_CASES) == _CORPUS_SIZE
+
+    ids = [case["id"] for case in _CASES]
+    raw_titles = [case["raw"] for case in _CASES]
+    identities = [(case["category"], case["id"], case["raw"]) for case in _CASES]
+
+    assert len(ids) == len(set(ids))
+    assert len(raw_titles) == len(set(raw_titles))
+    assert len(identities) == len(set(identities))
+    assert all(case["category"] for case in _CASES)
+    assert all("evidence" not in case for case in _CASES)
+    assert all(
+        case["expected"] is None or "evidence" not in case["expected"]
+        for case in _CASES
+    )
+    assert all("trace" not in case for case in _CASES)
+
+
+@pytest.mark.parametrize("case", _case_params())
+def test_generic_parser_matches_golden_corpus(case: dict[str, Any]) -> None:
+    parsed = parse_release_title(case["raw"])
+    expected = _expected_generic(case)
+
+    if expected is None:
+        assert parsed is None
+        return
+
+    assert parsed is not None
+    assert parsed.raw == case["raw"]
+    public_fields = {
+        field.name for field in fields(parsed) if field.name not in {"raw", "evidence"}
+    }
+    assert public_fields == set(_GENERIC_DEFAULTS)
+    assert _generic_snapshot(parsed) == expected
+
+
+@pytest.mark.parametrize("case", _case_params())
+def test_legacy_parser_matches_golden_corpus(case: dict[str, Any]) -> None:
+    parsed = raw_parser(case["raw"])
+    expected = case["legacy"]
+
+    if expected is None:
+        assert parsed is None
+        return
+
+    assert parsed is not None
+    assert asdict(parsed) == expected
+
+
+@pytest.mark.parametrize("case", _case_params())
+def test_traced_parser_matches_plain_result_and_has_valid_references(
+    case: dict[str, Any],
+) -> None:
+    outcome = parse_release_title_with_trace(case["raw"])
+
+    assert outcome.result == parse_release_title(case["raw"])
+    candidate_ids = {candidate.id for candidate in outcome.trace.candidates}
+    observation_ids = {observation.id for observation in outcome.trace.observations}
+    assert {decision.candidate_id for decision in outcome.trace.decisions} == (
+        candidate_ids
+    )
+    assert {
+        observation_id
+        for candidate in outcome.trace.candidates
+        for observation_id in candidate.observation_ids
+    } == observation_ids
+    for span in (
+        span for candidate in outcome.trace.candidates for span in candidate.spans
+    ):
+        assert 0 <= span.segment < len(outcome.trace.segments)
+        assert (
+            0 <= span.start < span.end <= len(outcome.trace.segments[span.segment].text)
+        )

backend/src/test/test_release_parser_properties.py

@@ -0,0 +1,331 @@
+"""Metamorphic and invariant checks for the public release parser.
+
+The suite intentionally avoids depending on tokenizer implementation details.
+It uses a fixed pseudo-random seed and keeps the combined generated corpus below
+500 cases so failures stay reproducible and the test remains cheap to run.
+"""
+
+from __future__ import annotations
+
+import math
+import random
+import re
+import string
+from collections.abc import Iterable
+
+from module.parser.analyser.tokenizer import (
+    MediaType,
+    ParsedRelease,
+    ReleaseKind,
+    parse_release_title,
+)
+
+SEED = 0xAB330
+STRUCTURED_EPISODE_CASES = 4 * 3 * 3 * 3
+STRUCTURED_MEDIA_CASES = 15 * 4
+METAMORPHIC_CASES = 60
+FUZZ_CASES = 220
+GENERATED_CASES = (
+    STRUCTURED_EPISODE_CASES + STRUCTURED_MEDIA_CASES + METAMORPHIC_CASES + FUZZ_CASES
+)
+
+
+def _normalized_titles(parsed: ParsedRelease) -> tuple[str, ...]:
+    """Compare title content without prescribing language-slot heuristics."""
+    return tuple(
+        sorted(
+            re.sub(r"\s+", " ", title).strip().casefold()
+            for title in (parsed.title_en, parsed.title_zh, parsed.title_jp)
+            if title
+        )
+    )
+
+
+def _core_identity(parsed: ParsedRelease) -> tuple[object, ...]:
+    """Fields that technical adornments must not change."""
+    return (
+        _normalized_titles(parsed),
+        parsed.group,
+        parsed.season,
+        parsed.episode,
+        parsed.episode_end,
+        parsed.episode_title,
+        parsed.media_type,
+        parsed.release_kind,
+        parsed.version,
+        parsed.year,
+    )
+
+
+def _title_contains(parsed: ParsedRelease, expected: str) -> bool:
+    expected = re.sub(r"\s+", " ", expected).strip().casefold()
+    return any(expected in title for title in _normalized_titles(parsed))
+
+
+def _assert_public_invariants(raw: str, parsed: ParsedRelease | None) -> None:
+    """Check only guarantees implied by the public result contract."""
+    assert parsed == parse_release_title(raw), f"non-deterministic parse for {raw!r}"
+    if parsed is None:
+        return
+
+    assert parsed.raw == raw
+    assert isinstance(parsed.media_type, MediaType)
+    assert isinstance(parsed.release_kind, ReleaseKind)
+    assert parsed.primary_title is not None
+    assert parsed.primary_title.strip()
+    assert _normalized_titles(parsed)
+
+    for title in (parsed.title_en, parsed.title_zh, parsed.title_jp):
+        if title is not None:
+            assert title == title.strip()
+
+    for number in (parsed.episode, parsed.episode_end):
+        if number is not None:
+            assert isinstance(number, (int, float)) and not isinstance(number, bool)
+            assert math.isfinite(float(number))
+            assert number >= 0
+
+    if parsed.episode_end is not None:
+        assert parsed.episode is not None
+    if parsed.season is not None:
+        assert isinstance(parsed.season, int) and not isinstance(parsed.season, bool)
+        assert parsed.season >= 0
+    if parsed.version is not None:
+        assert isinstance(parsed.version, int) and parsed.version >= 0
+
+    assert isinstance(parsed.codecs, tuple)
+    assert isinstance(parsed.audio, tuple)
+    assert isinstance(parsed.tags, tuple)
+    assert isinstance(parsed.evidence, tuple)
+
+
+def test_generated_corpus_stays_within_budget() -> None:
+    assert GENERATED_CASES == 448
+    assert GENERATED_CASES <= 500
+
+
+def test_technical_adornments_preserve_numbered_episode_identity() -> None:
+    titles = ("Sousou no Frieren", "葬送的芙莉莲", "Fate stay night", "86")
+    groups = ("LoliHouse", "字幕组", "Movie-Fan")
+    episodes = ((1, "01"), (7, "07"), (12.5, "12.5"))
+    adornments = (
+        " [1080p WEB-DL HEVC AAC MKV]",
+        " [720p BDRip AVC FLAC].mkv",
+        " 【2160p BluRay x265 Dual Audio】",
+    )
+
+    seen = 0
+    for title in titles:
+        for group in groups:
+            for expected_episode, episode_text in episodes:
+                baseline_raw = f"[{group}] {title} - {episode_text}"
+                baseline = parse_release_title(baseline_raw)
+                assert baseline is not None, baseline_raw
+                _assert_public_invariants(baseline_raw, baseline)
+
+                for adornment in adornments:
+                    raw = baseline_raw + adornment
+                    parsed = parse_release_title(raw)
+                    assert parsed is not None, raw
+                    _assert_public_invariants(raw, parsed)
+                    assert _core_identity(parsed) == _core_identity(baseline), raw
+                    assert parsed.episode == expected_episode, raw
+                    assert parsed.media_type is MediaType.EPISODE, raw
+                    assert parsed.release_kind is ReleaseKind.SINGLE, raw
+                    seen += 1
+
+    assert seen == STRUCTURED_EPISODE_CASES
+
+
+def test_structured_media_forms_survive_safe_wrapping_variants() -> None:
+    forms: tuple[
+        tuple[str, str, MediaType, ReleaseKind, int | float | None, int | float | None],
+        ...,
+    ] = (
+        (
+            "Meta Anime - 01",
+            "Meta Anime",
+            MediaType.EPISODE,
+            ReleaseKind.SINGLE,
+            1,
+            None,
+        ),
+        (
+            "Meta Anime - 12.5",
+            "Meta Anime",
+            MediaType.EPISODE,
+            ReleaseKind.SINGLE,
+            12.5,
+            None,
+        ),
+        (
+            "Meta Anime - 12v2",
+            "Meta Anime",
+            MediaType.EPISODE,
+            ReleaseKind.SINGLE,
+            12,
+            None,
+        ),
+        (
+            "Meta Anime S02E05",
+            "Meta Anime",
+            MediaType.EPISODE,
+            ReleaseKind.SINGLE,
+            5,
+            None,
+        ),
+        (
+            "Meta Anime [01-12]",
+            "Meta Anime",
+            MediaType.EPISODE,
+            ReleaseKind.RANGE,
+            1,
+            12,
+        ),
+        (
+            "Meta Anime Complete Batch",
+            "Meta Anime",
+            MediaType.EPISODE,
+            ReleaseKind.BATCH,
+            None,
+            None,
+        ),
+        ("迷宫饭 全24话", "迷宫饭", MediaType.EPISODE, ReleaseKind.COLLECTION, 1, 24),
+        ("Meta Anime OVA01", "Meta Anime", MediaType.OVA, ReleaseKind.SINGLE, 1, None),
+        ("Meta Anime OAD02", "Meta Anime", MediaType.OAD, ReleaseKind.SINGLE, 2, None),
+        (
+            "Meta Anime SP03",
+            "Meta Anime",
+            MediaType.SPECIAL,
+            ReleaseKind.SINGLE,
+            3,
+            None,
+        ),
+        (
+            "劇場版 Meta Anime",
+            "Meta Anime",
+            MediaType.MOVIE,
+            ReleaseKind.SINGLE,
+            None,
+            None,
+        ),
+        ("Meta Anime PV2", "Meta Anime", MediaType.PV, ReleaseKind.SINGLE, 2, None),
+        (
+            "Meta Anime NCOP",
+            "Meta Anime",
+            MediaType.OPENING,
+            ReleaseKind.SINGLE,
+            None,
+            None,
+        ),
+        (
+            "Meta Anime NCED",
+            "Meta Anime",
+            MediaType.ENDING,
+            ReleaseKind.SINGLE,
+            None,
+            None,
+        ),
+        ("Meta Anime", "Meta Anime", MediaType.UNKNOWN, ReleaseKind.SINGLE, None, None),
+    )
+
+    wrappers = (
+        lambda value: value,
+        lambda value: f"[MetaGroup] {value} [1080p WEB-DL]",
+        lambda value: f"【MetaGroup】 {value} 【720p BDRip AVC AAC】.mkv",
+        lambda value: f"\n  [MetaGroup] {value} [HEVC][1080P]  \n",
+    )
+
+    seen = 0
+    for core, title, media_type, release_kind, episode, episode_end in forms:
+        for wrap in wrappers:
+            raw = wrap(core)
+            parsed = parse_release_title(raw)
+            assert parsed is not None, raw
+            _assert_public_invariants(raw, parsed)
+            assert _title_contains(parsed, title), raw
+            assert parsed.media_type is media_type, raw
+            assert parsed.release_kind is release_kind, raw
+            assert parsed.episode == episode, raw
+            assert parsed.episode_end == episode_end, raw
+            seen += 1
+
+    assert seen == STRUCTURED_MEDIA_CASES
+
+
+def test_whitespace_bracket_and_metadata_variants_preserve_core_identity() -> None:
+    rng = random.Random(SEED)
+    titles = ("Mob Psycho 100", "Dungeon Meshi", "葬送的芙莉莲", "Fate stay night")
+    groups = ("Group-A", "Lilith-Raws", "喵萌奶茶屋")
+
+    for _ in range(METAMORPHIC_CASES):
+        title = rng.choice(titles)
+        group = rng.choice(groups)
+        season = rng.randint(1, 12)
+        episode = rng.randint(1, 99)
+        version = rng.randint(1, 3)
+        marker = f"S{season:02d}E{episode:02d}v{version}"
+        baseline_raw = f"[{group}] {title} {marker}"
+        variant_raw = (
+            f"\n  【{group}】 {title} {marker.lower()} "
+            "【WEB-DL 1080p HEVC AAC MKV】.mkv  \n"
+        )
+
+        baseline = parse_release_title(baseline_raw)
+        variant = parse_release_title(variant_raw)
+        assert baseline is not None, baseline_raw
+        assert variant is not None, variant_raw
+        _assert_public_invariants(baseline_raw, baseline)
+        _assert_public_invariants(variant_raw, variant)
+        assert _core_identity(variant) == _core_identity(baseline), variant_raw
+
+
+def _random_inputs(rng: random.Random, count: int) -> Iterable[str]:
+    atoms = (
+        "Anime",
+        "Title",
+        "葬送的芙莉莲",
+        "アニメ",
+        "OVA",
+        "OAD02",
+        "SP",
+        "PV2",
+        "NCOP",
+        "NCED",
+        "S02E05",
+        "Season 3",
+        "01-12",
+        "12.5",
+        "Complete Batch",
+        "劇場版",
+        "WEB-DL",
+        "BDRip",
+        "1080p",
+        "HEVC",
+        "AAC",
+        "MKV",
+        "[",
+        "]",
+        "(",
+        ")",
+        "【",
+        "】",
+    )
+    separators = ("", " ", "  ", " - ", "_", "/", ".", "\n")
+    alphabet = string.ascii_letters + string.digits + " []()_-./~中日語話季\n"
+
+    for index in range(count):
+        if index % 2:
+            parts = [rng.choice(atoms) for _ in range(rng.randint(0, 12))]
+            yield rng.choice(separators).join(parts)
+        else:
+            yield "".join(rng.choice(alphabet) for _ in range(rng.randint(0, 120)))
+
+
+def test_seeded_fuzz_inputs_are_total_and_deterministic() -> None:
+    rng = random.Random(SEED)
+    corpus = list(_random_inputs(rng, FUZZ_CASES))
+    assert len(corpus) == FUZZ_CASES
+
+    for raw in corpus:
+        _assert_public_invariants(raw, parse_release_title(raw))

backend/src/test/test_tokenizer_resolver.py

@@ -0,0 +1,331 @@
+from __future__ import annotations
+
+from itertools import permutations
+
+import pytest
+
+from module.parser.analyser.tokenizer.candidate import (
+    Candidate,
+    ClaimField,
+    Claims,
+    DecisionStatus,
+    OverlapPolicy,
+    ShadowedSpanPolicy,
+    Span,
+)
+from module.parser.analyser.tokenizer.resolver import resolve_candidates
+from module.parser.analyser.tokenizer.result import MediaType
+
+
+def _candidate(
+    candidate_id: str,
+    *,
+    start: int,
+    end: int,
+    claims: Claims,
+    priority: int,
+    **kwargs: object,
+) -> Candidate:
+    return Candidate(
+        id=candidate_id,
+        rule_id=candidate_id,
+        spans=(Span(0, start, end),),
+        claims=claims,
+        priority=priority,
+        **kwargs,  # type: ignore[arg-type]
+    )
+
+
+def test_span_is_half_open_and_validated() -> None:
+    assert Span(0, 1, 4).length == 3
+    assert Span(0, 1, 4).overlaps(Span(0, 3, 5))
+    assert not Span(0, 1, 4).overlaps(Span(0, 4, 5))
+    assert not Span(0, 1, 4).overlaps(Span(1, 1, 4))
+    assert Span(0, 1, 4).contains(Span(0, 2, 3))
+
+    with pytest.raises(ValueError):
+        Span(0, 3, 3)
+
+
+def test_resolution_is_independent_of_input_order() -> None:
+    explicit = _candidate(
+        "explicit",
+        start=10,
+        end=16,
+        claims=Claims(season=2, episode=5),
+        priority=100,
+        evidence=("season", "episode"),
+    )
+    trailing = _candidate(
+        "trailing",
+        start=20,
+        end=22,
+        claims=Claims(episode=99),
+        priority=70,
+        evidence=("episode",),
+    )
+
+    resolutions = [
+        resolve_candidates(order) for order in permutations((trailing, explicit))
+    ]
+
+    assert resolutions[0] == resolutions[1]
+    result = resolutions[0]
+    assert result.claims.season == 2
+    assert result.claims.episode == 5
+    assert result.decision_for("explicit").status is DecisionStatus.SELECTED
+    shadowed = result.decision_for("trailing")
+    assert shadowed.status is DecisionStatus.SHADOWED
+    assert shadowed.exclude_from_title
+    assert result.excluded_spans == (Span(0, 10, 16), Span(0, 20, 22))
+
+
+def test_movie_rejects_ambiguous_number_as_title_without_masking_it() -> None:
+    movie = _candidate(
+        "movie",
+        start=0,
+        end=4,
+        claims=Claims(media_type=MediaType.MOVIE),
+        priority=100,
+        blocks=frozenset({"ambiguous-episode"}),
+        evidence=("movie",),
+    )
+    title_number = _candidate(
+        "trailing-number",
+        start=12,
+        end=13,
+        claims=Claims(episode=0),
+        priority=70,
+        conflict_tags=frozenset({"ambiguous-episode"}),
+        preserve_as_title_on_conflict=True,
+    )
+
+    result = resolve_candidates((title_number, movie))
+
+    assert result.claims.media_type is MediaType.MOVIE
+    assert result.claims.episode is None
+    decision = result.decision_for("trailing-number")
+    assert decision.status is DecisionStatus.REJECTED_AS_TITLE
+    assert not decision.exclude_from_title
+    assert result.excluded_spans == (Span(0, 0, 4),)
+
+
+def test_overlapping_shadowed_field_reports_conflict_without_extra_mask() -> None:
+    combined = _candidate(
+        "combined",
+        start=2,
+        end=10,
+        claims=Claims(subtitle="GB", container="MKV"),
+        priority=80,
+        specificity=2,
+    )
+    container = _candidate(
+        "container",
+        start=5,
+        end=10,
+        claims=Claims(container="MKV"),
+        priority=70,
+    )
+
+    result = resolve_candidates((container, combined))
+
+    assert result.claims.subtitle == "GB"
+    assert result.claims.container == "MKV"
+    rejected = result.decision_for("container")
+    assert rejected.status is DecisionStatus.REJECTED_CONFLICT
+    assert result.excluded_spans == (Span(0, 2, 10),)
+
+
+def test_partially_overlapping_same_field_does_not_mask_loser_only_text() -> None:
+    winner = _candidate(
+        "winner", start=2, end=10, claims=Claims(episode=1), priority=80
+    )
+    partial_loser = _candidate(
+        "partial-loser", start=8, end=16, claims=Claims(episode=2), priority=70
+    )
+
+    result = resolve_candidates((partial_loser, winner))
+
+    assert result.claims.episode == 1
+    assert (
+        result.decision_for("partial-loser").status is DecisionStatus.REJECTED_CONFLICT
+    )
+    assert result.excluded_spans == (Span(0, 2, 10),)
+
+
+def test_different_fields_on_overlapping_exclusive_spans_are_rejected() -> None:
+    source = _candidate(
+        "source",
+        start=0,
+        end=6,
+        claims=Claims(source="WEB-DL"),
+        priority=80,
+    )
+    codec = _candidate(
+        "codec",
+        start=4,
+        end=8,
+        claims=Claims(codecs=("HEVC",)),
+        priority=70,
+    )
+
+    result = resolve_candidates((codec, source))
+
+    assert result.claims.source == "WEB-DL"
+    assert result.claims.codecs == ()
+    assert result.decision_for("codec").status is DecisionStatus.REJECTED_CONFLICT
+
+
+def test_shared_overlay_allows_explicit_transparent_overlap() -> None:
+    overlay = _candidate(
+        "overlay",
+        start=0,
+        end=12,
+        claims=Claims(tags=("dub",)),
+        priority=80,
+        overlap_policy=OverlapPolicy.SHARED,
+    )
+    resolution = _candidate(
+        "resolution",
+        start=2,
+        end=7,
+        claims=Claims(resolution="1080p"),
+        priority=70,
+    )
+
+    result = resolve_candidates((resolution, overlay))
+
+    assert result.claims.tags == ("dub",)
+    assert result.claims.resolution == "1080p"
+    assert result.decision_for("overlay").status is DecisionStatus.SELECTED
+    assert result.decision_for("resolution").status is DecisionStatus.SELECTED
+
+
+def test_shadowed_candidate_can_keep_its_span_in_title() -> None:
+    strong = _candidate(
+        "strong",
+        start=0,
+        end=4,
+        claims=Claims(episode=1),
+        priority=100,
+    )
+    ambiguous = _candidate(
+        "ambiguous",
+        start=8,
+        end=9,
+        claims=Claims(episode=2),
+        priority=70,
+        shadowed_span_policy=ShadowedSpanPolicy.KEEP,
+    )
+
+    result = resolve_candidates((ambiguous, strong))
+
+    assert result.decision_for("ambiguous").status is DecisionStatus.SHADOWED
+    assert result.excluded_spans == (Span(0, 0, 4),)
+
+
+def test_shadowed_candidate_excludes_marker_but_keeps_dependent_span() -> None:
+    winner = _candidate(
+        "winner", start=0, end=4, claims=Claims(episode=1), priority=100
+    )
+    marker = Span(0, 8, 12)
+    dependent_title = Span(0, 12, 24)
+    loser = Candidate(
+        id="loser",
+        rule_id="loser",
+        spans=(marker, dependent_title),
+        shadowed_spans=(marker,),
+        claims=Claims(episode=2, episode_title="Dependent title"),
+        priority=70,
+    )
+
+    result = resolve_candidates((loser, winner))
+
+    decision = result.decision_for("loser")
+    assert decision.status is DecisionStatus.SHADOWED
+    assert decision.excluded_spans == (marker,)
+    assert result.excluded_spans == (Span(0, 0, 4), marker)
+
+
+def test_repeatable_claims_are_case_insensitively_deduplicated() -> None:
+    first = _candidate(
+        "codec-a",
+        start=0,
+        end=4,
+        claims=Claims(codecs=("HEVC",), tags=("tag-a",)),
+        priority=10,
+    )
+    second = _candidate(
+        "codec-b",
+        start=5,
+        end=9,
+        claims=Claims(codecs=("hevc", "AVC"), tags=("TAG-A",)),
+        priority=10,
+    )
+
+    result = resolve_candidates((second, first))
+
+    assert result.claims.codecs == ("HEVC", "AVC")
+    assert result.claims.tags == ("tag-a",)
+    assert ClaimField.CODECS in result.decision_for("codec-a").won_fields
+
+
+def test_duplicate_candidate_ids_are_rejected() -> None:
+    first = _candidate(
+        "duplicate", start=0, end=1, claims=Claims(episode=1), priority=1
+    )
+    second = _candidate(
+        "duplicate", start=2, end=3, claims=Claims(episode=2), priority=2
+    )
+
+    with pytest.raises(ValueError, match="duplicate candidate id"):
+        resolve_candidates((first, second))
+
+
+def test_equal_priority_conflicting_claims_emit_stable_warning() -> None:
+    first = _candidate(
+        "episode-a", start=0, end=2, claims=Claims(episode=1), priority=70
+    )
+    second = _candidate(
+        "episode-b", start=4, end=6, claims=Claims(episode=2), priority=70
+    )
+
+    forward = resolve_candidates((first, second))
+    reverse = resolve_candidates((second, first))
+
+    assert forward == reverse
+    assert len(forward.warnings) == 1
+    assert forward.warnings[0].field is ClaimField.EPISODE
+    assert forward.warnings[0].candidate_ids == ("episode-a", "episode-b")
+
+
+def test_specificity_resolves_same_priority_without_ambiguity_warning() -> None:
+    explicit = _candidate(
+        "explicit",
+        start=0,
+        end=2,
+        claims=Claims(episode=1),
+        priority=70,
+        specificity=2,
+    )
+    generic = _candidate(
+        "generic", start=4, end=6, claims=Claims(episode=2), priority=70
+    )
+
+    result = resolve_candidates((generic, explicit))
+
+    assert result.claims.episode == 1
+    assert result.warnings == ()
+
+
+def test_fully_shadowed_low_priority_tie_does_not_warn() -> None:
+    winner = _candidate(
+        "winner", start=0, end=2, claims=Claims(episode=1), priority=100
+    )
+    low_a = _candidate("low-a", start=4, end=6, claims=Claims(episode=2), priority=70)
+    low_b = _candidate("low-b", start=8, end=10, claims=Claims(episode=3), priority=70)
+
+    result = resolve_candidates((low_b, winner, low_a))
+
+    assert result.claims.episode == 1
+    assert result.warnings == ()

backend/src/test/test_tokenizer_trace.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+from dataclasses import FrozenInstanceError
+
+import pytest
+
+from module.parser.analyser.tokenizer.candidate import (
+    Candidate,
+    Claims,
+    DecisionStatus,
+    Observation,
+    Span,
+)
+from module.parser.analyser.tokenizer.parser import (
+    parse_release_title,
+    parse_release_title_with_trace,
+)
+from module.parser.analyser.tokenizer.resolver import resolve_candidates
+from module.parser.analyser.tokenizer.result import MediaType
+from module.parser.analyser.tokenizer.trace import ParseTrace, TraceSegment
+
+
+def test_trace_canonicalizes_diagnostic_input_order() -> None:
+    early_observation = Observation(
+        id="episode-observation",
+        rule_id="episode",
+        kind="episode",
+        span=Span(0, 8, 10),
+        text="01",
+        captures=("01",),
+    )
+    late_observation = Observation(
+        id="resolution-observation",
+        rule_id="resolution",
+        kind="resolution",
+        span=Span(1, 0, 5),
+        text="1080p",
+    )
+    episode = Candidate(
+        id="episode",
+        rule_id="episode",
+        spans=(early_observation.span,),
+        claims=Claims(episode=1),
+        priority=90,
+        observation_ids=(early_observation.id,),
+    )
+    resolution_candidate = Candidate(
+        id="resolution",
+        rule_id="resolution",
+        spans=(late_observation.span,),
+        claims=Claims(resolution="1080p"),
+        priority=10,
+        observation_ids=(late_observation.id,),
+    )
+    candidates = (resolution_candidate, episode)
+    resolution = resolve_candidates(candidates)
+
+    trace = ParseTrace.from_resolution(
+        raw="Anime - 01 [1080p]",
+        normalized="Anime - 01 [1080p]",
+        resolution=resolution,
+        segments=(
+            TraceSegment(1, "1080p", 11, 18, 12, 17, "square"),
+            TraceSegment(0, "Anime - 01 ", 0, 11, 0, 11),
+        ),
+        observations=(late_observation, early_observation),
+        candidates=candidates,
+        residuals=("Anime", ""),
+    )
+
+    assert tuple(segment.index for segment in trace.segments) == (0, 1)
+    assert tuple(item.id for item in trace.observations) == (
+        "episode-observation",
+        "resolution-observation",
+    )
+    assert tuple(item.id for item in trace.candidates) == ("episode", "resolution")
+    assert trace.decision_for("episode").status is DecisionStatus.SELECTED
+
+    attributable_spans = {
+        span for decision in trace.decisions for span in decision.excluded_spans
+    }
+    for excluded in trace.excluded_spans:
+        cursor = excluded.start
+        for span in sorted(attributable_spans):
+            if span.segment != excluded.segment or span.end <= cursor:
+                continue
+            assert span.start <= cursor
+            cursor = max(cursor, span.end)
+            if cursor >= excluded.end:
+                break
+        assert cursor >= excluded.end
+
+
+def test_trace_exposes_title_rejections_and_is_frozen() -> None:
+    winner = Candidate(
+        id="movie",
+        rule_id="movie",
+        spans=(Span(0, 0, 4),),
+        claims=Claims(),
+        priority=100,
+        blocks=frozenset({"ambiguous"}),
+    )
+    rejected = Candidate(
+        id="title-number",
+        rule_id="bare-number",
+        spans=(Span(0, 8, 9),),
+        claims=Claims(episode=0),
+        priority=10,
+        conflict_tags=frozenset({"ambiguous"}),
+        preserve_as_title_on_conflict=True,
+    )
+    resolution = resolve_candidates((rejected, winner))
+    trace = ParseTrace.from_resolution(
+        raw="Movie X 0",
+        normalized="Movie X 0",
+        resolution=resolution,
+        candidates=(rejected, winner),
+    )
+
+    assert trace.rejected_as_title == ("title-number",)
+    with pytest.raises(FrozenInstanceError):
+        trace.raw = "changed"  # type: ignore[misc]
+
+
+def test_trace_missing_candidate_decision_raises_key_error() -> None:
+    trace = ParseTrace(raw="Anime", normalized="Anime")
+
+    with pytest.raises(KeyError):
+        trace.decision_for("missing")
+
+
+def test_real_parser_trace_reproduces_public_result_and_accounts_for_masks() -> None:
+    raw = "[Group] Anime - 01 [1080p WEB-DL HEVC AAC MKV]"
+
+    outcome = parse_release_title_with_trace(raw)
+
+    assert outcome.result == parse_release_title(raw)
+    assert outcome.result is not None
+    assert outcome.result.episode == 1
+    assert outcome.trace.claims.episode == 1
+    assert outcome.result.codecs == ("HEVC",)
+    assert outcome.result.audio == ("AAC",)
+    assert len(outcome.trace.candidates) == len(outcome.trace.decisions)
+    assert {candidate.id for candidate in outcome.trace.candidates} == {
+        decision.candidate_id for decision in outcome.trace.decisions
+    }
+
+    excluded_source_spans = tuple(
+        span for decision in outcome.trace.decisions for span in decision.excluded_spans
+    )
+    for excluded in outcome.trace.excluded_spans:
+        for position in range(excluded.start, excluded.end):
+            assert any(
+                source.segment == excluded.segment
+                and source.start <= position < source.end
+                for source in excluded_source_spans
+            )
+
+
+def test_real_parser_trace_preserves_movie_title_number_with_reason() -> None:
+    outcome = parse_release_title_with_trace("[Group] 劇場版 86 - 0 [1080p WEB-DL]")
+
+    assert outcome.result is not None
+    assert outcome.result.media_type is MediaType.MOVIE
+    assert outcome.result.episode is None
+    assert outcome.result.title_en == "86 - 0"
+    assert any(
+        candidate_id.startswith("episode.trailing:")
+        for candidate_id in outcome.trace.rejected_as_title
+    )
+
+
+def test_empty_input_still_returns_an_explainable_outcome() -> None:
+    outcome = parse_release_title_with_trace("  ")
+
+    assert outcome.result is None
+    assert outcome.trace.raw == "  "
+    assert outcome.trace.normalized == ""
+    assert outcome.trace.candidates == ()
+    assert outcome.trace.decisions == ()
+
+
+@pytest.mark.parametrize(
+    "raw",
+    (
+        "Anime S01E01 Season 1 Episode 1 [1080p]",
+        "Anime - 01 [1080p][1080P]",
+        "Anime - 01 [WEB-DL][web_dl]",
+    ),
+)
+def test_equivalent_spellings_do_not_emit_ambiguity_warnings(raw: str) -> None:
+    outcome = parse_release_title_with_trace(raw)
+
+    assert outcome.result is not None
+    assert outcome.trace.warnings == ()
+
+
+@pytest.mark.parametrize(
+    "raw",
+    (
+        "[Group] Anime (Alt [Nested]) - 01 [1080p]",
+        "[Group] Anime - 01 [1080p",
+    ),
+)
+def test_nested_or_unclosed_segment_coordinates_remain_valid(raw: str) -> None:
+    outcome = parse_release_title_with_trace(raw)
+
+    assert outcome.result is not None
+    for segment in outcome.trace.segments:
+        assert (
+            outcome.trace.normalized[segment.content_start : segment.content_end]
+            == segment.text
+        )
+        assert segment.outer_start <= segment.content_start
+        assert segment.content_end <= segment.outer_end