File size: 4,027 Bytes
ab5ea78
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
"""Verbatim span validation — the primary anti-hallucination control.

Three rules that must not be relaxed:

1. **Normalise whitespace only.** Not case, not punctuation, not diacritics.
   Every additional normalisation is a hole a fabrication can fit through.
2. **On failure the FIELD becomes `None`** and the rejection is recorded, so a
   reviewer can see what the control caught rather than only what it let past.
3. **Never repair a failed span** by rewriting it to something that does match.
   A repaired span is an unfalsifiable claim, which is precisely what this
   control exists to prevent.

If the provenance span itself is not verbatim, *every* guarded field on the
entry is rejected: the entry's only link to evidence is broken, so nothing on it
can be trusted.
"""

from __future__ import annotations

import re
import unicodedata

from ..models import Branch, Chunk, RejectedField

# Fields carrying a factual claim, and therefore span-guarded.
GUARDED_FIELDS: dict[str, tuple[str, ...]] = {
    "glossary": ("definition", "full_name", "source_wording", "formula_latex", "interpretation"),
    "rule": ("statement", "condition", "consequence"),
    "formula": ("formula_latex",),
    # Generation, not extraction — it cannot be span-checked at all.
    "summary": (),
}


def normalise_ws(text: str) -> str:
    return re.sub(r"\s+", " ", unicodedata.normalize("NFKC", text)).strip()


def span_present(span: str, source: str) -> bool:
    if not span or not span.strip():
        return False
    return normalise_ws(span) in normalise_ws(source)


def evidence_text(chunk_ids: list[str], chunks: list[Chunk]) -> str:
    """The text a span is validated against.

    Includes each chunk's HEADING as well as its body. The heading is part of
    the source document and is often where a term is formally named — the
    reference standard heads a section "Physical of Availability (PA)" while
    the body never repeats the phrase. Excluding it would reject a correct,
    verbatim quotation of the document's own section title, which is precisely
    the wording we are required to preserve.
    """
    by_id = {c.chunk_id: c for c in chunks}
    parts: list[str] = []
    for chunk_id in chunk_ids:
        chunk = by_id.get(chunk_id)
        if chunk is None:
            continue
        if chunk.heading:
            parts.append(chunk.heading)
        parts.append(chunk.text)
    return "\n".join(parts)


def validate_entry(
    entry, branch: Branch, source_text: str, label: str
) -> tuple[object, list[RejectedField]]:
    """Returns `(entry, rejections)`; the entry is mutated in place.

    Also checks each guarded field's own value against the source where the
    field is expected to be quoted: `source_wording` and `full_name` are
    literal transcriptions, so a value that cannot be located is a silent
    normalisation — exactly the failure this pipeline is required to surface.
    """
    rejections: list[RejectedField] = []
    guarded = GUARDED_FIELDS.get(branch, ())
    span_ok = span_present(entry.provenance.span, source_text)

    for field in guarded:
        value = getattr(entry, field, None)
        if value is None:
            continue

        if not span_ok:
            reason = "provenance.span not found verbatim in evidence"
        elif field in _TRANSCRIBED and not span_present(str(value), source_text):
            reason = f"{field} is not a verbatim transcription of the source"
        else:
            continue

        rejections.append(
            RejectedField(
                entry_term=label,
                field=field,
                offending_value=str(value)[:300],
                reason=reason,
                branch=branch,
            )
        )
        setattr(entry, field, None)

    return entry, rejections


# Fields that claim to be copied from the document word for word. A definition
# may legitimately be assembled across sentences; a "full name" may not.
_TRANSCRIBED = frozenset({"full_name", "source_wording"})