Upload folder using huggingface_hub

3e363e2 verified 13 days ago

17 kB

"""Deterministic golden and adversarial case corpora (PLAN sections 54 and 55).
Golden cases are hand-specified graphs with a known correct answer, plus a
programmatically expanded family built from the ontology so the corpus is large
enough to catch regressions rather than merely illustrating them.  Every case is
deterministic, so a release either reproduces them exactly or it does not.
The adversarial family covers the section 55 list: wrong relation, wrong
direction, wrong type, wrong hop count, almost-valid path, missing edge,
contradictory edge, duplicate edge, distractor hub, high-degree irrelevant node,
cyclic graph, multiple valid answers, no valid answer, and competing paths.
"""

from __future__ import annotations

from collections.abc import Iterator, Sequence
from dataclasses import dataclass

from graph.graph import EdgeRecord, GraphStore
from graph.synthetic.ontology import RELATION_BY_NAME, RELATIONS
from graph.types import GraphQuery, ReasoningStatus


@dataclass(frozen=True)
class GoldenCase:
    """One deterministic reasoning case with its expected outcome."""

    case_id: str
    graph: GraphStore
    query: GraphQuery
    expected_answers: frozenset[str]
    expected_status: ReasoningStatus
    #: What competency this case pins down, used to group failures.
    category: str
    #: Paths that must NOT be returned as supporting evidence.
    forbidden_paths: tuple[tuple[tuple[str, str, str], ...], ...] = ()

    def __post_init__(self) -> None:
        for answer in self.expected_answers:
            if answer not in self.graph.nodes:
                raise ValueError(
                    f"case {self.case_id!r} expects unknown node {answer!r}"
                )


def _relation_attributes() -> dict[str, dict[str, str]]:
    return {
        relation.name: {"functional": str(relation.functional).lower()}
        for relation in RELATIONS
    }


def _store(
    nodes: dict[str, str], edges: Sequence[tuple[str, str, str]], version: str
) -> GraphStore:
    """Build a store from ``(source, relation, target)`` triples, typing from the ontology."""
    records = []
    for source, relation, target in edges:
        if relation not in RELATION_BY_NAME:
            raise KeyError(f"golden case uses unknown relation {relation!r}")
        records.append(
            EdgeRecord(
                source_id=source,
                relation_id=relation,
                target_id=target,
                source_type=nodes[source],
                target_type=nodes[target],
                relation_name=relation,
                attributes={},
            )
        )
    return GraphStore(
        nodes,
        records,
        graph_version=version,
        relation_attributes=_relation_attributes(),
    )


def _query(start: str, text: str, max_hops: int, **kwargs: object) -> GraphQuery:
    return GraphQuery(
        text=text,
        start_entities=(start,),
        start_entity_ids=(start,),
        max_hops=max_hops,
        return_paths=True,
        verify=True,
        **kwargs,  # type: ignore[arg-type]
    )


# -- the canonical PLAN section 54 case ------------------------------------------


def canonical_case() -> GoldenCase:
    """Alice -> Acme -> Project1 -> Supplier1, the case named in PLAN section 54."""
    nodes = {
        "alice": "Person",
        "acme": "Company",
        "project1": "Project",
        "supplier1": "Supplier",
    }
    graph = _store(
        nodes,
        (
            ("alice", "works_at", "acme"),
            ("acme", "owns", "project1"),
            ("project1", "uses", "supplier1"),
        ),
        "golden-canonical@1",
    )
    return GoldenCase(
        case_id="canonical_employer_project_supplier",
        graph=graph,
        query=_query(
            "alice",
            "Which supplier is used by a project of the company that employs alice?",
            3,
        ),
        expected_answers=frozenset({"supplier1"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="canonical",
    )


# -- relation-sequence discrimination (the PAT-specific golden family) -----------


def relation_sequence_cases() -> Iterator[GoldenCase]:
    """Same endpoint, different relation sequence: only one route may count.
    PLAN section 54 asks specifically for cases where the endpoint is shared but
    the relation sequence differs, because endpoint similarity alone must not be
    sufficient evidence.
    """
    nodes = {
        "p": "Person",
        "c1": "Company",
        "c2": "Company",
        "proj": "Project",
        "sup": "Supplier",
    }
    # Two routes reach ``proj``: via the employer (works_at, owns) and via a
    # partner of a different company (partnered_with, owns).  Only the first
    # answers "the company that employs p".
    graph = _store(
        nodes,
        (
            ("p", "works_at", "c1"),
            ("c1", "owns", "proj"),
            ("c1", "partnered_with", "c2"),
            ("c2", "owns", "proj"),
            ("proj", "uses", "sup"),
        ),
        "golden-relation-seq@1",
    )
    yield GoldenCase(
        case_id="shared_endpoint_distinct_sequence",
        graph=graph,
        query=_query("p", "What project does p's employer run?", 3),
        expected_answers=frozenset({"proj"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="relation_sequence",
        forbidden_paths=((("p", "works_at", "c1"), ("c1", "partnered_with", "c2")),),
    )

    # Direction matters: authored_by runs Document -> Person, so a Person cannot
    # reach a Document along it.
    direction_nodes = {"writer": "Person", "doc": "Document", "proj2": "Project"}
    direction_graph = _store(
        direction_nodes,
        (("doc", "authored_by", "writer"), ("proj2", "documented_by", "doc")),
        "golden-direction@1",
    )
    yield GoldenCase(
        case_id="wrong_direction_has_no_answer",
        graph=direction_graph,
        query=_query("writer", "Which document did writer author?", 2),
        expected_answers=frozenset(),
        expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
        category="direction",
    )


def type_discrimination_cases() -> Iterator[GoldenCase]:
    """An invalid intermediate type must not be masked by a valid endpoint type."""
    nodes = {
        "person": "Person",
        "company": "Company",
        "project": "Project",
        "doc": "Document",
        "supplier": "Supplier",
    }
    graph = _store(
        nodes,
        (
            ("person", "works_at", "company"),
            ("company", "owns", "project"),
            ("project", "uses", "supplier"),
            ("project", "documented_by", "doc"),
        ),
        "golden-types@1",
    )
    yield GoldenCase(
        case_id="required_intermediate_type_enforced",
        graph=graph,
        query=_query(
            "person",
            "Which supplier serves a project of person's employer?",
            3,
            required_types=("Supplier",),
        ),
        expected_answers=frozenset({"supplier"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="typed_path",
    )


def constraint_cases() -> Iterator[GoldenCase]:
    """Hop-count and forbidden-relation constraints must bind (PLAN section 17)."""
    nodes = {
        "a": "Person",
        "b": "Company",
        "c": "Project",
        "d": "Supplier",
        "loc": "Location",
    }
    graph = _store(
        nodes,
        (
            ("a", "works_at", "b"),
            ("b", "owns", "c"),
            ("c", "uses", "d"),
            ("b", "located_in", "loc"),
        ),
        "golden-constraints@1",
    )
    yield GoldenCase(
        case_id="exact_hops_two_excludes_three_hop_answer",
        graph=graph,
        query=_query(
            "a", "Which project two hops from a?", 3, exact_hops=2, min_hops=2
        ),
        expected_answers=frozenset({"c"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="constraints",
    )
    yield GoldenCase(
        case_id="forbidden_relation_blocks_route",
        graph=graph,
        query=_query(
            "a",
            "Which entity is reachable from a without using owns?",
            3,
            forbidden_relations=("owns",),
        ),
        expected_answers=frozenset({"b"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="constraints",
    )


def evidence_cases() -> Iterator[GoldenCase]:
    """Missing evidence and contradiction must be distinguished from falsehood."""
    missing_nodes = {"x": "Person", "y": "Company", "z": "Project"}
    missing = _store(missing_nodes, (("x", "works_at", "y"),), "golden-missing@1")
    yield GoldenCase(
        case_id="missing_final_edge_is_insufficient_evidence",
        graph=missing,
        query=_query("x", "What project does x's employer run?", 3),
        expected_answers=frozenset(),
        expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
        category="missing_evidence",
    )

    # works_at is functional, so two employers is a genuine contradiction.
    conflict_nodes = {"w": "Person", "e1": "Company", "e2": "Company"}
    conflict = _store(
        conflict_nodes,
        (("w", "works_at", "e1"), ("w", "works_at", "e2")),
        "golden-contradiction@1",
    )
    yield GoldenCase(
        case_id="functional_relation_with_two_values_is_contradiction",
        graph=conflict,
        query=_query("w", "Which company employs w?", 1),
        expected_answers=frozenset({"e1", "e2"}),
        expected_status=ReasoningStatus.CONTRADICTED,
        category="contradiction",
    )

    # owns is set-valued, so two projects is simply two facts.
    multi_nodes = {"m": "Company", "p1": "Project", "p2": "Project"}
    multi = _store(
        multi_nodes, (("m", "owns", "p1"), ("m", "owns", "p2")), "golden-setvalued@1"
    )
    yield GoldenCase(
        case_id="set_valued_relation_with_two_values_is_not_contradiction",
        graph=multi,
        query=_query("m", "Which projects does m own?", 1),
        expected_answers=frozenset({"p1", "p2"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="contradiction",
    )


def adversarial_cases(count: int = 120) -> Iterator[GoldenCase]:
    """Generate the PLAN section 55 adversarial families deterministically.
    Each family is emitted at several sizes so a failure is attributable to the
    family rather than to one particular graph.
    """
    families = (
        _distractor_hub_case,
        _duplicate_edge_case,
        _almost_valid_path_case,
        _cycle_case,
        _no_valid_answer_case,
        _multiple_valid_answers_case,
        _wrong_hop_count_case,
    )
    emitted = 0
    scale = 2
    while emitted < count:
        for family in families:
            if emitted >= count:
                return
            yield family(scale)
            emitted += 1
        scale += 1


def _distractor_hub_case(scale: int) -> GoldenCase:
    """A high-degree irrelevant hub must not attract the answer."""
    nodes = {"src": "Person", "emp": "Company", "goal": "Project"}
    edges = [("src", "works_at", "emp"), ("emp", "owns", "goal")]
    for index in range(scale * 3):
        hub = f"hub{index}"
        nodes[hub] = "Team"
        edges.append(("src", "member_of", hub))
        for inner in range(scale):
            decoy = f"decoy{index}_{inner}"
            nodes[decoy] = "Project"
            edges.append((hub, "assigned_to", decoy))
    return GoldenCase(
        case_id=f"distractor_hub_scale{scale}",
        graph=_store(nodes, edges, f"adv-hub@{scale}"),
        query=_query("src", "What project does src's employer run?", 3),
        expected_answers=frozenset({"goal"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="distractor_hub",
    )


def _duplicate_edge_case(scale: int) -> GoldenCase:
    """A duplicated edge must not multiply the evidence or change the answer."""
    nodes = {"p": "Person", "c": "Company", "j": "Project"}
    edges = [("p", "works_at", "c")] + [("c", "owns", "j")] * max(2, scale)
    return GoldenCase(
        case_id=f"duplicate_edge_scale{scale}",
        graph=_store(nodes, edges, f"adv-dup@{scale}"),
        query=_query("p", "What project does p's employer run?", 3),
        expected_answers=frozenset({"j"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="duplicate_edge",
    )


def _almost_valid_path_case(scale: int) -> GoldenCase:
    """A route that is one relation wrong must not be accepted."""
    nodes = {"p": "Person", "t": "Team", "c": "Company", "j": "Project"}
    edges: list[tuple[str, str, str]] = [
        ("p", "member_of", "t"),
        ("t", "part_of", "c"),
        ("c", "owns", "j"),
    ]
    for index in range(scale):
        extra = f"other{index}"
        nodes[extra] = "Project"
        edges.append(("c", "owns", extra))
    return GoldenCase(
        case_id=f"almost_valid_path_scale{scale}",
        graph=_store(nodes, edges, f"adv-almost@{scale}"),
        # p has no works_at edge at all, so the employer intent is unanswerable
        # even though a structurally similar team route exists.
        query=_query("p", "Which company employs p?", 1),
        expected_answers=frozenset(),
        expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
        category="almost_valid",
    )


def _cycle_case(scale: int) -> GoldenCase:
    """Cycles must not cause repeated-node evidence or non-termination."""
    nodes = {f"proj{index}": "Project" for index in range(max(2, scale + 1))}
    nodes["p"] = "Person"
    nodes["c"] = "Company"
    keys = [key for key in nodes if key.startswith("proj")]
    edges = [("p", "works_at", "c"), ("c", "owns", keys[0])]
    for index, key in enumerate(keys):
        edges.append((key, "depends_on", keys[(index + 1) % len(keys)]))
    return GoldenCase(
        case_id=f"cyclic_dependency_scale{scale}",
        graph=_store(nodes, edges, f"adv-cycle@{scale}"),
        query=_query("p", "What project does p's employer run?", 4),
        expected_answers=frozenset({keys[0]}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="cyclic",
    )


def _no_valid_answer_case(scale: int) -> GoldenCase:
    """A graph with no answering path must yield no answer."""
    nodes = {"p": "Person"}
    edges: list[tuple[str, str, str]] = []
    for index in range(max(1, scale)):
        account = f"acct{index}"
        nodes[account] = "Account"
        edges.append(("p", "holds", account))
    return GoldenCase(
        case_id=f"no_valid_answer_scale{scale}",
        graph=_store(nodes, edges, f"adv-none@{scale}"),
        query=_query("p", "Which company employs p?", 3),
        expected_answers=frozenset(),
        expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
        category="no_answer",
    )


def _multiple_valid_answers_case(scale: int) -> GoldenCase:
    """All genuinely valid answers must be returned, not just the first."""
    nodes = {"p": "Person", "c": "Company"}
    edges = [("p", "works_at", "c")]
    expected = set()
    for index in range(max(2, scale)):
        project = f"proj{index}"
        nodes[project] = "Project"
        edges.append(("c", "owns", project))
        expected.add(project)
    return GoldenCase(
        case_id=f"multiple_valid_answers_scale{scale}",
        graph=_store(nodes, edges, f"adv-multi@{scale}"),
        query=_query("p", "What project does p's employer run?", 2),
        expected_answers=frozenset(expected),
        expected_status=ReasoningStatus.SUPPORTED,
        category="multiple_answers",
    )


def _wrong_hop_count_case(scale: int) -> GoldenCase:
    """An answer at the wrong depth must be excluded by an exact-hop constraint."""
    nodes = {"p": "Person", "c": "Company", "j": "Project", "s": "Supplier"}
    edges = [("p", "works_at", "c"), ("c", "owns", "j"), ("j", "uses", "s")]
    return GoldenCase(
        case_id=f"wrong_hop_count_scale{scale}",
        graph=_store(nodes, edges, f"adv-hops@{scale}"),
        query=_query(
            "p",
            "Which entity is exactly three hops from p?",
            3,
            exact_hops=3,
            min_hops=3,
        ),
        expected_answers=frozenset({"s"}),
        expected_status=ReasoningStatus.SUPPORTED,
        category="hop_count",
    )


def golden_corpus() -> tuple[GoldenCase, ...]:
    """The deterministic suite every release must pass."""
    return (
        canonical_case(),
        *relation_sequence_cases(),
        *type_discrimination_cases(),
        *constraint_cases(),
        *evidence_cases(),
    )


def full_corpus(adversarial: int = 120) -> tuple[GoldenCase, ...]:
    """Golden plus adversarial cases."""
    return (*golden_corpus(), *adversarial_cases(adversarial))


def expected_path_keys(case: GoldenCase) -> frozenset[tuple[tuple[str, str, str], ...]]:
    """Path keys forbidden as supporting evidence for ``case``."""
    return frozenset(case.forbidden_paths)

	"""Deterministic golden and adversarial case corpora (PLAN sections 54 and 55).

	Golden cases are hand-specified graphs with a known correct answer, plus a
	programmatically expanded family built from the ontology so the corpus is large
	enough to catch regressions rather than merely illustrating them. Every case is
	deterministic, so a release either reproduces them exactly or it does not.

	The adversarial family covers the section 55 list: wrong relation, wrong
	direction, wrong type, wrong hop count, almost-valid path, missing edge,
	contradictory edge, duplicate edge, distractor hub, high-degree irrelevant node,
	cyclic graph, multiple valid answers, no valid answer, and competing paths.
	"""

	from __future__ import annotations

	from collections.abc import Iterator, Sequence
	from dataclasses import dataclass

	from graph.graph import EdgeRecord, GraphStore
	from graph.synthetic.ontology import RELATION_BY_NAME, RELATIONS
	from graph.types import GraphQuery, ReasoningStatus


	@dataclass(frozen=True)
	class GoldenCase:
	"""One deterministic reasoning case with its expected outcome."""

	case_id: str
	graph: GraphStore
	query: GraphQuery
	expected_answers: frozenset[str]
	expected_status: ReasoningStatus
	#: What competency this case pins down, used to group failures.
	category: str
	#: Paths that must NOT be returned as supporting evidence.
	forbidden_paths: tuple[tuple[tuple[str, str, str], ...], ...] = ()

	def __post_init__(self) -> None:
	for answer in self.expected_answers:
	if answer not in self.graph.nodes:
	raise ValueError(
	f"case {self.case_id!r} expects unknown node {answer!r}"
	)


	def _relation_attributes() -> dict[str, dict[str, str]]:
	return {
	relation.name: {"functional": str(relation.functional).lower()}
	for relation in RELATIONS
	}


	def _store(
	nodes: dict[str, str], edges: Sequence[tuple[str, str, str]], version: str
	) -> GraphStore:
	"""Build a store from ``(source, relation, target)`` triples, typing from the ontology."""
	records = []
	for source, relation, target in edges:
	if relation not in RELATION_BY_NAME:
	raise KeyError(f"golden case uses unknown relation {relation!r}")
	records.append(
	EdgeRecord(
	source_id=source,
	relation_id=relation,
	target_id=target,
	source_type=nodes[source],
	target_type=nodes[target],
	relation_name=relation,
	attributes={},
	)
	)
	return GraphStore(
	nodes,
	records,
	graph_version=version,
	relation_attributes=_relation_attributes(),
	)


	def _query(start: str, text: str, max_hops: int, **kwargs: object) -> GraphQuery:
	return GraphQuery(
	text=text,
	start_entities=(start,),
	start_entity_ids=(start,),
	max_hops=max_hops,
	return_paths=True,
	verify=True,
	**kwargs, # type: ignore[arg-type]
	)


	# -- the canonical PLAN section 54 case ------------------------------------------


	def canonical_case() -> GoldenCase:
	"""Alice -> Acme -> Project1 -> Supplier1, the case named in PLAN section 54."""
	nodes = {
	"alice": "Person",
	"acme": "Company",
	"project1": "Project",
	"supplier1": "Supplier",
	}
	graph = _store(
	nodes,
	(
	("alice", "works_at", "acme"),
	("acme", "owns", "project1"),
	("project1", "uses", "supplier1"),
	),
	"golden-canonical@1",
	)
	return GoldenCase(
	case_id="canonical_employer_project_supplier",
	graph=graph,
	query=_query(
	"alice",
	"Which supplier is used by a project of the company that employs alice?",
	3,
	),
	expected_answers=frozenset({"supplier1"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="canonical",
	)


	# -- relation-sequence discrimination (the PAT-specific golden family) -----------


	def relation_sequence_cases() -> Iterator[GoldenCase]:
	"""Same endpoint, different relation sequence: only one route may count.

	PLAN section 54 asks specifically for cases where the endpoint is shared but
	the relation sequence differs, because endpoint similarity alone must not be
	sufficient evidence.
	"""
	nodes = {
	"p": "Person",
	"c1": "Company",
	"c2": "Company",
	"proj": "Project",
	"sup": "Supplier",
	}
	# Two routes reach ``proj``: via the employer (works_at, owns) and via a
	# partner of a different company (partnered_with, owns). Only the first
	# answers "the company that employs p".
	graph = _store(
	nodes,
	(
	("p", "works_at", "c1"),
	("c1", "owns", "proj"),
	("c1", "partnered_with", "c2"),
	("c2", "owns", "proj"),
	("proj", "uses", "sup"),
	),
	"golden-relation-seq@1",
	)
	yield GoldenCase(
	case_id="shared_endpoint_distinct_sequence",
	graph=graph,
	query=_query("p", "What project does p's employer run?", 3),
	expected_answers=frozenset({"proj"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="relation_sequence",
	forbidden_paths=((("p", "works_at", "c1"), ("c1", "partnered_with", "c2")),),
	)

	# Direction matters: authored_by runs Document -> Person, so a Person cannot
	# reach a Document along it.
	direction_nodes = {"writer": "Person", "doc": "Document", "proj2": "Project"}
	direction_graph = _store(
	direction_nodes,
	(("doc", "authored_by", "writer"), ("proj2", "documented_by", "doc")),
	"golden-direction@1",
	)
	yield GoldenCase(
	case_id="wrong_direction_has_no_answer",
	graph=direction_graph,
	query=_query("writer", "Which document did writer author?", 2),
	expected_answers=frozenset(),
	expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
	category="direction",
	)


	def type_discrimination_cases() -> Iterator[GoldenCase]:
	"""An invalid intermediate type must not be masked by a valid endpoint type."""
	nodes = {
	"person": "Person",
	"company": "Company",
	"project": "Project",
	"doc": "Document",
	"supplier": "Supplier",
	}
	graph = _store(
	nodes,
	(
	("person", "works_at", "company"),
	("company", "owns", "project"),
	("project", "uses", "supplier"),
	("project", "documented_by", "doc"),
	),
	"golden-types@1",
	)
	yield GoldenCase(
	case_id="required_intermediate_type_enforced",
	graph=graph,
	query=_query(
	"person",
	"Which supplier serves a project of person's employer?",
	3,
	required_types=("Supplier",),
	),
	expected_answers=frozenset({"supplier"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="typed_path",
	)


	def constraint_cases() -> Iterator[GoldenCase]:
	"""Hop-count and forbidden-relation constraints must bind (PLAN section 17)."""
	nodes = {
	"a": "Person",
	"b": "Company",
	"c": "Project",
	"d": "Supplier",
	"loc": "Location",
	}
	graph = _store(
	nodes,
	(
	("a", "works_at", "b"),
	("b", "owns", "c"),
	("c", "uses", "d"),
	("b", "located_in", "loc"),
	),
	"golden-constraints@1",
	)
	yield GoldenCase(
	case_id="exact_hops_two_excludes_three_hop_answer",
	graph=graph,
	query=_query(
	"a", "Which project two hops from a?", 3, exact_hops=2, min_hops=2
	),
	expected_answers=frozenset({"c"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="constraints",
	)
	yield GoldenCase(
	case_id="forbidden_relation_blocks_route",
	graph=graph,
	query=_query(
	"a",
	"Which entity is reachable from a without using owns?",
	3,
	forbidden_relations=("owns",),
	),
	expected_answers=frozenset({"b"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="constraints",
	)


	def evidence_cases() -> Iterator[GoldenCase]:
	"""Missing evidence and contradiction must be distinguished from falsehood."""
	missing_nodes = {"x": "Person", "y": "Company", "z": "Project"}
	missing = _store(missing_nodes, (("x", "works_at", "y"),), "golden-missing@1")
	yield GoldenCase(
	case_id="missing_final_edge_is_insufficient_evidence",
	graph=missing,
	query=_query("x", "What project does x's employer run?", 3),
	expected_answers=frozenset(),
	expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
	category="missing_evidence",
	)

	# works_at is functional, so two employers is a genuine contradiction.
	conflict_nodes = {"w": "Person", "e1": "Company", "e2": "Company"}
	conflict = _store(
	conflict_nodes,
	(("w", "works_at", "e1"), ("w", "works_at", "e2")),
	"golden-contradiction@1",
	)
	yield GoldenCase(
	case_id="functional_relation_with_two_values_is_contradiction",
	graph=conflict,
	query=_query("w", "Which company employs w?", 1),
	expected_answers=frozenset({"e1", "e2"}),
	expected_status=ReasoningStatus.CONTRADICTED,
	category="contradiction",
	)

	# owns is set-valued, so two projects is simply two facts.
	multi_nodes = {"m": "Company", "p1": "Project", "p2": "Project"}
	multi = _store(
	multi_nodes, (("m", "owns", "p1"), ("m", "owns", "p2")), "golden-setvalued@1"
	)
	yield GoldenCase(
	case_id="set_valued_relation_with_two_values_is_not_contradiction",
	graph=multi,
	query=_query("m", "Which projects does m own?", 1),
	expected_answers=frozenset({"p1", "p2"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="contradiction",
	)


	def adversarial_cases(count: int = 120) -> Iterator[GoldenCase]:
	"""Generate the PLAN section 55 adversarial families deterministically.

	Each family is emitted at several sizes so a failure is attributable to the
	family rather than to one particular graph.
	"""
	families = (
	_distractor_hub_case,
	_duplicate_edge_case,
	_almost_valid_path_case,
	_cycle_case,
	_no_valid_answer_case,
	_multiple_valid_answers_case,
	_wrong_hop_count_case,
	)
	emitted = 0
	scale = 2
	while emitted < count:
	for family in families:
	if emitted >= count:
	return
	yield family(scale)
	emitted += 1
	scale += 1


	def _distractor_hub_case(scale: int) -> GoldenCase:
	"""A high-degree irrelevant hub must not attract the answer."""
	nodes = {"src": "Person", "emp": "Company", "goal": "Project"}
	edges = [("src", "works_at", "emp"), ("emp", "owns", "goal")]
	for index in range(scale * 3):
	hub = f"hub{index}"
	nodes[hub] = "Team"
	edges.append(("src", "member_of", hub))
	for inner in range(scale):
	decoy = f"decoy{index}_{inner}"
	nodes[decoy] = "Project"
	edges.append((hub, "assigned_to", decoy))
	return GoldenCase(
	case_id=f"distractor_hub_scale{scale}",
	graph=_store(nodes, edges, f"adv-hub@{scale}"),
	query=_query("src", "What project does src's employer run?", 3),
	expected_answers=frozenset({"goal"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="distractor_hub",
	)


	def _duplicate_edge_case(scale: int) -> GoldenCase:
	"""A duplicated edge must not multiply the evidence or change the answer."""
	nodes = {"p": "Person", "c": "Company", "j": "Project"}
	edges = [("p", "works_at", "c")] + [("c", "owns", "j")] * max(2, scale)
	return GoldenCase(
	case_id=f"duplicate_edge_scale{scale}",
	graph=_store(nodes, edges, f"adv-dup@{scale}"),
	query=_query("p", "What project does p's employer run?", 3),
	expected_answers=frozenset({"j"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="duplicate_edge",
	)


	def _almost_valid_path_case(scale: int) -> GoldenCase:
	"""A route that is one relation wrong must not be accepted."""
	nodes = {"p": "Person", "t": "Team", "c": "Company", "j": "Project"}
	edges: list[tuple[str, str, str]] = [
	("p", "member_of", "t"),
	("t", "part_of", "c"),
	("c", "owns", "j"),
	]
	for index in range(scale):
	extra = f"other{index}"
	nodes[extra] = "Project"
	edges.append(("c", "owns", extra))
	return GoldenCase(
	case_id=f"almost_valid_path_scale{scale}",
	graph=_store(nodes, edges, f"adv-almost@{scale}"),
	# p has no works_at edge at all, so the employer intent is unanswerable
	# even though a structurally similar team route exists.
	query=_query("p", "Which company employs p?", 1),
	expected_answers=frozenset(),
	expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
	category="almost_valid",
	)


	def _cycle_case(scale: int) -> GoldenCase:
	"""Cycles must not cause repeated-node evidence or non-termination."""
	nodes = {f"proj{index}": "Project" for index in range(max(2, scale + 1))}
	nodes["p"] = "Person"
	nodes["c"] = "Company"
	keys = [key for key in nodes if key.startswith("proj")]
	edges = [("p", "works_at", "c"), ("c", "owns", keys[0])]
	for index, key in enumerate(keys):
	edges.append((key, "depends_on", keys[(index + 1) % len(keys)]))
	return GoldenCase(
	case_id=f"cyclic_dependency_scale{scale}",
	graph=_store(nodes, edges, f"adv-cycle@{scale}"),
	query=_query("p", "What project does p's employer run?", 4),
	expected_answers=frozenset({keys[0]}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="cyclic",
	)


	def _no_valid_answer_case(scale: int) -> GoldenCase:
	"""A graph with no answering path must yield no answer."""
	nodes = {"p": "Person"}
	edges: list[tuple[str, str, str]] = []
	for index in range(max(1, scale)):
	account = f"acct{index}"
	nodes[account] = "Account"
	edges.append(("p", "holds", account))
	return GoldenCase(
	case_id=f"no_valid_answer_scale{scale}",
	graph=_store(nodes, edges, f"adv-none@{scale}"),
	query=_query("p", "Which company employs p?", 3),
	expected_answers=frozenset(),
	expected_status=ReasoningStatus.INSUFFICIENT_EVIDENCE,
	category="no_answer",
	)


	def _multiple_valid_answers_case(scale: int) -> GoldenCase:
	"""All genuinely valid answers must be returned, not just the first."""
	nodes = {"p": "Person", "c": "Company"}
	edges = [("p", "works_at", "c")]
	expected = set()
	for index in range(max(2, scale)):
	project = f"proj{index}"
	nodes[project] = "Project"
	edges.append(("c", "owns", project))
	expected.add(project)
	return GoldenCase(
	case_id=f"multiple_valid_answers_scale{scale}",
	graph=_store(nodes, edges, f"adv-multi@{scale}"),
	query=_query("p", "What project does p's employer run?", 2),
	expected_answers=frozenset(expected),
	expected_status=ReasoningStatus.SUPPORTED,
	category="multiple_answers",
	)


	def _wrong_hop_count_case(scale: int) -> GoldenCase:
	"""An answer at the wrong depth must be excluded by an exact-hop constraint."""
	nodes = {"p": "Person", "c": "Company", "j": "Project", "s": "Supplier"}
	edges = [("p", "works_at", "c"), ("c", "owns", "j"), ("j", "uses", "s")]
	return GoldenCase(
	case_id=f"wrong_hop_count_scale{scale}",
	graph=_store(nodes, edges, f"adv-hops@{scale}"),
	query=_query(
	"p",
	"Which entity is exactly three hops from p?",
	3,
	exact_hops=3,
	min_hops=3,
	),
	expected_answers=frozenset({"s"}),
	expected_status=ReasoningStatus.SUPPORTED,
	category="hop_count",
	)


	def golden_corpus() -> tuple[GoldenCase, ...]:
	"""The deterministic suite every release must pass."""
	return (
	canonical_case(),
	*relation_sequence_cases(),
	*type_discrimination_cases(),
	*constraint_cases(),
	*evidence_cases(),
	)


	def full_corpus(adversarial: int = 120) -> tuple[GoldenCase, ...]:
	"""Golden plus adversarial cases."""
	return (golden_corpus(), adversarial_cases(adversarial))


	def expected_path_keys(case: GoldenCase) -> frozenset[tuple[tuple[str, str, str], ...]]:
	"""Path keys forbidden as supporting evidence for ``case``."""
	return frozenset(case.forbidden_paths)