RAG อ้างเอกสารเก่า? ตรวจ Citation ให้ตรงรุ่นด้วย SHA-256 และ UTF-8

"""ตรวจตำแหน่งคำอ้างอิงกับ revision ปัจจุบันที่แอปเชื่อถือ (Python 3.12+)."""

from __future__ import annotations

from dataclasses import dataclass
from hashlib import sha256
from typing import Mapping


@dataclass(frozen=True)
class SourceSnapshot:
    source_id: str
    revision: str
    canonical_utf8: bytes
    expected_sha256: str


@dataclass(frozen=True)
class CitationCandidate:
    source_id: str
    revision: str
    digest: str
    start: int
    end: int
    quote: str


@dataclass(frozen=True)
class GateResult:
    status: str
    reason: str


def validate_citation(
    candidate: CitationCandidate,
    trusted_manifest: Mapping[str, SourceSnapshot],
) -> GateResult:
    """ตรวจ citation เดียวกับสำเนา manifest ที่ผู้เรียกตรึงไว้ตลอดการตรวจหนึ่งครั้ง."""
    if type(candidate) is not CitationCandidate:
        return GateResult("REJECTED", "invalid_candidate_type")
    if not all(type(value) is str and value for value in
               (candidate.source_id, candidate.revision, candidate.digest, candidate.quote)):
        return GateResult("REJECTED", "invalid_text_field")
    if len(candidate.quote) > 1_000_000:
        return GateResult("REJECTED", "quote_too_large")
    try:
        quote_bytes = candidate.quote.encode("utf-8", errors="strict")
    except UnicodeError:
        return GateResult("REJECTED", "invalid_quote_utf8")
    if len(quote_bytes) > 1_000_000:
        return GateResult("REJECTED", "quote_too_large")
    if type(candidate.start) is not int or type(candidate.end) is not int:
        return GateResult("REJECTED", "invalid_offset_type")
    # manifest เป็น mapping ที่แอปเชื่อถือและตรึงไว้ตลอดการตรวจหนึ่งครั้ง
    try:
        if len(trusted_manifest) > 1000:
            return GateResult("REJECTED", "manifest_too_large")
        manifest = dict(trusted_manifest)
    except (TypeError, ValueError):
        return GateResult("REJECTED", "invalid_trusted_manifest")
    if len(manifest) > 1000:
        return GateResult("REJECTED", "manifest_too_large")
    source = manifest.get(candidate.source_id)
    if type(source) is not SourceSnapshot:
        return GateResult("REJECTED", "source_not_allowlisted")
    if type(source.source_id) is not str or type(source.revision) is not str:
        return GateResult("REJECTED", "invalid_manifest_entry")
    if source.source_id != candidate.source_id:
        return GateResult("REJECTED", "manifest_source_mismatch")
    if type(source.canonical_utf8) is not bytes:
        return GateResult("REJECTED", "invalid_source_bytes")
    if (type(source.expected_sha256) is not str or len(source.expected_sha256) != 64
            or any(char not in "0123456789abcdef" for char in source.expected_sha256)):
        return GateResult("REJECTED", "invalid_manifest_digest")
    if len(source.canonical_utf8) > 1_000_000:
        return GateResult("REJECTED", "source_too_large")
    try:
        source.canonical_utf8.decode("utf-8", errors="strict")
    except UnicodeError:
        return GateResult("REJECTED", "invalid_source_utf8")
    if candidate.revision != source.revision:
        return GateResult("REJECTED", "stale_revision")
    actual_digest = sha256(source.canonical_utf8).hexdigest()
    if actual_digest != source.expected_sha256:
        return GateResult("REJECTED", "source_digest_mismatch")
    if candidate.digest != source.expected_sha256:
        return GateResult("REJECTED", "candidate_digest_mismatch")
    data = source.canonical_utf8
    if not (0 <= candidate.start < candidate.end <= len(data)):
        return GateResult("REJECTED", "span_out_of_bounds")
    selected = data[candidate.start:candidate.end]
    try:
        selected.decode("utf-8", errors="strict")
    except UnicodeError:
        return GateResult("REJECTED", "invalid_utf8_boundary_or_quote")
    if selected != quote_bytes:
        return GateResult("REJECTED", "quote_mismatch")
    return GateResult("QUOTE_VERIFIED", "current_source_quote_located")

"""แสดง fixture สังเคราะห์สำหรับ current และ stale citation; Python 3.12+ only."""

import json
from hashlib import sha256

from citation_gate import CitationCandidate, SourceSnapshot, validate_citation


V1_TEXT = "ข้อกำหนดวิดีโอ: ต้องขออนุมัติก่อนเผยแพร่"
V2_TEXT = "ข้อกำหนดวิดีโอ: ต้องได้รับอนุมัติก่อนส่งออก"
SOURCE_ID = "video-policy-demo"


def source_snapshot(text: str, revision: str) -> SourceSnapshot:
    raw = text.encode("utf-8")
    return SourceSnapshot(SOURCE_ID, revision, raw, sha256(raw).hexdigest())


def citation_for(source: SourceSnapshot, quote: str) -> CitationCandidate:
    raw_quote = quote.encode("utf-8")
    start = source.canonical_utf8.index(raw_quote)
    return CitationCandidate(
        source_id=source.source_id,
        revision=source.revision,
        digest=sha256(source.canonical_utf8).hexdigest(),
        start=start,
        end=start + len(raw_quote),
        quote=quote,
    )


def main() -> None:
    v1 = source_snapshot(V1_TEXT, "v1")
    v2 = source_snapshot(V2_TEXT, "v2")
    old_citation = citation_for(v1, "ต้องขออนุมัติก่อนเผยแพร่")
    cases = (
        ("current-v1", {SOURCE_ID: v1}),
        ("stale-v1-after-v2", {SOURCE_ID: v2}),
    )
    output = []
    for label, manifest in cases:
        result = validate_citation(old_citation, manifest)
        output.append({"case": label, "status": result.status, "reason": result.reason})
    print(json.dumps(output, ensure_ascii=False, separators=(",", ":")))


if __name__ == "__main__":
    main()

"""Regression cases for the bounded mechanical citation gate."""

import unittest
from dataclasses import FrozenInstanceError
from hashlib import sha256

from citation_gate import CitationCandidate, SourceSnapshot, validate_citation


V1 = "ข้อกำหนดวิดีโอ: ต้องขออนุมัติก่อนเผยแพร่"
V2 = "ข้อกำหนดวิดีโอ: ต้องได้รับอนุมัติก่อนส่งออก"


def snapshot(text: str, revision: str = "v1") -> SourceSnapshot:
    raw = text.encode("utf-8")
    return SourceSnapshot("video-policy-demo", revision, raw, sha256(raw).hexdigest())


def candidate(src: SourceSnapshot, quote: str, start: int | None = None) -> CitationCandidate:
    raw = src.canonical_utf8
    q = quote.encode("utf-8")
    offset = raw.index(q) if start is None else start
    return CitationCandidate(src.source_id, src.revision, sha256(raw).hexdigest(),
                             offset, offset + len(q), quote)


class CitationGateTests(unittest.TestCase):
    def test_accepts_current_exact_thai_quote(self):
        src = snapshot(V2, "v2")
        self.assertEqual(validate_citation(candidate(src, "ต้องได้รับอนุมัติก่อนส่งออก"), {src.source_id: src}).status,
                         "QUOTE_VERIFIED")

    def test_rejects_old_revision_after_pointer_advances_even_if_old_hash_matches(self):
        old = snapshot(V1, "v1")
        current = snapshot(V2, "v2")
        result = validate_citation(candidate(old, "ต้องขออนุมัติก่อนเผยแพร่"), {current.source_id: current})
        self.assertEqual(result.reason, "stale_revision")

    def test_unallowlisted_source_is_rejected(self):
        src = snapshot(V1)
        item = candidate(src, "ต้องขออนุมัติก่อนเผยแพร่")
        item = CitationCandidate("other", item.revision, item.digest, item.start, item.end, item.quote)
        self.assertEqual(validate_citation(item, {src.source_id: src}).reason,
                         "source_not_allowlisted")

    def test_modified_source_bytes_fail_digest(self):
        src = snapshot(V1)
        item = candidate(src, "ต้องขออนุมัติก่อนเผยแพร่")
        changed = SourceSnapshot(src.source_id, src.revision, V2.encode("utf-8"),
                                 src.expected_sha256)
        self.assertEqual(validate_citation(item, {changed.source_id: changed}).reason,
                         "source_digest_mismatch")

    def test_candidate_cannot_choose_a_different_digest(self):
        src = snapshot(V1)
        item = candidate(src, "ต้องขออนุมัติก่อนเผยแพร่")
        forged = CitationCandidate(item.source_id, item.revision, "0" * 64,
                                   item.start, item.end, item.quote)
        self.assertEqual(validate_citation(forged, {src.source_id: src}).reason,
                         "candidate_digest_mismatch")

    def test_invalid_or_boolean_offsets_are_rejected(self):
        src = snapshot(V1)
        item = candidate(src, "ต้องขออนุมัติก่อนเผยแพร่")
        for start, end in ((True, item.end), (-1, item.end), (item.end, item.end),
                           (item.start, len(src.canonical_utf8) + 1)):
            bad = CitationCandidate(item.source_id, item.revision, item.digest,
                                    start, end, item.quote)
            self.assertEqual(validate_citation(bad, {src.source_id: src}).status, "REJECTED")

    def test_byte_offset_splitting_thai_utf8_sequence_is_rejected(self):
        src = snapshot(V1)
        item = candidate(src, "ข้อกำหนด")
        bad = CitationCandidate(item.source_id, item.revision, item.digest,
                                item.start + 1, item.end, item.quote)
        self.assertEqual(validate_citation(bad, {src.source_id: src}).reason,
                         "invalid_utf8_boundary_or_quote")

    def test_wrong_display_quote_is_rejected(self):
        src = snapshot(V1)
        item = candidate(src, "ต้องขออนุมัติก่อนเผยแพร่")
        bad = CitationCandidate(item.source_id, item.revision, item.digest,
                                item.start, item.end, "ส่งออก")
        self.assertEqual(validate_citation(bad, {src.source_id: src}).reason, "quote_mismatch")

    def test_manifest_source_identity_must_match_lookup_key(self):
        src = snapshot(V1)
        misbound = SourceSnapshot("different-id", src.revision, src.canonical_utf8,
                                  src.expected_sha256)
        self.assertEqual(validate_citation(candidate(src, "ต้องขออนุมัติก่อนเผยแพร่"),
                                           {src.source_id: misbound}).reason,
                         "manifest_source_mismatch")

    def test_snapshot_records_are_immutable(self):
        src = snapshot(V1)
        with self.assertRaises(FrozenInstanceError):
            src.revision = "v2"  # type: ignore[misc]


if __name__ == "__main__":
    unittest.main()

# ตัวอย่าง Citation Gate

ตัวอย่างจำลองสำหรับ Python 3.12 ขึ้นไป ใช้ standard library เท่านั้น ไม่มี environment variable และไม่ติดตั้งแพ็กเกจ ไม่เรียก LLM/API และไม่ต่อฐานข้อมูล ค่าใช้จ่าย API เป็นศูนย์; การรันใช้ CPU ในเครื่องตามปกติและไม่มีการวัดต้นทุนหรือ latency

## วิธีลองด้วยตนเอง

ติดตั้ง Python 3.12 หรือใหม่กว่าโดยใช้ runtime ที่เชื่อถือได้ แล้วจากโฟลเดอร์นี้รันคำสั่ง:

```text
cd code
python demo.py
python -m unittest -v
```

ไม่มี `pip install` หรือ dependency ภายนอก input ของ demo เป็น fixture สมมติภายในไฟล์เท่านั้น ไม่รับข้อมูลผู้ใช้ ไม่เชื่อม network และไม่แก้ระบบภายนอก ผลตัวอย่างด้านล่างเป็น **ผลคาดหมาย ไม่ใช่ผลรันจริง**:

```json
[{"case":"current-v1","status":"QUOTE_VERIFIED","reason":"current_source_quote_located"},{"case":"stale-v1-after-v2","status":"REJECTED","reason":"stale_revision"}]
```

เมื่อรันทดสอบคาดว่าจะพบ 10 กรณีและจบด้วย `OK` ตัวอย่างและชุดทดสอบยังไม่ได้รันในการจัดทำบทความ สถานะจึงเป็น **NOT RUN**

เพื่อดูผลหนึ่งรายการ ให้สร้าง `SourceSnapshot` สำหรับ `video-policy-demo`, revision `v2`, ข้อความ UTF-8 `ข้อกำหนดวิดีโอ: ต้องได้รับอนุมัติก่อนส่งออก` และ `expected_sha256` ที่แอปคำนวณขณะ ingest จาก bytes ชุดนั้น จากนั้นสร้าง candidate ที่มี digest เดียวกัน, `start=44`/`end=125` ซึ่งเป็น byte span ครึ่งเปิดของ quote `ต้องได้รับอนุมัติก่อนส่งออก`, และใส่ quote เดียวกัน ผลคาดหมายคือ `GateResult(status='QUOTE_VERIFIED', reason='current_source_quote_located')` หาก manifest เปลี่ยนเป็น v3 แล้วส่ง candidate v2 ผลจะเป็น `stale_revision` แม้ hash ของ v2 ยังถูกต้อง

## สัญญาและขอบเขต

แอปเป็นเจ้าของ manifest ที่เชื่อถือได้และเป็นผู้เลือก revision ปัจจุบัน ส่วน candidate เป็นข้อมูลไม่น่าเชื่อถือ จึงไม่สามารถกำหนด hash authority หรือขอเลือก historical revision เองได้ manifest เก็บ `expected_sha256` ที่แอปคำนวณและผูกกับ bytes ระหว่าง ingest; validator คำนวณ hash จาก bytes ปัจจุบันซ้ำแล้วเทียบกับ manifest ก่อนเทียบ digest ของ candidate โค้ดตรวจหนึ่ง citation ต่อการเรียก จำกัด manifest ไว้ 1,000 รายการและ source กับ quote ไว้ไม่เกิน 1 MB; ระบบจริงควรปรับเพดานตามขนาดรับเข้าและทรัพยากรของตนเอง ผู้เรียกต้องส่ง manifest mapping ที่เชื่อถือได้และคงที่ตลอดการตรวจ

`QUOTE_VERIFIED` หมายถึงพบ quote ตรงกันใน bytes ของ source revision ที่ manifest ระบุว่าเป็นปัจจุบัน พร้อม digest และขอบเขต UTF-8 ที่ถูกต้องเท่านั้น ไม่ยืนยันตัวตน/ความน่าเชื่อถือของผู้เผยแพร่ ไม่ยืนยันว่านโยบายยังเป็นจริงในโลกภายนอก ไม่ตรวจว่า quote สนับสนุนคำกล่าวอ้างทั้งประโยคหรือไม่ และไม่อนุญาตให้ทำ action ใด

ก่อนใช้จริงต้องนิยาม canonicalization ให้ชัด เช่น line ending, Unicode normalization, การดึง HTML และเวลาอัปเดต revision/hash ร่วมกัน กติกาการเปลี่ยนเหล่านี้ต้องสร้าง revision ใหม่ การคัดลอก manifest ระหว่างเรียกเป็น snapshot แบบ in-memory เพื่อให้การตรวจครั้งเดียวใช้ค่าคงที่ ไม่ได้แก้ race condition ระหว่างระบบ ingest หลายโหนด, ACL/สิทธิ์อ่าน, การเลือกแหล่งทางการ, หรือการดึงข้อมูลสด

ไม่มี credential, การติดตั้ง, API, ค่าใช้จ่ายบริการ หรือผลการทดสอบ runtime ในตัวอย่างนี้