Profiles

def apply_all(graph: KnowledgeGraph) -> list[ValidationResult]:
    """Validate every assertion node and write each status back.

    Args:
        graph: The graph to validate and update in place.

    Returns:
        The results, one per assertion node.

    """
    return [apply_validation(graph, nid) for nid in list(graph.nodes) if graph.nodes[nid].type in ASSERTION_TYPES]

def apply_validation(graph: KnowledgeGraph, node_id: str) -> ValidationResult:
    """Validate a node and write the resolved status into its properties.

    Args:
        graph: The graph.
        node_id: The node to validate and update.

    Returns:
        The validation result whose status was applied.

    """
    result = validate_node(graph, node_id)
    node = graph.nodes[node_id]
    node.properties["status"] = result.status
    node.touch()
    return result

def conclusion_status(graph: KnowledgeGraph) -> ConclusionResult:
    """Report whether a project graph's question has been answered.

    Args:
        graph: The project graph.

    Returns:
        A :class:`ConclusionResult` with the blocking reasons, if any.
    """
    reasons: list[str] = []
    questions = graph.nodes_of_type(_QUESTION)
    if not questions:
        reasons.append("no question node in graph")
    for question in questions:
        answers = graph.incoming(question.id, "answers")
        if not answers:
            reasons.append(f"question {question.id!r} has no answering result/finding")
        for answer in answers:
            if not graph.outgoing(answer.source, "supported_by"):
                reasons.append(f"answer {answer.source!r} to {question.id!r} has no supporting evidence")
        for edge in graph.incoming(question.id, "addresses"):
            hypothesis = graph.nodes.get(edge.source)
            status = hypothesis.properties.get("status") if hypothesis else None
            if status not in _RESOLVED:
                reasons.append(f"hypothesis {edge.source!r} is unresolved (status={status!r})")
    return ConclusionResult(concluded=not reasons, reasons=tuple(reasons))

def coverage(graph: KnowledgeGraph) -> dict[str, Coverage]:
    """Classify support origin for every assertion node in a graph.

    Args:
        graph: The graph to analyse.

    Returns:
        A mapping from assertion node id to its coverage.

    """
    result: dict[str, Coverage] = {}
    for node in graph.nodes.values():
        if node.type not in ASSERTION_TYPES:
            continue
        own = literature = other = False
        kinds: list[str] = []
        for edge in graph.outgoing(node.id, "supported_by"):
            evidence = graph.nodes.get(edge.target)
            if evidence is None or evidence.type != _EVIDENCE:
                continue
            kind = str(evidence.properties.get("kind", "other"))
            kinds.append(kind)
            if kind == "literature":
                literature = True
            elif kind in _OWN_KINDS:
                own = True
            else:
                other = True
        result[node.id] = Coverage(node.id, own, literature, other, kinds)
    return result

def evidence_changed(node: Node, current_digest: str | None) -> bool:
    """Report whether an evidence node's artefact has drifted.

    Args:
        node: The evidence node holding a baseline ``digest`` property.
        current_digest: The artefact's current digest, or None if unknown.

    Returns:
        ``True`` if both digests are present and differ.

    """
    baseline = node.properties.get("digest")
    if not baseline or current_digest is None:
        return False
    return baseline != current_digest

def find_stale_assertions(graph: KnowledgeGraph, resolver: DigestResolver) -> list[Node]:
    """Return assertion nodes with at least one changed supporting evidence.

    Args:
        graph: The graph to sweep.
        resolver: Resolves an evidence node's current digest (or None).

    Returns:
        The affected assertion nodes, in insertion order.

    """
    out: list[Node] = []
    for node in graph.nodes.values():
        if node.type not in ASSERTION_TYPES:
            continue
        for eid in _linked_evidence(graph, node.id, "supported_by"):
            if evidence_changed(graph.nodes[eid], resolver(graph.nodes[eid])):
                out.append(node)
                break
    return out

def find_stale_evidence(graph: KnowledgeGraph, resolver: DigestResolver) -> list[Node]:
    """Return evidence nodes whose artefact has changed.

    Args:
        graph: The graph to sweep.
        resolver: Resolves an evidence node's current digest (or None).

    Returns:
        The changed evidence nodes.

    """
    return [n for n in graph.nodes.values() if n.type == _EVIDENCE and evidence_changed(n, resolver(n))]

def graph_from_legacy(data: dict[str, Any]) -> KnowledgeGraph:
    """Convert a legacy provenance-graph dict into a KnowledgeGraph.

    Args:
        data: A legacy ``ProvenanceGraph.to_dict()`` mapping (statements or the
            pre-v2 ``claims`` key, plus evidence/activities/relations/
            cross_references).

    Returns:
        The equivalent generic knowledge graph (research profile).

    """
    graph = KnowledgeGraph(article_id=data.get("article_id"))
    if data.get("metadata") is not None:
        graph.metadata = dict(data["metadata"])
    for item in data.get("statements", data.get("claims", [])):
        graph.add_node(_statement_node(item))
    for item in data.get("evidence", []):
        graph.add_node(_evidence_node(item))
    for item in data.get("activities", []):
        graph.add_node(_activity_node(item))
    for item in data.get("relations", []):
        graph.add_edge(_edge(item, "subject_id", "object_id"))
    for item in data.get("cross_references", []):
        graph.add_edge(_edge(item, "subject_id", "target"))
    return graph

def mark_stale(graph: KnowledgeGraph, resolver: DigestResolver) -> list[Node]:
    """Flip affected ``valid`` assertion nodes to ``stale`` and return them.

    Args:
        graph: The graph to update in place.
        resolver: Resolves an evidence node's current digest (or None).

    Returns:
        The assertion nodes whose status changed to ``stale``.

    """
    changed: list[Node] = []
    for node in find_stale_assertions(graph, resolver):
        if node.properties.get("status") == VALID:
            node.properties["status"] = STALE
            node.touch()
            changed.append(node)
    return changed

def promote(graph: KnowledgeGraph, *, article_id: str) -> KnowledgeGraph:
    """Promote a concluded project's own knowledge into a new article graph.

    Keeps every node the project produced itself (those without a
    ``source_gid`` stamp) and the edges among them; edges into imported nodes are
    rewired as cross-article references to the imported node's origin.

    Args:
        graph: The concluded project graph.
        article_id: ``article_id`` for the promoted article graph.

    Returns:
        A new :class:`KnowledgeGraph` holding the promoted knowledge.

    Raises:
        ValueError: If the project is not concluded.
    """
    result = conclusion_status(graph)
    if not result.concluded:
        raise ValueError("project is not concluded: " + "; ".join(result.reasons))

    keep = {nid for nid, node in graph.nodes.items() if SOURCE_GID_KEY not in node.properties}
    out = KnowledgeGraph(article_id=article_id, metadata={"promoted_from": graph.article_id})
    for nid in keep:
        node = graph.nodes[nid]
        out.add_node(
            Node(
                type=node.type,
                text=node.text,
                id=node.id,
                tags=list(node.tags),
                properties=dict(node.properties),
                created_at=node.created_at,
                updated_at=node.updated_at,
            )
        )
    for edge in graph.edges.values():
        if edge.source not in keep:
            continue
        if edge.target in keep:
            _copy_edge(out, edge, edge.target)
            continue
        target = graph.nodes.get(edge.target)
        origin = target.properties.get(SOURCE_GID_KEY) if target else None
        if origin is not None:
            _copy_edge(out, edge, origin)
    return out

def validate_all(graph: KnowledgeGraph) -> dict[str, ValidationResult]:
    """Validate every assertion node without mutating the graph.

    Args:
        graph: The graph to validate.

    Returns:
        A mapping from node id to validation result.

    """
    return {n.id: validate_node(graph, n.id) for n in graph.nodes.values() if n.type in ASSERTION_TYPES}

def validate_node(graph: KnowledgeGraph, node_id: str) -> ValidationResult:
    """Resolve an assertion node's status from its supporting/refuting evidence.

    Args:
        graph: The graph holding the node and its evidence.
        node_id: The id of the node to validate.

    Returns:
        The validation result.

    Raises:
        KeyError: If the node is not held by the graph.

    """
    if node_id not in graph.nodes:
        raise KeyError(node_id)
    supporting = _linked_evidence(graph, node_id, "supported_by")
    refuting = _linked_evidence(graph, node_id, "refuted_by")
    if not supporting and not refuting:
        return ValidationResult(node_id, UNRESOLVED, supporting, refuting, "no supporting or refuting evidence linked")
    if supporting and refuting:
        reason = f"conflicting evidence: {len(supporting)} supporting, {len(refuting)} refuting"
        return ValidationResult(node_id, NEEDS_REVIEW, supporting, refuting, reason)
    if refuting:
        return ValidationResult(node_id, INVALID, supporting, refuting, f"refuted by {len(refuting)} piece(s)")
    return ValidationResult(node_id, VALID, supporting, refuting, f"supported by {len(supporting)} piece(s)")