imbad0202__academic-research-skills/scripts/test_claim_intent_manifest.py

"""Manifest tests for v3.8 claim_ref_alignment_audit_agent (T-M1, T-M2, T-M3).

Per spec §7.3 in
docs/design/2026-05-15-issue-103-claim-alignment-audit-spec.md.

These tests pin manifest-level contracts that pipeline-level T-P1..T-P11 do
not exercise:

- **T-M1 — Three-set diff**: emitted ∩ intended ∩ supported with drift
  detection in both directions (`EMITTED_NOT_INTENDED` from drafted prose
  introducing claims the writer did not pre-commit to, `INTENDED_NOT_EMITTED`
  from manifest claims dropped during drafting). Verifies the §"Manifest
  cross-reference (D6)" three-set-diff table directly.

- **T-M2 — Missing manifest**: `manifests=[]` plus the `SENTINEL_MANIFEST_ID`
  on each citation must keep every defect_stage path emitting. Per §9
  acceptance bullet `claim_intent_manifest absent → MANIFEST-MISSING
  advisory + fallback flow exercised in test`. This test pins the
  fallback's pipeline-side invariants (all six substantive defect_stages
  still routable; sentinel propagates through to entries); the
  MANIFEST-MISSING annotation surface itself lives in the finalizer (spec
  §13 step 8) and is not covered here.

- **T-M3 — Constraint inheritance**: an `MNC-{m}` global constraint MUST
  apply to a citation whose claim-level entry only declares an `NC-C{n}-{m}`
  constraint. The judge invocation MUST receive both, in `constraint_id`
  sorted order (`_active_constraints_for_claim` invariant per spec §4 step
  3 + M-INV-3).

Spec §7 names the file `tests/test_claim_intent_manifest.py`. Repo
convention places tests under `scripts/test_*.py` (CI uses
`python -m unittest scripts.test_*`); the spec-named stem is preserved.

Run:
    python -m unittest scripts.test_claim_intent_manifest -v
"""
from __future__ import annotations

import unittest
from typing import Any, Callable

try:
    from scripts.claim_audit_pipeline import run_audit_pipeline
    from scripts._claim_audit_constants import SENTINEL_MANIFEST_ID

    _MODULE_IMPORT_ERR: Exception | None = None
except Exception as exc:  # pragma: no cover — RED-phase import pathway
    _MODULE_IMPORT_ERR = exc
    SENTINEL_MANIFEST_ID = "M-0000-00-00T00:00:00Z-0000"

    def run_audit_pipeline(*args: Any, **kwargs: Any) -> Any:
        raise _MODULE_IMPORT_ERR  # type: ignore[misc]


MANIFEST_ID = "M-2026-05-15T10:00:00Z-a1b2"
MANIFEST_ID_B = "M-2026-05-15T10:00:01Z-c3d4"
AUDIT_RUN_ID = "2026-05-15T10:10:00Z-9f8e"
NOW = "2026-05-15T10:11:00Z"


def _claim(
    *,
    claim_id: str,
    claim_text: str,
    ncs: list[dict[str, str]] | None = None,
    planned_refs: list[str] | None = None,
) -> dict[str, Any]:
    out: dict[str, Any] = {
        "claim_id": claim_id,
        "claim_text": claim_text,
        "intended_evidence_kind": "empirical",
        "planned_refs": planned_refs or [],
    }
    if ncs:
        out["negative_constraints"] = ncs
    return out


def _manifest(
    *,
    manifest_id: str = MANIFEST_ID,
    claims: list[dict[str, Any]],
    mncs: list[dict[str, str]] | None = None,
) -> dict[str, Any]:
    return {
        "manifest_version": "1.0",
        "manifest_id": manifest_id,
        "emitted_by": "synthesis_agent",
        "emitted_at": "2026-05-15T09:55:00Z",
        "claims": claims,
        "manifest_negative_constraints": mncs or [],
    }


def _citation(
    *,
    claim_id: str,
    claim_text: str,
    ref_slug: str = "smith2024preprints",
    anchor_kind: str = "page",
    anchor_value: str = "12",
    section_path: str = "3. Results > 3.1 Overview",
    scoped_manifest_id: str = MANIFEST_ID,
) -> dict[str, Any]:
    return {
        "claim_id": claim_id,
        "scoped_manifest_id": scoped_manifest_id,
        "claim_text": claim_text,
        "ref_slug": ref_slug,
        "anchor_kind": anchor_kind,
        "anchor_value": anchor_value,
        "section_path": section_path,
    }


def _config(**overrides: Any) -> dict[str, Any]:
    base: dict[str, Any] = {
        "max_claims_per_paper": 100,
        "judge_model": "gpt-5.5-xhigh",
        "gold_set_path": None,
        "cache_dir": None,
    }
    base.update(overrides)
    return base


def _retrieval_ok(
    *,
    excerpt: str = "The cited page reports the figure verbatim.",
) -> Callable[[dict[str, Any]], dict[str, Any]]:
    def fn(citation: dict[str, Any]) -> dict[str, Any]:
        return {"ref_retrieval_method": "api", "retrieved_excerpt": excerpt}

    return fn


def _retrieval_failed_paywall() -> Callable[[dict[str, Any]], dict[str, Any]]:
    def fn(citation: dict[str, Any]) -> dict[str, Any]:
        return {"ref_retrieval_method": "failed", "retrieved_excerpt": None}

    return fn


def _retrieval_not_found() -> Callable[[dict[str, Any]], dict[str, Any]]:
    def fn(citation: dict[str, Any]) -> dict[str, Any]:
        return {"ref_retrieval_method": "not_found", "retrieved_excerpt": None}

    return fn


def _retrieval_dispatch(
    by_slug: dict[str, dict[str, Any]],
) -> Callable[[dict[str, Any]], dict[str, Any]]:
    """Per-slug retrieval router so a single run can mix paywall / 200 / 404."""

    def fn(citation: dict[str, Any]) -> dict[str, Any]:
        return by_slug[citation["ref_slug"]]

    return fn


def _judge_dispatch(
    by_claim_text: dict[str, dict[str, Any]],
) -> Callable[..., dict[str, Any]]:
    """Per-claim-text judge router for T-M1's mixed SUPPORTED / UNSUPPORTED / AMBIGUOUS."""
    captured_kwargs: list[dict[str, Any]] = []

    def fn(**kwargs: Any) -> dict[str, Any]:
        captured_kwargs.append(kwargs)
        verdict = by_claim_text[kwargs["claim_text"]]
        return verdict

    fn.captured = captured_kwargs  # type: ignore[attr-defined]
    return fn


class _ManifestTestBase(unittest.TestCase):
    """Skip cleanly when scripts.claim_audit_pipeline is not yet importable.

    During the RED phase (Step 4 of the TDD plan in spec §13), the test
    suite imports cleanly even when the module is missing — the contract
    documents the wished-for API.
    """

    @classmethod
    def setUpClass(cls) -> None:
        if _MODULE_IMPORT_ERR is not None:
            raise unittest.SkipTest(
                f"scripts.claim_audit_pipeline not importable yet: {_MODULE_IMPORT_ERR!r} "
                "(expected during RED phase — implementation lands in spec §13 step 5)"
            )

    def run_pipeline(self, **kwargs: Any) -> dict[str, list[dict[str, Any]]]:
        defaults: dict[str, Any] = {
            "manifests": [],
            "citations": [],
            "corpus": [],
            "config": _config(),
            "audit_run_id": AUDIT_RUN_ID,
            "now_iso": NOW,
            "retrieve_fn": _retrieval_ok(),
            "judge_fn": _judge_dispatch({}),
        }
        defaults.update(kwargs)
        return run_audit_pipeline(**defaults)


# ---------------------------------------------------------------------------
# T-M1 — Three-set diff (emitted ∩ intended ∩ supported, drift detection).
# ---------------------------------------------------------------------------


class TM1ThreeSetDiff(_ManifestTestBase):
    """T-M1: manifest set-diff produces both INTENDED_NOT_EMITTED and EMITTED_NOT_INTENDED."""

    def test_three_set_diff_emits_both_drift_kinds(self) -> None:
        # Manifest intends 3 claims (C-001, C-002, C-003); none of the
        # manifest's claims carry a negative constraint, so no constraint
        # absorption confounds drift detection (T-P8 absorbs ENTIRE manifest
        # when ANY citation in that manifest VIOLATEs).
        manifest = _manifest(
            claims=[
                _claim(
                    claim_id="C-001",
                    claim_text="Sample preprints accounted for 67% of corpus.",
                    planned_refs=["smith2024preprints"],
                ),
                _claim(
                    claim_id="C-002",
                    claim_text="Inflection point appeared in mid-2024.",
                    planned_refs=["zhao2026"],
                ),
                _claim(
                    claim_id="C-003",
                    claim_text="The Pew baseline established trust decline.",
                    planned_refs=["pew2023"],
                ),
            ],
        )

        # Prose emits 4 citations:
        #   - C-001 with original manifest claim_text  → intended ∩ emitted ∩ supported
        #   - C-002 with original manifest claim_text  → intended ∩ emitted, judge AMBIGUOUS (NOT in supported)
        #   - Drifted EMITTED text 1: "Median half-life dropped to 11 days." (NOT in manifest)
        #   - Drifted EMITTED text 2: "Citation density doubled after 2024." (NOT in manifest)
        #   - C-003 manifest claim is DROPPED (intended ∉ emitted)
        emitted_citations = [
            _citation(
                claim_id="C-001",
                claim_text="Sample preprints accounted for 67% of corpus.",
                ref_slug="smith2024preprints",
            ),
            _citation(
                claim_id="C-002",
                claim_text="Inflection point appeared in mid-2024.",
                ref_slug="zhao2026",
                anchor_value="14",
            ),
            _citation(
                claim_id="C-100",  # Drifted — not in manifest claims[]
                claim_text="Median half-life dropped to 11 days.",
                ref_slug="brown2025",
                anchor_value="22",
            ),
            _citation(
                claim_id="C-101",  # Drifted — not in manifest claims[]
                claim_text="Citation density doubled after 2024.",
                ref_slug="jones2025",
                anchor_value="3",
            ),
        ]

        judge_fn = _judge_dispatch(
            {
                "Sample preprints accounted for 67% of corpus.": {
                    "judgment": "SUPPORTED",
                    "rationale": "Cited page contains the 67% figure verbatim.",
                },
                "Inflection point appeared in mid-2024.": {
                    "judgment": "AMBIGUOUS",
                    "defect_stage_hint": "source_description",
                    "rationale": "Source mentions mid-2024 but does not pinpoint inflection.",
                },
                "Median half-life dropped to 11 days.": {
                    "judgment": "SUPPORTED",
                    "rationale": "Source reports the 11-day figure.",
                },
                "Citation density doubled after 2024.": {
                    "judgment": "UNSUPPORTED",
                    "defect_stage_hint": "source_description",
                    "rationale": "Source shows a 1.4x increase, not 2x.",
                },
            }
        )

        out = self.run_pipeline(
            manifests=[manifest],
            citations=emitted_citations,
            judge_fn=judge_fn,
        )

        # ---- Emitted ∩ Intended (manifest claim_text appears in emitted) ----
        manifest_texts = {c["claim_text"] for c in manifest["claims"]}
        emitted_texts = {c["claim_text"] for c in emitted_citations}
        emitted_intersect_intended = manifest_texts & emitted_texts
        self.assertEqual(
            emitted_intersect_intended,
            {
                "Sample preprints accounted for 67% of corpus.",
                "Inflection point appeared in mid-2024.",
            },
            "Setup invariant: expected exactly 2 emitted∩intended texts",
        )

        # ---- Supported subset (judge=SUPPORTED post-pipeline) ----
        supported_texts = {
            r["claim_text"]
            for r in out["claim_audit_results"]
            if r["judgment"] == "SUPPORTED"
        }
        # Drifted "Median half-life..." judged SUPPORTED but is NOT in manifest,
        # so the supported∩intended is C-001 only.
        self.assertEqual(
            supported_texts & manifest_texts,
            {"Sample preprints accounted for 67% of corpus."},
            "supported∩intended must equal the 1 SUPPORTED claim that is also in manifest",
        )

        # Judge invocation count — one call per emitted citation. Pins
        # idempotence: the same fixture should not silently re-judge the
        # same (claim_text, ref, anchor) tuple twice in a single run.
        self.assertEqual(
            len(judge_fn.captured),
            4,
            "expected exactly one judge call per emitted citation (4 citations)",
        )

        drifts = out["claim_drifts"]
        drift_kinds = sorted(d["drift_kind"] for d in drifts)
        self.assertIn("EMITTED_NOT_INTENDED", drift_kinds)
        self.assertIn("INTENDED_NOT_EMITTED", drift_kinds)

        # ---- INTENDED_NOT_EMITTED: C-003 dropped ----
        # Cardinalities below are derived from setup (3 manifest claims;
        # 4 emitted citations; 2 of them drift; 1 manifest claim dropped),
        # NOT a policy on how many drifts may emit. The set assertions
        # that follow pin the content invariant; exact counts only guard
        # against over-emission of duplicate rows.
        intended_not_emitted = [
            d for d in drifts if d["drift_kind"] == "INTENDED_NOT_EMITTED"
        ]
        self.assertEqual(len(intended_not_emitted), 1, "expected exactly C-003 dropped")
        self.assertEqual(intended_not_emitted[0]["manifest_claim_id"], "C-003")
        self.assertEqual(intended_not_emitted[0]["scoped_manifest_id"], MANIFEST_ID)
        self.assertEqual(
            intended_not_emitted[0]["claim_text"],
            "The Pew baseline established trust decline.",
        )

        # ---- EMITTED_NOT_INTENDED: 2 drifted texts ----
        emitted_not_intended = [
            d for d in drifts if d["drift_kind"] == "EMITTED_NOT_INTENDED"
        ]
        self.assertEqual(
            len(emitted_not_intended),
            2,
            "expected exactly 2 EMITTED_NOT_INTENDED drift rows",
        )
        self.assertEqual(
            {d["claim_text"] for d in emitted_not_intended},
            {
                "Median half-life dropped to 11 days.",
                "Citation density doubled after 2024.",
            },
        )
        for d in emitted_not_intended:
            self.assertIsNone(d["manifest_claim_id"])
            self.assertIsNone(d["scoped_manifest_id"])
            self.assertIsNotNone(d["section_path"])


# ---------------------------------------------------------------------------
# T-M2 — Missing manifest: MANIFEST-MISSING fallback, all defect_stages emit.
# ---------------------------------------------------------------------------


class TM2MissingManifestFallback(_ManifestTestBase):
    """T-M2: manifests=[] + sentinel scoped_manifest_id keeps every defect_stage routable."""

    def test_sentinel_fallback_emits_all_defect_stages(self) -> None:
        # Six citations, one per substantive routing path the agent must
        # cover in MANIFEST-MISSING fallback per §9 acceptance criterion.
        # The sentinel scoped_manifest_id MUST propagate through to every
        # emitted entry (INV-15 sentinel branch).
        citations = [
            # Path 1: anchor=none → RETRIEVAL_FAILED + not_applicable + not_attempted.
            _citation(
                claim_id="C-001",
                claim_text="No locator claim.",
                ref_slug="ref1",
                anchor_kind="none",
                anchor_value="",
                scoped_manifest_id=SENTINEL_MANIFEST_ID,
            ),
            # Path 2: retrieval=failed (paywall) → RETRIEVAL_FAILED + not_applicable + failed.
            _citation(
                claim_id="C-002",
                claim_text="Paywalled source claim.",
                ref_slug="ref2_paywall",
                scoped_manifest_id=SENTINEL_MANIFEST_ID,
            ),
            # Path 3: retrieval=not_found → RETRIEVAL_FAILED + retrieval_existence.
            _citation(
                claim_id="C-003",
                claim_text="Fabricated reference claim.",
                ref_slug="ref3_404",
                scoped_manifest_id=SENTINEL_MANIFEST_ID,
            ),
            # Path 4: SUPPORTED.
            _citation(
                claim_id="C-004",
                claim_text="Supported claim.",
                ref_slug="ref4_ok",
                scoped_manifest_id=SENTINEL_MANIFEST_ID,
            ),
            # Path 5: AMBIGUOUS with citation_anchor hint.
            _citation(
                claim_id="C-005",
                claim_text="Ambiguous anchor claim.",
                ref_slug="ref5_ok",
                scoped_manifest_id=SENTINEL_MANIFEST_ID,
            ),
            # Path 6: UNSUPPORTED with synthesis_overclaim hint.
            _citation(
                claim_id="C-006",
                claim_text="Overclaim synthesis claim.",
                ref_slug="ref6_ok",
                scoped_manifest_id=SENTINEL_MANIFEST_ID,
            ),
        ]

        # Per-slug retrieval routing.
        retrieval_table = {
            "ref1": {"ref_retrieval_method": "api", "retrieved_excerpt": "ignored"},
            "ref2_paywall": {"ref_retrieval_method": "failed", "retrieved_excerpt": None},
            "ref3_404": {"ref_retrieval_method": "not_found", "retrieved_excerpt": None},
            "ref4_ok": {"ref_retrieval_method": "api", "retrieved_excerpt": "OK"},
            "ref5_ok": {"ref_retrieval_method": "api", "retrieved_excerpt": "OK"},
            "ref6_ok": {"ref_retrieval_method": "api", "retrieved_excerpt": "OK"},
        }

        judge_fn = _judge_dispatch(
            {
                "Supported claim.": {
                    "judgment": "SUPPORTED",
                    "rationale": "Source contains the assertion.",
                },
                "Ambiguous anchor claim.": {
                    "judgment": "AMBIGUOUS",
                    "defect_stage_hint": "citation_anchor",
                    "rationale": "Anchor points adjacent passage but not exact.",
                },
                "Overclaim synthesis claim.": {
                    "judgment": "UNSUPPORTED",
                    "defect_stage_hint": "synthesis_overclaim",
                    "rationale": "Claim aggregates beyond what the source supports.",
                },
            }
        )

        out = self.run_pipeline(
            manifests=[],
            citations=citations,
            retrieve_fn=_retrieval_dispatch(retrieval_table),
            judge_fn=judge_fn,
        )

        results = out["claim_audit_results"]
        self.assertEqual(
            len(results),
            6,
            "every citation must emit a row in MANIFEST-MISSING fallback (per §9 'all defect_stages still emit')",
        )

        # Sentinel propagation: every entry carries the sentinel scoped_manifest_id.
        for r in results:
            self.assertEqual(
                r["scoped_manifest_id"],
                SENTINEL_MANIFEST_ID,
                f"scoped_manifest_id MUST stay sentinel under MANIFEST-MISSING fallback; got {r['scoped_manifest_id']!r}",
            )

        by_claim = {r["claim_id"]: r for r in results}

        # Path 1 — anchor=none.
        self.assertEqual(by_claim["C-001"]["judgment"], "RETRIEVAL_FAILED")
        self.assertEqual(by_claim["C-001"]["defect_stage"], "not_applicable")
        self.assertEqual(by_claim["C-001"]["ref_retrieval_method"], "not_attempted")

        # Path 2 — paywall.
        self.assertEqual(by_claim["C-002"]["judgment"], "RETRIEVAL_FAILED")
        self.assertEqual(by_claim["C-002"]["defect_stage"], "not_applicable")
        self.assertEqual(by_claim["C-002"]["ref_retrieval_method"], "failed")

        # Path 3 — fabricated.
        self.assertEqual(by_claim["C-003"]["judgment"], "RETRIEVAL_FAILED")
        self.assertEqual(by_claim["C-003"]["defect_stage"], "retrieval_existence")
        self.assertEqual(by_claim["C-003"]["ref_retrieval_method"], "not_found")

        # Path 4 — SUPPORTED.
        self.assertEqual(by_claim["C-004"]["judgment"], "SUPPORTED")
        self.assertIsNone(by_claim["C-004"]["defect_stage"])

        # Path 5 — AMBIGUOUS + citation_anchor.
        self.assertEqual(by_claim["C-005"]["judgment"], "AMBIGUOUS")
        self.assertEqual(by_claim["C-005"]["defect_stage"], "citation_anchor")

        # Path 6 — UNSUPPORTED + synthesis_overclaim.
        self.assertEqual(by_claim["C-006"]["judgment"], "UNSUPPORTED")
        self.assertEqual(by_claim["C-006"]["defect_stage"], "synthesis_overclaim")

        # Aggregate guarantees: every emitted entry stays inside the audit
        # path; nothing leaks to constraint_violations or sampling
        # summaries when no manifest constraints exist.
        self.assertEqual(
            out["constraint_violations"],
            [],
            "no manifest → no MNC/NC scope → no constraint_violations rows",
        )
        self.assertEqual(
            out["audit_sampling_summaries"],
            [],
            "N=6 ≤ cap=100 → no sampling summary",
        )


# ---------------------------------------------------------------------------
# T-M3 — Constraint inheritance: MNC applies even when not redeclared at claim level.
# ---------------------------------------------------------------------------


class TM3ConstraintInheritance(_ManifestTestBase):
    """T-M3: judge_fn receives MNC + claim-level NC merged + sorted by constraint_id."""

    def test_mnc_inherited_alongside_claim_level_nc(self) -> None:
        manifest = _manifest(
            mncs=[
                {
                    "constraint_id": "MNC-1",
                    "rule": "No unqualified causal language.",
                },
            ],
            claims=[
                _claim(
                    claim_id="C-001",
                    claim_text="Treatment caused outcome.",
                    planned_refs=["smith2024"],
                    ncs=[
                        {
                            "constraint_id": "NC-C001-1",
                            "rule": "No claims about secondary endpoints.",
                        }
                    ],
                ),
            ],
        )

        captured: list[dict[str, Any]] = []

        def judge_fn(**kwargs: Any) -> dict[str, Any]:
            captured.append(kwargs)
            return {
                "judgment": "SUPPORTED",
                "rationale": "Source supports the assertion.",
            }

        self.run_pipeline(
            manifests=[manifest],
            citations=[
                _citation(
                    claim_id="C-001",
                    claim_text="Treatment caused outcome.",
                    ref_slug="smith2024",
                )
            ],
            judge_fn=judge_fn,
        )

        self.assertEqual(
            len(captured),
            1,
            "judge MUST be invoked exactly once for the single cited claim",
        )
        ac = captured[0]["active_constraints"]

        ids = [c["constraint_id"] for c in ac]
        self.assertEqual(
            ids,
            ["MNC-1", "NC-C001-1"],
            "active_constraints must be sorted by constraint_id (M-INV-3 + §4 step 3)",
        )

        # Scope tags carried through so downstream readers can distinguish
        # global from claim-scoped (the in-runtime tag the cache excludes).
        scopes_by_id = {c["constraint_id"]: c["scope"] for c in ac}
        self.assertEqual(scopes_by_id, {"MNC-1": "MNC", "NC-C001-1": "NC"})

    def test_claim_without_nc_still_inherits_mnc(self) -> None:
        """M-INV-3: claim-level NC can ADD; cannot DROP global MNC.

        A claim that declares NO claim-level constraints must still see the
        manifest's MNCs in its active_constraints set.
        """
        manifest = _manifest(
            mncs=[
                {
                    "constraint_id": "MNC-1",
                    "rule": "No unqualified causal language.",
                },
                {
                    "constraint_id": "MNC-2",
                    "rule": "No proprietary benchmarks.",
                },
            ],
            claims=[
                _claim(
                    claim_id="C-002",
                    claim_text="Effect observed across all subgroups.",
                    planned_refs=["smith2024"],
                    ncs=None,  # No claim-level NCs.
                ),
            ],
        )

        captured: list[dict[str, Any]] = []

        def judge_fn(**kwargs: Any) -> dict[str, Any]:
            captured.append(kwargs)
            return {
                "judgment": "SUPPORTED",
                "rationale": "Source supports the assertion.",
            }

        self.run_pipeline(
            manifests=[manifest],
            citations=[
                _citation(
                    claim_id="C-002",
                    claim_text="Effect observed across all subgroups.",
                    ref_slug="smith2024",
                )
            ],
            judge_fn=judge_fn,
        )

        ac = captured[0]["active_constraints"]
        ids = [c["constraint_id"] for c in ac]
        self.assertEqual(
            ids,
            ["MNC-1", "MNC-2"],
            "MNCs MUST inherit even when claim declares no NCs (M-INV-3)",
        )


if __name__ == "__main__":  # pragma: no cover
    unittest.main()