Upload folder using huggingface_hub

3e363e2 verified 19 days ago

15.8 kB

"""Intent-driven task construction: the query states *what*, never *how*.
Why this module exists
----------------------
The previous generator wrote the gold relation sequence into
``GraphQuery.required_relation_sequence`` and into the question text, then
computed the gold answer as "every path matching that sequence".  The reasoner
filtered its own candidate expansion by the same field.  The task was therefore
"walk the sequence you were given", which exhaustive traversal solves perfectly
and which no amount of training can improve.  Every metric pinned at 1.0.
Here a task is built from a :class:`~graph.synthetic.ontology.QueryIntent`:
* the query carries a natural-language paraphrase and the start entity;
* the relation composition that answers it is **never** placed in the query;
* ``max_hops`` is a stage-level bound shared by every intent at that stage, so
  path length is not a giveaway either;
* gold answers are the endpoints of composition-matching paths, while endpoints
  of *other* same-length paths from the same start become hard negatives.
A search that does not understand the query therefore cannot choose among the
competing endpoints, which is exactly the discrimination PLAN sections 12.5 and
16 ask the model to learn.
"""

from __future__ import annotations

import random
from collections.abc import Mapping, Sequence
from dataclasses import dataclass

from graph.graph import EdgeRecord, GraphStore
from graph.synthetic.ontology import (
    RELATION_BY_NAME,
    QueryIntent,
)
from graph.types import GraphPath, GraphQuery, PathStep, ReasoningStatus

#: Stage-level hop budgets.  Every intent at a stage shares the same bound so
#: ``max_hops`` never reveals the composition length.
STAGE_HOP_BUDGET: Mapping[int, int] = {
    0: 2,
    1: 2,
    2: 3,
    3: 4,
    4: 4,
    5: 5,
    6: 5,
    7: 5,
    8: 5,
    9: 5,
    10: 5,
    11: 5,
    12: 5,
    13: 6,
    14: 6,
}


@dataclass(frozen=True)
class TaskBuild:
    """The outcome of constructing one task over a (possibly mutated) graph."""

    graph: GraphStore
    query: GraphQuery
    answers: tuple[str, ...]
    negative_answers: tuple[str, ...]
    supporting_paths: tuple[GraphPath, ...]
    competing_paths: tuple[GraphPath, ...]
    conflicting_paths: tuple[GraphPath, ...]
    status: ReasoningStatus
    intent_id: str
    composition: tuple[str, ...]
    start_id: str


def paths_matching_composition(
    graph: GraphStore, start_id: str, composition: Sequence[str]
) -> tuple[GraphPath, ...]:
    """Enumerate simple paths from ``start_id`` that walk ``composition`` exactly."""
    results: list[GraphPath] = []

    def walk(
        node: str, index: int, steps: tuple[PathStep, ...], visited: frozenset[str]
    ) -> None:
        if index == len(composition):
            results.append(GraphPath(steps=steps))
            return
        for edge in graph.outgoing(node):
            if edge.relation_id != composition[index] or edge.target_id in visited:
                continue
            walk(
                edge.target_id,
                index + 1,
                (
                    *steps,
                    PathStep(
                        source_id=edge.source_id,
                        relation_id=edge.relation_id,
                        target_id=edge.target_id,
                        source_type=edge.source_type,
                        target_type=edge.target_type,
                        edge_attributes=edge.attributes,
                    ),
                ),
                visited | {edge.target_id},
            )

    walk(start_id, 0, (), frozenset({start_id}))
    return tuple(results)


def competing_paths(
    graph: GraphStore,
    start_id: str,
    length: int,
    composition: Sequence[str],
    *,
    limit: int = 64,
) -> tuple[GraphPath, ...]:
    """Same-length paths from the same start that do **not** walk ``composition``.
    These are the hard negatives: identical hop count, identical origin, and
    often an endpoint of the same entity type.  Only the relation sequence
    differs, so discriminating them requires reading the query.
    """
    found: list[GraphPath] = []
    for path in graph.iter_paths((start_id,), length):
        if path.hop_count != length:
            continue
        if tuple(path.relation_sequence) == tuple(composition):
            continue
        found.append(path)
        if len(found) >= limit:
            break
    return tuple(found)


def plant_composition(
    graph: GraphStore,
    start_id: str,
    composition: Sequence[str],
    type_lookup: Mapping[str, str],
    relation_lookup: Mapping[str, str],
    rng: random.Random,
) -> GraphStore:
    """Add the minimum edges needed for ``composition`` to be walkable from ``start_id``.
    Planting keeps every generated task answerable without weakening it: the
    planted path is one of many same-length routes out of ``start_id``, so the
    model still has to pick the right one.
    """
    nodes_by_type: dict[str, list[str]] = {}
    for node_id, node_type in graph.nodes.items():
        nodes_by_type.setdefault(node_type, []).append(node_id)

    edges = list(graph.edges)
    current = start_id
    visited = {start_id}
    for canonical in composition:
        spec = RELATION_BY_NAME[canonical]
        # Use the builder's own canonical -> graph-local mapping.  Inferring it
        # from edge type signatures is ambiguous whenever two relations share a
        # domain/range pair (``member_of`` and ``manages`` are both
        # Person -> Team), and on a renamed graph that ambiguity silently planted
        # edges under the canonical name that no query could then match.
        relation_id = relation_lookup[canonical]
        target_type = type_lookup[spec.range]
        existing = [
            edge.target_id
            for edge in graph.outgoing(current)
            if edge.relation_id == relation_id and edge.target_id not in visited
        ]
        if existing:
            current = existing[0]
            visited.add(current)
            continue
        candidates = [
            node for node in nodes_by_type.get(target_type, ()) if node not in visited
        ]
        if not candidates:
            raise RuntimeError(
                f"cannot plant {canonical!r}: no unused node of type {target_type!r} exists"
            )
        target = candidates[rng.randrange(len(candidates))]
        edges.append(
            EdgeRecord(
                source_id=current,
                relation_id=relation_id,
                target_id=target,
                source_type=graph.nodes[current],
                target_type=graph.nodes[target],
                relation_name=relation_id,
                attributes={"weight": "1"},
            )
        )
        current = target
        visited.add(current)
    return GraphStore(
        graph.nodes,
        edges,
        graph_version=graph.graph_version,
        node_attributes=graph.node_attributes,
        relation_attributes=graph.relation_attributes,
    )


def candidate_starts(
    graph: GraphStore, intent: QueryIntent, type_lookup: Mapping[str, str]
) -> tuple[str, ...]:
    """Nodes typed as the composition's domain, ordered for determinism."""
    domain = type_lookup[RELATION_BY_NAME[intent.composition[0]].domain]
    return tuple(node for node, node_type in graph.nodes.items() if node_type == domain)


def build_task(
    graph: GraphStore,
    intent: QueryIntent,
    *,
    stage: int,
    rng: random.Random,
    type_lookup: Mapping[str, str],
    relation_lookup: Mapping[str, str],
    answer_mode: str = "entity",
    held_out_templates: bool = False,
    variant: str = "standard",
) -> TaskBuild:
    """Construct one well-posed task for ``intent`` over ``graph``.
    ``variant`` selects the reasoning scenario: ``standard`` answers the intent,
    ``missing_edge`` removes the supporting evidence so the correct response is
    ``INSUFFICIENT_EVIDENCE``, and ``contradiction`` introduces a second
    conflicting value for the same subject/relation claim.
    """
    composition = tuple(relation_lookup[name] for name in intent.composition)
    starts = candidate_starts(graph, intent, type_lookup)
    if not starts:
        raise RuntimeError(
            f"graph has no node of the domain type required by intent {intent.intent_id!r}"
        )

    start_id = _select_start(graph, starts, composition, rng)
    supporting = paths_matching_composition(graph, start_id, composition)
    if not supporting:
        graph = plant_composition(
            graph, start_id, intent.composition, type_lookup, relation_lookup, rng
        )
        supporting = paths_matching_composition(graph, start_id, composition)
        if not supporting:
            raise RuntimeError(
                f"planting failed for intent {intent.intent_id!r} from {start_id!r}"
            )

    conflicts: tuple[GraphPath, ...] = ()
    status = ReasoningStatus.SUPPORTED

    if variant == "missing_edge":
        graph = _remove_final_edges(graph, supporting)
        supporting = paths_matching_composition(graph, start_id, composition)
        if not supporting:
            status = ReasoningStatus.INSUFFICIENT_EVIDENCE
    elif variant == "contradiction":
        graph, conflicts = _introduce_contradiction(graph, supporting, rng)
        supporting = paths_matching_composition(graph, start_id, composition)
        if conflicts:
            status = ReasoningStatus.CONTRADICTED

    competing = competing_paths(graph, start_id, len(composition), composition)
    answers = (
        tuple(
            dict.fromkeys(path.endpoint_id for path in supporting if path.endpoint_id)
        )
        if status in {ReasoningStatus.SUPPORTED, ReasoningStatus.CONTRADICTED}
        else ()
    )
    negatives = tuple(
        dict.fromkeys(
            path.endpoint_id
            for path in competing
            if path.endpoint_id and path.endpoint_id not in set(answers)
        )
    )

    templates = intent.held_out_templates if held_out_templates else intent.templates
    if not templates:
        templates = intent.templates
    text = templates[rng.randrange(len(templates))].format(start=start_id)

    budget = STAGE_HOP_BUDGET.get(stage, 5)
    max_hops = max(budget, len(composition))
    query = GraphQuery(
        text=text,
        start_entities=(start_id,),
        start_entity_ids=(start_id,),
        # Deliberately empty: the composition, its length, and the answer type
        # are all withheld.  They are the answer, not the question.
        allowed_relations=(),
        required_relations=(),
        required_relation_sequence=(),
        required_types=(),
        min_hops=1,
        max_hops=max_hops,
        exact_hops=None,
        return_paths=True,
        verify=True,
        operation="find_entities" if answer_mode == "entity" else answer_mode,
        answer_mode=answer_mode,
        target_entity_ids=_target_ids(answer_mode, answers, negatives, status, rng),
    )

    if answer_mode == "boolean":
        status = _boolean_status(status, query.target_entity_ids, answers)

    return TaskBuild(
        graph=graph,
        query=query,
        answers=answers,
        negative_answers=negatives[:16],
        supporting_paths=supporting,
        competing_paths=competing[:16],
        conflicting_paths=conflicts,
        status=status,
        intent_id=intent.intent_id,
        composition=tuple(intent.composition),
        start_id=start_id,
    )


def _select_start(
    graph: GraphStore,
    starts: Sequence[str],
    composition: Sequence[str],
    rng: random.Random,
) -> str:
    """Prefer a start that already walks the composition and has competitors.
    A start with competing same-length routes makes the task discriminative; one
    without them would be answerable by hop-count alone.
    """
    walkable = [
        node for node in starts if paths_matching_composition(graph, node, composition)
    ]
    contested = [
        node
        for node in walkable
        if competing_paths(graph, node, len(composition), composition, limit=2)
    ]
    pool = contested or walkable or list(starts)
    return pool[rng.randrange(len(pool))]


def _remove_final_edges(
    graph: GraphStore, supporting: Sequence[GraphPath]
) -> GraphStore:
    """Delete the last edge of every supporting path, removing the evidence."""
    doomed = {
        (path.steps[-1].source_id, path.steps[-1].relation_id, path.steps[-1].target_id)
        for path in supporting
        if path.steps
    }
    edges = [
        edge
        for edge in graph.edges
        if (edge.source_id, edge.relation_id, edge.target_id) not in doomed
    ]
    return GraphStore(
        graph.nodes,
        edges,
        graph_version=graph.graph_version,
        node_attributes=graph.node_attributes,
        relation_attributes=graph.relation_attributes,
    )


def _introduce_contradiction(
    graph: GraphStore, supporting: Sequence[GraphPath], rng: random.Random
) -> tuple[GraphStore, tuple[GraphPath, ...]]:
    """Add a second, conflicting target for the first claim on a supporting path.
    A contradiction is a *single* subject/relation asserting two different
    values.  Two different multi-hop routes to different endpoints are competing
    explanations, not a contradiction, and are handled separately.
    """
    if not supporting or not supporting[0].steps:
        return graph, ()
    first = supporting[0].steps[0]
    same_type = [
        node
        for node, node_type in graph.nodes.items()
        if node_type == first.target_type
        and node not in {first.source_id, first.target_id}
    ]
    if not same_type:
        return graph, ()
    rival = same_type[rng.randrange(len(same_type))]
    if graph.has_edge(first.source_id, first.relation_id, rival):
        return graph, ()
    edges = [
        *graph.edges,
        EdgeRecord(
            source_id=first.source_id,
            relation_id=first.relation_id,
            target_id=rival,
            source_type=graph.nodes[first.source_id],
            target_type=graph.nodes[rival],
            relation_name=first.relation_id,
            attributes={"contradictory": "true"},
        ),
    ]
    mutated = GraphStore(
        graph.nodes,
        edges,
        graph_version=graph.graph_version,
        node_attributes=graph.node_attributes,
        relation_attributes=graph.relation_attributes,
    )
    claims = tuple(
        GraphPath(
            steps=(
                PathStep(
                    source_id=first.source_id,
                    relation_id=first.relation_id,
                    target_id=target,
                    source_type=graph.nodes[first.source_id],
                    target_type=graph.nodes[target],
                ),
            )
        )
        for target in (first.target_id, rival)
    )
    return mutated, claims


def _target_ids(
    answer_mode: str,
    answers: Sequence[str],
    negatives: Sequence[str],
    status: ReasoningStatus,
    rng: random.Random,
) -> tuple[str, ...]:
    """Boolean tasks need a concrete claim to judge; other modes need none."""
    if answer_mode != "boolean":
        return ()
    truthful = bool(answers) and rng.random() < 0.5
    if truthful:
        return (answers[rng.randrange(len(answers))],)
    if negatives:
        return (negatives[rng.randrange(len(negatives))],)
    return (answers[0],) if answers else ()


def _boolean_status(
    status: ReasoningStatus, targets: Sequence[str], answers: Sequence[str]
) -> ReasoningStatus:
    if status is not ReasoningStatus.SUPPORTED:
        return status
    if not targets:
        return status
    return (
        ReasoningStatus.SUPPORTED
        if set(targets) & set(answers)
        else ReasoningStatus.UNSUPPORTED
    )

	"""Intent-driven task construction: the query states what, never how.

	Why this module exists
	----------------------
	The previous generator wrote the gold relation sequence into
	``GraphQuery.required_relation_sequence`` and into the question text, then
	computed the gold answer as "every path matching that sequence". The reasoner
	filtered its own candidate expansion by the same field. The task was therefore
	"walk the sequence you were given", which exhaustive traversal solves perfectly
	and which no amount of training can improve. Every metric pinned at 1.0.

	Here a task is built from a :class:`~graph.synthetic.ontology.QueryIntent`:

	* the query carries a natural-language paraphrase and the start entity;
	* the relation composition that answers it is never placed in the query;
	* ``max_hops`` is a stage-level bound shared by every intent at that stage, so
	path length is not a giveaway either;
	* gold answers are the endpoints of composition-matching paths, while endpoints
	of other same-length paths from the same start become hard negatives.

	A search that does not understand the query therefore cannot choose among the
	competing endpoints, which is exactly the discrimination PLAN sections 12.5 and
	16 ask the model to learn.
	"""

	from __future__ import annotations

	import random
	from collections.abc import Mapping, Sequence
	from dataclasses import dataclass

	from graph.graph import EdgeRecord, GraphStore
	from graph.synthetic.ontology import (
	RELATION_BY_NAME,
	QueryIntent,
	)
	from graph.types import GraphPath, GraphQuery, PathStep, ReasoningStatus

	#: Stage-level hop budgets. Every intent at a stage shares the same bound so
	#: ``max_hops`` never reveals the composition length.
	STAGE_HOP_BUDGET: Mapping[int, int] = {
	0: 2,
	1: 2,
	2: 3,
	3: 4,
	4: 4,
	5: 5,
	6: 5,
	7: 5,
	8: 5,
	9: 5,
	10: 5,
	11: 5,
	12: 5,
	13: 6,
	14: 6,
	}


	@dataclass(frozen=True)
	class TaskBuild:
	"""The outcome of constructing one task over a (possibly mutated) graph."""

	graph: GraphStore
	query: GraphQuery
	answers: tuple[str, ...]
	negative_answers: tuple[str, ...]
	supporting_paths: tuple[GraphPath, ...]
	competing_paths: tuple[GraphPath, ...]
	conflicting_paths: tuple[GraphPath, ...]
	status: ReasoningStatus
	intent_id: str
	composition: tuple[str, ...]
	start_id: str


	def paths_matching_composition(
	graph: GraphStore, start_id: str, composition: Sequence[str]
	) -> tuple[GraphPath, ...]:
	"""Enumerate simple paths from ``start_id`` that walk ``composition`` exactly."""
	results: list[GraphPath] = []

	def walk(
	node: str, index: int, steps: tuple[PathStep, ...], visited: frozenset[str]
	) -> None:
	if index == len(composition):
	results.append(GraphPath(steps=steps))
	return
	for edge in graph.outgoing(node):
	if edge.relation_id != composition[index] or edge.target_id in visited:
	continue
	walk(
	edge.target_id,
	index + 1,
	(
	*steps,
	PathStep(
	source_id=edge.source_id,
	relation_id=edge.relation_id,
	target_id=edge.target_id,
	source_type=edge.source_type,
	target_type=edge.target_type,
	edge_attributes=edge.attributes,
	),
	),
	visited \| {edge.target_id},
	)

	walk(start_id, 0, (), frozenset({start_id}))
	return tuple(results)


	def competing_paths(
	graph: GraphStore,
	start_id: str,
	length: int,
	composition: Sequence[str],
	*,
	limit: int = 64,
	) -> tuple[GraphPath, ...]:
	"""Same-length paths from the same start that do not walk ``composition``.

	These are the hard negatives: identical hop count, identical origin, and
	often an endpoint of the same entity type. Only the relation sequence
	differs, so discriminating them requires reading the query.
	"""
	found: list[GraphPath] = []
	for path in graph.iter_paths((start_id,), length):
	if path.hop_count != length:
	continue
	if tuple(path.relation_sequence) == tuple(composition):
	continue
	found.append(path)
	if len(found) >= limit:
	break
	return tuple(found)


	def plant_composition(
	graph: GraphStore,
	start_id: str,
	composition: Sequence[str],
	type_lookup: Mapping[str, str],
	relation_lookup: Mapping[str, str],
	rng: random.Random,
	) -> GraphStore:
	"""Add the minimum edges needed for ``composition`` to be walkable from ``start_id``.

	Planting keeps every generated task answerable without weakening it: the
	planted path is one of many same-length routes out of ``start_id``, so the
	model still has to pick the right one.
	"""
	nodes_by_type: dict[str, list[str]] = {}
	for node_id, node_type in graph.nodes.items():
	nodes_by_type.setdefault(node_type, []).append(node_id)

	edges = list(graph.edges)
	current = start_id
	visited = {start_id}
	for canonical in composition:
	spec = RELATION_BY_NAME[canonical]
	# Use the builder's own canonical -> graph-local mapping. Inferring it
	# from edge type signatures is ambiguous whenever two relations share a
	# domain/range pair (``member_of`` and ``manages`` are both
	# Person -> Team), and on a renamed graph that ambiguity silently planted
	# edges under the canonical name that no query could then match.
	relation_id = relation_lookup[canonical]
	target_type = type_lookup[spec.range]
	existing = [
	edge.target_id
	for edge in graph.outgoing(current)
	if edge.relation_id == relation_id and edge.target_id not in visited
	]
	if existing:
	current = existing[0]
	visited.add(current)
	continue
	candidates = [
	node for node in nodes_by_type.get(target_type, ()) if node not in visited
	]
	if not candidates:
	raise RuntimeError(
	f"cannot plant {canonical!r}: no unused node of type {target_type!r} exists"
	)
	target = candidates[rng.randrange(len(candidates))]
	edges.append(
	EdgeRecord(
	source_id=current,
	relation_id=relation_id,
	target_id=target,
	source_type=graph.nodes[current],
	target_type=graph.nodes[target],
	relation_name=relation_id,
	attributes={"weight": "1"},
	)
	)
	current = target
	visited.add(current)
	return GraphStore(
	graph.nodes,
	edges,
	graph_version=graph.graph_version,
	node_attributes=graph.node_attributes,
	relation_attributes=graph.relation_attributes,
	)


	def candidate_starts(
	graph: GraphStore, intent: QueryIntent, type_lookup: Mapping[str, str]
	) -> tuple[str, ...]:
	"""Nodes typed as the composition's domain, ordered for determinism."""
	domain = type_lookup[RELATION_BY_NAME[intent.composition[0]].domain]
	return tuple(node for node, node_type in graph.nodes.items() if node_type == domain)


	def build_task(
	graph: GraphStore,
	intent: QueryIntent,
	*,
	stage: int,
	rng: random.Random,
	type_lookup: Mapping[str, str],
	relation_lookup: Mapping[str, str],
	answer_mode: str = "entity",
	held_out_templates: bool = False,
	variant: str = "standard",
	) -> TaskBuild:
	"""Construct one well-posed task for ``intent`` over ``graph``.

	``variant`` selects the reasoning scenario: ``standard`` answers the intent,
	``missing_edge`` removes the supporting evidence so the correct response is
	``INSUFFICIENT_EVIDENCE``, and ``contradiction`` introduces a second
	conflicting value for the same subject/relation claim.
	"""
	composition = tuple(relation_lookup[name] for name in intent.composition)
	starts = candidate_starts(graph, intent, type_lookup)
	if not starts:
	raise RuntimeError(
	f"graph has no node of the domain type required by intent {intent.intent_id!r}"
	)

	start_id = _select_start(graph, starts, composition, rng)
	supporting = paths_matching_composition(graph, start_id, composition)
	if not supporting:
	graph = plant_composition(
	graph, start_id, intent.composition, type_lookup, relation_lookup, rng
	)
	supporting = paths_matching_composition(graph, start_id, composition)
	if not supporting:
	raise RuntimeError(
	f"planting failed for intent {intent.intent_id!r} from {start_id!r}"
	)

	conflicts: tuple[GraphPath, ...] = ()
	status = ReasoningStatus.SUPPORTED

	if variant == "missing_edge":
	graph = _remove_final_edges(graph, supporting)
	supporting = paths_matching_composition(graph, start_id, composition)
	if not supporting:
	status = ReasoningStatus.INSUFFICIENT_EVIDENCE
	elif variant == "contradiction":
	graph, conflicts = _introduce_contradiction(graph, supporting, rng)
	supporting = paths_matching_composition(graph, start_id, composition)
	if conflicts:
	status = ReasoningStatus.CONTRADICTED

	competing = competing_paths(graph, start_id, len(composition), composition)
	answers = (
	tuple(
	dict.fromkeys(path.endpoint_id for path in supporting if path.endpoint_id)
	)
	if status in {ReasoningStatus.SUPPORTED, ReasoningStatus.CONTRADICTED}
	else ()
	)
	negatives = tuple(
	dict.fromkeys(
	path.endpoint_id
	for path in competing
	if path.endpoint_id and path.endpoint_id not in set(answers)
	)
	)

	templates = intent.held_out_templates if held_out_templates else intent.templates
	if not templates:
	templates = intent.templates
	text = templates[rng.randrange(len(templates))].format(start=start_id)

	budget = STAGE_HOP_BUDGET.get(stage, 5)
	max_hops = max(budget, len(composition))
	query = GraphQuery(
	text=text,
	start_entities=(start_id,),
	start_entity_ids=(start_id,),
	# Deliberately empty: the composition, its length, and the answer type
	# are all withheld. They are the answer, not the question.
	allowed_relations=(),
	required_relations=(),
	required_relation_sequence=(),
	required_types=(),
	min_hops=1,
	max_hops=max_hops,
	exact_hops=None,
	return_paths=True,
	verify=True,
	operation="find_entities" if answer_mode == "entity" else answer_mode,
	answer_mode=answer_mode,
	target_entity_ids=_target_ids(answer_mode, answers, negatives, status, rng),
	)

	if answer_mode == "boolean":
	status = _boolean_status(status, query.target_entity_ids, answers)

	return TaskBuild(
	graph=graph,
	query=query,
	answers=answers,
	negative_answers=negatives[:16],
	supporting_paths=supporting,
	competing_paths=competing[:16],
	conflicting_paths=conflicts,
	status=status,
	intent_id=intent.intent_id,
	composition=tuple(intent.composition),
	start_id=start_id,
	)


	def _select_start(
	graph: GraphStore,
	starts: Sequence[str],
	composition: Sequence[str],
	rng: random.Random,
	) -> str:
	"""Prefer a start that already walks the composition and has competitors.

	A start with competing same-length routes makes the task discriminative; one
	without them would be answerable by hop-count alone.
	"""
	walkable = [
	node for node in starts if paths_matching_composition(graph, node, composition)
	]
	contested = [
	node
	for node in walkable
	if competing_paths(graph, node, len(composition), composition, limit=2)
	]
	pool = contested or walkable or list(starts)
	return pool[rng.randrange(len(pool))]


	def _remove_final_edges(
	graph: GraphStore, supporting: Sequence[GraphPath]
	) -> GraphStore:
	"""Delete the last edge of every supporting path, removing the evidence."""
	doomed = {
	(path.steps[-1].source_id, path.steps[-1].relation_id, path.steps[-1].target_id)
	for path in supporting
	if path.steps
	}
	edges = [
	edge
	for edge in graph.edges
	if (edge.source_id, edge.relation_id, edge.target_id) not in doomed
	]
	return GraphStore(
	graph.nodes,
	edges,
	graph_version=graph.graph_version,
	node_attributes=graph.node_attributes,
	relation_attributes=graph.relation_attributes,
	)


	def _introduce_contradiction(
	graph: GraphStore, supporting: Sequence[GraphPath], rng: random.Random
	) -> tuple[GraphStore, tuple[GraphPath, ...]]:
	"""Add a second, conflicting target for the first claim on a supporting path.

	A contradiction is a single subject/relation asserting two different
	values. Two different multi-hop routes to different endpoints are competing
	explanations, not a contradiction, and are handled separately.
	"""
	if not supporting or not supporting[0].steps:
	return graph, ()
	first = supporting[0].steps[0]
	same_type = [
	node
	for node, node_type in graph.nodes.items()
	if node_type == first.target_type
	and node not in {first.source_id, first.target_id}
	]
	if not same_type:
	return graph, ()
	rival = same_type[rng.randrange(len(same_type))]
	if graph.has_edge(first.source_id, first.relation_id, rival):
	return graph, ()
	edges = [
	*graph.edges,
	EdgeRecord(
	source_id=first.source_id,
	relation_id=first.relation_id,
	target_id=rival,
	source_type=graph.nodes[first.source_id],
	target_type=graph.nodes[rival],
	relation_name=first.relation_id,
	attributes={"contradictory": "true"},
	),
	]
	mutated = GraphStore(
	graph.nodes,
	edges,
	graph_version=graph.graph_version,
	node_attributes=graph.node_attributes,
	relation_attributes=graph.relation_attributes,
	)
	claims = tuple(
	GraphPath(
	steps=(
	PathStep(
	source_id=first.source_id,
	relation_id=first.relation_id,
	target_id=target,
	source_type=graph.nodes[first.source_id],
	target_type=graph.nodes[target],
	),
	)
	)
	for target in (first.target_id, rival)
	)
	return mutated, claims


	def _target_ids(
	answer_mode: str,
	answers: Sequence[str],
	negatives: Sequence[str],
	status: ReasoningStatus,
	rng: random.Random,
	) -> tuple[str, ...]:
	"""Boolean tasks need a concrete claim to judge; other modes need none."""
	if answer_mode != "boolean":
	return ()
	truthful = bool(answers) and rng.random() < 0.5
	if truthful:
	return (answers[rng.randrange(len(answers))],)
	if negatives:
	return (negatives[rng.randrange(len(negatives))],)
	return (answers[0],) if answers else ()


	def _boolean_status(
	status: ReasoningStatus, targets: Sequence[str], answers: Sequence[str]
	) -> ReasoningStatus:
	if status is not ReasoningStatus.SUPPORTED:
	return status
	if not targets:
	return status
	return (
	ReasoningStatus.SUPPORTED
	if set(targets) & set(answers)
	else ReasoningStatus.UNSUPPORTED
	)