routers/governed_graph_operations.py

# SPDX-License-Identifier: Apache-2.0
# Copyright 2026 Lutar, Stephen P. - SZL Holdings
# Signed-off-by: Stephen Lutar <stephenlutar2@gmail.com>
"""Governed Graph Operations: deterministic topology analysis for A11oy.

Taxonomy home: governance/services.  This module turns a proposed graph of
bounded jobs and local loops into an auditable execution *plan*.  It never runs
agents, calls providers, mutates repositories, or emits a receipt: the API is a
pure, deterministic analysis surface.  A later executor must re-check the
returned contract against policy and obtain the required receipts/human gates.

The design deliberately distinguishes:

* data dependencies from control/resource dependencies;
* bounded local loops from illegal top-level graph cycles;
* declared parallel work from hidden shared-resource conflicts;
* expected fan-in from structurally received inputs;
* maker context from an independent verifier context; and
* graph agreement from external truth anchors.

Doctrine v11: analysis is MODELED, side effects are zero, and GET signs nothing.
"""

import hashlib
import json
from pathlib import Path
from typing import Any


SCHEMA = "szl.governed-graph/v1"
IMPLEMENTATION_STATUS = "REAL"
EVIDENCE_LABEL = "MODELED"
EXECUTION_MODE = "PLAN_ONLY"
MAX_BODY_BYTES = 96 * 1024
MAX_NODES = 64
MAX_LIST_ITEMS = 64
MAX_TEXT = 512

_ROLES = {
    "scope",
    "planner",
    "worker",
    "reducer",
    "verifier",
    "governance",
    "human_gate",
    "synthesizer",
    "loop",
    "publisher",
}
_AUTHORITIES = {"READ_ONLY", "PROPOSE", "WRITE_GOVERNED", "HUMAN"}
_ANCHOR_TYPES = {"test", "source", "human", "receipt", "runtime"}
_TOP_LEVEL_KEYS = {
    "schema",
    "graph_id",
    "goal",
    "external_inputs",
    "nodes",
    "anchors",
    "budget",
}
_NODE_KEYS = {
    "id",
    "label",
    "role",
    "depends_on",
    "control_after",
    "consumes",
    "produces",
    "reads",
    "writes",
    "resources",
    "fresh_context",
    "verifier_for",
    "side_effecting",
    "authority",
    "max_iterations",
    "exit_conditions",
}
_ANCHOR_KEYS = {"id", "type", "nodes", "required", "description"}
_BUDGET_KEYS = {
    "max_nodes",
    "max_parallel",
    "max_depth",
    "max_total_iterations",
}


class GraphContractError(ValueError):
    """Raised when a graph request does not satisfy the bounded input contract."""


def _canonical_json(value: Any) -> str:
    return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True)


def _sha256(value: Any) -> str:
    return hashlib.sha256(_canonical_json(value).encode("utf-8")).hexdigest()


def _plain_object(value: Any, field: str) -> dict:
    if not isinstance(value, dict):
        raise GraphContractError(f"{field} must be an object")
    return value


def _bounded_text(value: Any, field: str, *, maximum: int = MAX_TEXT) -> str:
    if not isinstance(value, str):
        raise GraphContractError(f"{field} must be a string")
    cleaned = value.strip()
    if not cleaned or len(cleaned) > maximum:
        raise GraphContractError(f"{field} must contain 1-{maximum} characters")
    return cleaned


def _string_list(value: Any, field: str, *, maximum: int = MAX_LIST_ITEMS) -> list[str]:
    if value is None:
        return []
    if not isinstance(value, list) or len(value) > maximum:
        raise GraphContractError(f"{field} must be a list with at most {maximum} items")
    result = []
    for index, item in enumerate(value):
        result.append(_bounded_text(item, f"{field}[{index}]", maximum=128))
    if len(result) != len(set(result)):
        raise GraphContractError(f"{field} contains duplicate values")
    return result


def _bounded_int(value: Any, field: str, *, minimum: int, maximum: int) -> int:
    if isinstance(value, bool) or not isinstance(value, int):
        raise GraphContractError(f"{field} must be an integer")
    if value < minimum or value > maximum:
        raise GraphContractError(f"{field} must be between {minimum} and {maximum}")
    return value


def validate_graph_contract(payload: Any) -> dict:
    """Validate and normalize the public graph contract; reject unknown fields."""

    raw = _plain_object(payload, "request")
    unknown = sorted(set(raw) - _TOP_LEVEL_KEYS)
    if unknown:
        raise GraphContractError(f"unknown top-level fields: {', '.join(unknown)}")
    if raw.get("schema") != SCHEMA:
        raise GraphContractError(f"schema must equal {SCHEMA}")

    graph_id = _bounded_text(raw.get("graph_id"), "graph_id", maximum=128)
    goal = _bounded_text(raw.get("goal"), "goal", maximum=2048)
    external_inputs = _string_list(raw.get("external_inputs", []), "external_inputs")

    raw_nodes = raw.get("nodes")
    if not isinstance(raw_nodes, list) or not raw_nodes:
        raise GraphContractError("nodes must be a non-empty list")
    if len(raw_nodes) > MAX_NODES:
        raise GraphContractError(f"nodes exceeds the hard cap of {MAX_NODES}")

    nodes = []
    seen_ids = set()
    for index, item in enumerate(raw_nodes):
        node = _plain_object(item, f"nodes[{index}]")
        extra = sorted(set(node) - _NODE_KEYS)
        if extra:
            raise GraphContractError(
                f"nodes[{index}] has unknown fields: {', '.join(extra)}"
            )
        node_id = _bounded_text(node.get("id"), f"nodes[{index}].id", maximum=64)
        if node_id in seen_ids:
            raise GraphContractError(f"duplicate node id: {node_id}")
        seen_ids.add(node_id)
        role = _bounded_text(node.get("role"), f"nodes[{index}].role", maximum=32)
        if role not in _ROLES:
            raise GraphContractError(f"unsupported role for {node_id}: {role}")
        authority = node.get("authority", "READ_ONLY")
        if authority not in _AUTHORITIES:
            raise GraphContractError(f"unsupported authority for {node_id}: {authority}")
        side_effecting = node.get("side_effecting", False)
        fresh_context = node.get("fresh_context", False)
        if not isinstance(side_effecting, bool) or not isinstance(fresh_context, bool):
            raise GraphContractError(
                f"side_effecting and fresh_context must be booleans for {node_id}"
            )
        max_iterations = node.get("max_iterations")
        if max_iterations is not None:
            max_iterations = _bounded_int(
                max_iterations,
                f"nodes[{index}].max_iterations",
                minimum=1,
                maximum=32,
            )
        normalized = {
            "id": node_id,
            "label": _bounded_text(
                node.get("label", node_id), f"nodes[{index}].label", maximum=128
            ),
            "role": role,
            "depends_on": _string_list(
                node.get("depends_on", []), f"nodes[{index}].depends_on"
            ),
            "control_after": _string_list(
                node.get("control_after", []), f"nodes[{index}].control_after"
            ),
            "consumes": _string_list(
                node.get("consumes", []), f"nodes[{index}].consumes"
            ),
            "produces": _string_list(
                node.get("produces", []), f"nodes[{index}].produces"
            ),
            "reads": _string_list(node.get("reads", []), f"nodes[{index}].reads"),
            "writes": _string_list(node.get("writes", []), f"nodes[{index}].writes"),
            "resources": _string_list(
                node.get("resources", []), f"nodes[{index}].resources"
            ),
            "fresh_context": fresh_context,
            "verifier_for": _string_list(
                node.get("verifier_for", []), f"nodes[{index}].verifier_for"
            ),
            "side_effecting": side_effecting,
            "authority": authority,
            "max_iterations": max_iterations,
            "exit_conditions": _string_list(
                node.get("exit_conditions", []), f"nodes[{index}].exit_conditions"
            ),
        }
        nodes.append(normalized)

    raw_anchors = raw.get("anchors", [])
    if not isinstance(raw_anchors, list) or len(raw_anchors) > MAX_LIST_ITEMS:
        raise GraphContractError(
            f"anchors must be a list with at most {MAX_LIST_ITEMS} items"
        )
    anchors = []
    anchor_ids = set()
    for index, item in enumerate(raw_anchors):
        anchor = _plain_object(item, f"anchors[{index}]")
        extra = sorted(set(anchor) - _ANCHOR_KEYS)
        if extra:
            raise GraphContractError(
                f"anchors[{index}] has unknown fields: {', '.join(extra)}"
            )
        anchor_id = _bounded_text(anchor.get("id"), f"anchors[{index}].id", maximum=64)
        if anchor_id in anchor_ids:
            raise GraphContractError(f"duplicate anchor id: {anchor_id}")
        anchor_ids.add(anchor_id)
        anchor_type = _bounded_text(
            anchor.get("type"), f"anchors[{index}].type", maximum=32
        )
        if anchor_type not in _ANCHOR_TYPES:
            raise GraphContractError(f"unsupported anchor type: {anchor_type}")
        required = anchor.get("required", True)
        if not isinstance(required, bool):
            raise GraphContractError(f"anchors[{index}].required must be boolean")
        anchors.append(
            {
                "id": anchor_id,
                "type": anchor_type,
                "nodes": _string_list(
                    anchor.get("nodes", []), f"anchors[{index}].nodes"
                ),
                "required": required,
                "description": _bounded_text(
                    anchor.get("description", anchor_id),
                    f"anchors[{index}].description",
                    maximum=256,
                ),
            }
        )

    raw_budget = _plain_object(raw.get("budget", {}), "budget")
    extra_budget = sorted(set(raw_budget) - _BUDGET_KEYS)
    if extra_budget:
        raise GraphContractError(f"budget has unknown fields: {', '.join(extra_budget)}")
    budget = {
        "max_nodes": _bounded_int(
            raw_budget.get("max_nodes", 32), "budget.max_nodes", minimum=1, maximum=64
        ),
        "max_parallel": _bounded_int(
            raw_budget.get("max_parallel", 6),
            "budget.max_parallel",
            minimum=1,
            maximum=16,
        ),
        "max_depth": _bounded_int(
            raw_budget.get("max_depth", 12), "budget.max_depth", minimum=1, maximum=32
        ),
        "max_total_iterations": _bounded_int(
            raw_budget.get("max_total_iterations", 48),
            "budget.max_total_iterations",
            minimum=1,
            maximum=256,
        ),
    }
    return {
        "schema": SCHEMA,
        "graph_id": graph_id,
        "goal": goal,
        "external_inputs": external_inputs,
        "nodes": nodes,
        "anchors": anchors,
        "budget": budget,
    }


def _issue(code: str, message: str, *, nodes: list[str], severity: str) -> dict:
    return {
        "code": code,
        "severity": severity,
        "nodes": sorted(set(nodes)),
        "message": message,
    }


def analyse_graph(payload: Any) -> dict:
    """Return a deterministic, non-effecting topology and governance analysis."""

    graph = validate_graph_contract(payload)
    nodes = graph["nodes"]
    by_id = {node["id"]: node for node in nodes}
    blockers = []
    advisories = []

    dependencies = {}
    children = {node_id: [] for node_id in by_id}
    data_edges = []
    control_edges = []
    for node in nodes:
        node_id = node["id"]
        overlap = set(node["depends_on"]) & set(node["control_after"])
        if overlap:
            raise GraphContractError(
                f"{node_id} repeats dependencies as both data and control edges: "
                + ", ".join(sorted(overlap))
            )
        deps = list(node["depends_on"]) + list(node["control_after"])
        for dep in deps:
            if dep == node_id:
                raise GraphContractError(f"{node_id} cannot depend on itself")
            if dep not in by_id:
                raise GraphContractError(f"{node_id} references unknown dependency: {dep}")
            children[dep].append(node_id)
        dependencies[node_id] = deps
        data_edges.extend(
            {"source": dep, "target": node_id, "kind": "data"}
            for dep in node["depends_on"]
        )
        control_edges.extend(
            {"source": dep, "target": node_id, "kind": "control"}
            for dep in node["control_after"]
        )

    indegree = {node_id: len(dependencies[node_id]) for node_id in by_id}
    ready = sorted(node_id for node_id, degree in indegree.items() if degree == 0)
    layers = []
    order = []
    while ready:
        layer = ready
        layers.append(layer)
        order.extend(layer)
        next_ready = []
        for node_id in layer:
            for child in sorted(children[node_id]):
                indegree[child] -= 1
                if indegree[child] == 0:
                    next_ready.append(child)
        ready = sorted(next_ready)
    if len(order) != len(nodes):
        cycle_nodes = sorted(node_id for node_id, degree in indegree.items() if degree > 0)
        raise GraphContractError(
            "top-level graph contains a cycle; model retries as bounded loop nodes: "
            + ", ".join(cycle_nodes)
        )

    if len(nodes) > graph["budget"]["max_nodes"]:
        blockers.append(
            _issue(
                "NODE_BUDGET_EXCEEDED",
                "Declared node count exceeds the graph budget.",
                nodes=list(by_id),
                severity="BLOCKER",
            )
        )
    if len(layers) > graph["budget"]["max_depth"]:
        blockers.append(
            _issue(
                "DEPTH_BUDGET_EXCEEDED",
                "Critical-path depth exceeds the graph budget.",
                nodes=order,
                severity="BLOCKER",
            )
        )

    loop_iterations = 0
    for node in nodes:
        if node["role"] == "loop":
            if node["max_iterations"] is None or not node["exit_conditions"]:
                blockers.append(
                    _issue(
                        "UNBOUNDED_LOOP",
                        "Loop nodes require max_iterations and at least one exit condition.",
                        nodes=[node["id"]],
                        severity="BLOCKER",
                    )
                )
            else:
                loop_iterations += node["max_iterations"]
        elif node["max_iterations"] is not None or node["exit_conditions"]:
            advisories.append(
                _issue(
                    "LOOP_FIELDS_ON_NON_LOOP",
                    "Loop bounds are ignored unless the node role is loop.",
                    nodes=[node["id"]],
                    severity="ADVISORY",
                )
            )
    if loop_iterations > graph["budget"]["max_total_iterations"]:
        blockers.append(
            _issue(
                "ITERATION_BUDGET_EXCEEDED",
                "Sum of local loop caps exceeds the total iteration budget.",
                nodes=[node["id"] for node in nodes if node["role"] == "loop"],
                severity="BLOCKER",
            )
        )

    external_inputs = set(graph["external_inputs"])
    fake_edges = []
    input_gaps = []
    fan_in = []
    for node in nodes:
        supplied = set(external_inputs)
        for dep in node["depends_on"]:
            produced = set(by_id[dep]["produces"])
            overlap = sorted(produced & set(node["consumes"]))
            if not overlap:
                fake_edges.append(
                    {
                        "source": dep,
                        "target": node["id"],
                        "reason": "consumer declares no use of predecessor output",
                    }
                )
            supplied.update(produced)
        missing = sorted(set(node["consumes"]) - supplied)
        if missing:
            input_gaps.append({"node": node["id"], "missing": missing})
            blockers.append(
                _issue(
                    "INPUT_CONTRACT_GAP",
                    f"Node consumes unavailable inputs: {', '.join(missing)}.",
                    nodes=[node["id"]],
                    severity="BLOCKER",
                )
            )
        if node["role"] in {"reducer", "synthesizer"} and len(node["depends_on"]) > 1:
            received = sum(
                1
                for dep in node["depends_on"]
                if set(by_id[dep]["produces"]) & set(node["consumes"])
            )
            fan_in.append(
                {
                    "node": node["id"],
                    "expected": len(node["depends_on"]),
                    "contractually_received": received,
                    "complete": received == len(node["depends_on"]),
                }
            )
            if received != len(node["depends_on"]):
                blockers.append(
                    _issue(
                        "FAN_IN_INCOMPLETE",
                        "Reducer/synthesizer does not consume a declared output from every worker.",
                        nodes=[node["id"]] + node["depends_on"],
                        severity="BLOCKER",
                    )
                )
    for edge in fake_edges:
        advisories.append(
            _issue(
                "FAKE_DATA_EDGE",
                "No declared artifact crosses this data edge; remove it or make it a control edge.",
                nodes=[edge["source"], edge["target"]],
                severity="ADVISORY",
            )
        )

    hidden_resource_edges = []
    for layer_index, layer in enumerate(layers):
        for left_index, left_id in enumerate(layer):
            left = by_id[left_id]
            for right_id in layer[left_index + 1 :]:
                right = by_id[right_id]
                write_conflicts = sorted(
                    set(left["writes"]) & (set(right["writes"]) | set(right["reads"]))
                    | set(right["writes"]) & set(left["reads"])
                )
                resource_conflicts = sorted(set(left["resources"]) & set(right["resources"]))
                if write_conflicts or resource_conflicts:
                    conflict = {
                        "nodes": [left_id, right_id],
                        "layer": layer_index,
                        "writes": write_conflicts,
                        "resources": resource_conflicts,
                    }
                    hidden_resource_edges.append(conflict)
                    blockers.append(
                        _issue(
                            "HIDDEN_RESOURCE_EDGE",
                            "Nodes scheduled in parallel share mutable state or an exclusive resource.",
                            nodes=[left_id, right_id],
                            severity="BLOCKER",
                        )
                    )

    for node in nodes:
        if node["role"] == "verifier":
            if not node["fresh_context"]:
                blockers.append(
                    _issue(
                        "VERIFIER_CONTEXT_NOT_FRESH",
                        "Verifier must declare fresh_context=true.",
                        nodes=[node["id"]],
                        severity="BLOCKER",
                    )
                )
            if not node["verifier_for"]:
                blockers.append(
                    _issue(
                        "VERIFIER_TARGET_MISSING",
                        "Verifier must name at least one artifact-producing target.",
                        nodes=[node["id"]],
                        severity="BLOCKER",
                    )
                )
            for target in node["verifier_for"]:
                if target not in by_id:
                    raise GraphContractError(
                        f"{node['id']} verifies unknown node: {target}"
                    )
                target_artifacts = set(by_id[target]["produces"])
                consumed_artifacts = set(node["consumes"])
                if (
                    target not in node["depends_on"]
                    or not target_artifacts & consumed_artifacts
                ):
                    blockers.append(
                        _issue(
                            "VERIFIER_ARTIFACT_NOT_BOUND",
                            "Verifier target must be a data dependency with a consumed artifact.",
                            nodes=[node["id"], target],
                            severity="BLOCKER",
                        )
                    )

    anchor_nodes = set()
    required_anchor_types = set()
    for anchor in graph["anchors"]:
        for node_id in anchor["nodes"]:
            if node_id not in by_id:
                raise GraphContractError(
                    f"anchor {anchor['id']} references unknown node: {node_id}"
                )
        if anchor["required"]:
            required_anchor_types.add(anchor["type"])
            anchor_nodes.update(anchor["nodes"])
    terminals = sorted(node_id for node_id, outgoing in children.items() if not outgoing)
    if not required_anchor_types & {"test", "source", "human", "runtime"}:
        blockers.append(
            _issue(
                "EXTERNAL_TRUTH_ANCHOR_MISSING",
                "At least one required test, source, runtime, or human anchor is required.",
                nodes=terminals,
                severity="BLOCKER",
            )
        )
    unanchored_terminals = sorted(set(terminals) - anchor_nodes)
    if unanchored_terminals:
        blockers.append(
            _issue(
                "TERMINAL_NOT_ANCHORED",
                "Every terminal outcome must be covered by a declared anchor.",
                nodes=unanchored_terminals,
                severity="BLOCKER",
            )
        )

    for node in nodes:
        if not node["side_effecting"]:
            continue
        if node["authority"] not in {"WRITE_GOVERNED", "HUMAN"}:
            blockers.append(
                _issue(
                    "WRITE_AUTHORITY_MISSING",
                    "Side-effecting nodes require WRITE_GOVERNED or HUMAN authority.",
                    nodes=[node["id"]],
                    severity="BLOCKER",
                )
            )
        receipt_bound = any(
            anchor["type"] == "receipt"
            and anchor["required"]
            and node["id"] in anchor["nodes"]
            for anchor in graph["anchors"]
        )
        if not receipt_bound:
            blockers.append(
                _issue(
                    "WRITE_RECEIPT_ANCHOR_MISSING",
                    "Side-effecting nodes require a mandatory receipt anchor.",
                    nodes=[node["id"]],
                    severity="BLOCKER",
                )
            )

    distance = {}
    predecessor = {}
    for node_id in order:
        deps = dependencies[node_id]
        if not deps:
            distance[node_id] = 1
            predecessor[node_id] = None
            continue
        best = max(deps, key=lambda dep: (distance[dep], dep))
        distance[node_id] = distance[best] + 1
        predecessor[node_id] = best
    critical_end = max(order, key=lambda node_id: (distance[node_id], node_id))
    critical_path = []
    cursor = critical_end
    while cursor is not None:
        critical_path.append(cursor)
        cursor = predecessor[cursor]
    critical_path.reverse()

    max_parallel = graph["budget"]["max_parallel"]
    schedule = []
    batch_index = 0
    for layer_index, layer in enumerate(layers):
        for start in range(0, len(layer), max_parallel):
            batch = layer[start : start + max_parallel]
            schedule.append(
                {"batch": batch_index, "topology_layer": layer_index, "nodes": batch}
            )
            batch_index += 1

    blockers = sorted(
        blockers,
        key=lambda issue: (issue["code"], issue["nodes"], issue["message"]),
    )
    advisories = sorted(
        advisories,
        key=lambda issue: (issue["code"], issue["nodes"], issue["message"]),
    )
    contract_digest = _sha256(graph)
    return {
        "ok": True,
        "schema": "szl.governed-graph.analysis/v1",
        "graph_id": graph["graph_id"],
        "contract_digest": contract_digest,
        "plan_id": f"ggp-{contract_digest[:20]}",
        "implementation_status": IMPLEMENTATION_STATUS,
        "evidence_label": EVIDENCE_LABEL,
        "decision": "READY_TO_ORCHESTRATE" if not blockers else "REVISE",
        "execution": {
            "mode": EXECUTION_MODE,
            "authorized": False,
            "effectors": 0,
            "provider_calls": 0,
            "writes": 0,
            "note": (
                "This endpoint computes a plan only. An executor must re-validate "
                "the exact digest, pass policy, obtain human approval where declared, "
                "and emit receipts for every write."
            ),
        },
        "topology": {
            "node_count": len(nodes),
            "data_edge_count": len(data_edges),
            "control_edge_count": len(control_edges),
            "layer_count": len(layers),
            "max_declared_parallel": max((len(layer) for layer in layers), default=0),
            "scheduled_parallel_cap": max_parallel,
            "layers": layers,
            "schedule": schedule,
            "critical_path": critical_path,
            "critical_path_nodes": len(critical_path),
            "terminals": terminals,
            "edges": data_edges + control_edges,
        },
        "contracts": {
            "fake_edges": fake_edges,
            "input_gaps": input_gaps,
            "hidden_resource_edges": hidden_resource_edges,
            "fan_in": fan_in,
            "anchor_types": sorted(required_anchor_types),
            "anchored_nodes": sorted(anchor_nodes),
            "bounded_loop_iterations": loop_iterations,
        },
        "gates": {
            "pass": not blockers,
            "blocker_count": len(blockers),
            "advisory_count": len(advisories),
            "blockers": blockers,
            "advisories": advisories,
        },
        "normalized_contract": graph,
    }


def _node(
    node_id: str,
    label: str,
    role: str,
    *,
    depends_on: list[str] | None = None,
    control_after: list[str] | None = None,
    consumes: list[str] | None = None,
    produces: list[str] | None = None,
    writes: list[str] | None = None,
    resources: list[str] | None = None,
    fresh_context: bool = False,
    verifier_for: list[str] | None = None,
    side_effecting: bool = False,
    authority: str = "READ_ONLY",
    max_iterations: int | None = None,
    exit_conditions: list[str] | None = None,
) -> dict:
    return {
        "id": node_id,
        "label": label,
        "role": role,
        "depends_on": depends_on or [],
        "control_after": control_after or [],
        "consumes": consumes or [],
        "produces": produces or [],
        "reads": [],
        "writes": writes or [],
        "resources": resources or [],
        "fresh_context": fresh_context,
        "verifier_for": verifier_for or [],
        "side_effecting": side_effecting,
        "authority": authority,
        "max_iterations": max_iterations,
        "exit_conditions": exit_conditions or [],
    }


def sample_contract(sample_id: str = "protected-release") -> dict:
    """Return one of the audited, deterministic demonstration contracts."""

    common_budget = {
        "max_nodes": 24,
        "max_parallel": 4,
        "max_depth": 10,
        "max_total_iterations": 24,
    }
    if sample_id == "research-diamond":
        return {
            "schema": SCHEMA,
            "graph_id": "research-diamond",
            "goal": "Triangulate a claim from independent primary sources.",
            "external_inputs": ["research.question"],
            "nodes": [
                _node("scope", "Freeze question and source rules", "scope", consumes=["research.question"], produces=["scope.contract"]),
                _node("papers", "Primary paper review", "worker", depends_on=["scope"], consumes=["scope.contract"], produces=["finding.papers"]),
                _node("repos", "Licensed repository review", "worker", depends_on=["scope"], consumes=["scope.contract"], produces=["finding.repos"]),
                _node("operators", "Production operator evidence", "worker", depends_on=["scope"], consumes=["scope.contract"], produces=["finding.operators"]),
                _node("reduce", "Deterministic evidence reduce", "reducer", depends_on=["papers", "repos", "operators"], consumes=["finding.papers", "finding.repos", "finding.operators"], produces=["evidence.bundle"]),
                _node("verify", "Fresh-context source verification", "verifier", depends_on=["reduce"], consumes=["evidence.bundle"], produces=["verification.result"], fresh_context=True, verifier_for=["reduce"]),
                _node("synthesize", "Cited answer or abstention", "synthesizer", depends_on=["verify"], consumes=["verification.result"], produces=["research.answer"]),
            ],
            "anchors": [
                {"id": "source-resolution", "type": "source", "nodes": ["verify", "synthesize"], "required": True, "description": "Every material claim resolves to a primary source."},
            ],
            "budget": common_budget,
        }
    if sample_id == "bounded-repair":
        return {
            "schema": SCHEMA,
            "graph_id": "bounded-repair",
            "goal": "Repair a failing change without an unbounded retry cycle.",
            "external_inputs": ["failure.evidence"],
            "nodes": [
                _node("scope", "Freeze failure evidence", "scope", consumes=["failure.evidence"], produces=["failure.contract"]),
                _node("repair", "Bounded diagnose-test-repair loop", "loop", depends_on=["scope"], consumes=["failure.contract"], produces=["repair.artifact"], max_iterations=3, exit_conditions=["tests_pass", "same_failure_twice", "budget_exhausted"]),
                _node("verify", "Independent regression verification", "verifier", depends_on=["repair"], consumes=["repair.artifact"], produces=["verification.result"], fresh_context=True, verifier_for=["repair"]),
                _node("human", "Human disposition", "human_gate", depends_on=["verify"], consumes=["verification.result"], produces=["human.decision"], authority="HUMAN"),
            ],
            "anchors": [
                {"id": "executed-tests", "type": "test", "nodes": ["verify"], "required": True, "description": "Regression tests run against the repaired artifact."},
                {"id": "human-disposition", "type": "human", "nodes": ["human"], "required": True, "description": "A human owns the final disposition."},
            ],
            "budget": common_budget,
        }
    if sample_id != "protected-release":
        raise GraphContractError(f"unknown sample: {sample_id}")
    return {
        "schema": SCHEMA,
        "graph_id": "protected-release",
        "goal": "Produce an exact-head, independently verified protected release.",
        "external_inputs": ["release.scope"],
        "nodes": [
            _node("scope", "Freeze authority and exact base", "scope", consumes=["release.scope"], produces=["scope.lock"]),
            _node("architecture", "Architecture review", "worker", depends_on=["scope"], consumes=["scope.lock"], produces=["finding.architecture"], writes=["worktree:architecture"]),
            _node("security", "Security review", "worker", depends_on=["scope"], consumes=["scope.lock"], produces=["finding.security"], writes=["worktree:security"]),
            _node("tests", "Test and failure audit", "worker", depends_on=["scope"], consumes=["scope.lock"], produces=["finding.tests"], writes=["worktree:tests"]),
            _node("licensing", "License and provenance review", "worker", depends_on=["scope"], consumes=["scope.lock"], produces=["finding.licensing"], writes=["worktree:licensing"]),
            _node("reduce", "Deterministic finding reduce", "reducer", depends_on=["architecture", "security", "tests", "licensing"], consumes=["finding.architecture", "finding.security", "finding.tests", "finding.licensing"], produces=["review.bundle"]),
            _node("verify", "Fresh-context exact-head verification", "verifier", depends_on=["reduce"], consumes=["review.bundle"], produces=["verification.result"], fresh_context=True, verifier_for=["reduce"]),
            _node("governance", "Policy and protection gate", "governance", depends_on=["verify"], consumes=["verification.result"], produces=["governance.decision"]),
            _node("publish", "Human-approved protected publish", "publisher", depends_on=["governance"], consumes=["governance.decision"], produces=["release.receipt"], writes=["github:protected-main", "huggingface:space"], resources=["release-lane"], side_effecting=True, authority="HUMAN"),
        ],
        "anchors": [
            {"id": "source-head", "type": "source", "nodes": ["scope"], "required": True, "description": "Exact reviewed Git commit and immutable source identity."},
            {"id": "checks", "type": "test", "nodes": ["verify"], "required": True, "description": "Required checks observed on the exact reviewed head."},
            {"id": "human-review", "type": "human", "nodes": ["publish"], "required": True, "description": "Independent authorized approval before publish."},
            {"id": "write-receipt", "type": "receipt", "nodes": ["publish"], "required": True, "description": "Every release write emits a source-bound receipt."},
        ],
        "budget": common_budget,
    }


def status_payload(ns: str = "a11oy") -> dict:
    return {
        "ok": True,
        "service": "a11oy.governed-graph-operations",
        "implementation_status": IMPLEMENTATION_STATUS,
        "evidence_label": EVIDENCE_LABEL,
        "execution_mode": EXECUTION_MODE,
        "schema": SCHEMA,
        "routes": {
            "page": "/graph-operations",
            "status": f"/api/{ns}/v1/graph-operations/status",
            "sample": f"/api/{ns}/v1/graph-operations/sample/{{sample_id}}",
            "analyse": f"/api/{ns}/v1/graph-operations/analyse",
        },
        "samples": ["protected-release", "research-diamond", "bounded-repair"],
        "limits": {"body_bytes": MAX_BODY_BYTES, "nodes": MAX_NODES},
        "truth_boundary": {
            "effectors": 0,
            "writes": 0,
            "provider_calls": 0,
            "receipts_emitted": 0,
            "note": "Real deterministic analyzer; proposed execution remains MODELED.",
        },
        "doctrine": {
            "version": "v11",
            "lambda": "Conjecture 1",
            "khipu_bft": "Conjecture 2",
            "locked_count": 8,
        },
    }


def register(app, ns: str = "a11oy") -> dict:
    """Attach the page and pure-analysis API before the SPA catch-all."""

    from fastapi import Request
    from fastapi.responses import FileResponse, JSONResponse

    page_path = Path(__file__).resolve().parent.parent / "pages" / "graph-operations.html"
    prefix = f"/api/{ns}/v1/graph-operations"

    @app.get("/graph-operations", include_in_schema=False)
    async def graph_operations_page():
        if not page_path.is_file():
            return JSONResponse(
                {
                    "ok": False,
                    "state": "UNAVAILABLE",
                    "reason": "graph operations page is absent from this image",
                },
                status_code=503,
            )
        return FileResponse(
            str(page_path),
            media_type="text/html",
            headers={"cache-control": "no-store"},
        )

    @app.get(prefix + "/status")
    async def graph_operations_status():
        return JSONResponse(status_payload(ns), headers={"cache-control": "no-store"})

    @app.get(prefix + "/sample/{sample_id}")
    async def graph_operations_sample(sample_id: str):
        try:
            contract = sample_contract(sample_id)
        except GraphContractError as exc:
            return JSONResponse({"ok": False, "error": str(exc)}, status_code=404)
        return JSONResponse(
            {"ok": True, "contract": contract, "analysis": analyse_graph(contract)},
            headers={"cache-control": "no-store"},
        )

    @app.post(prefix + "/analyse")
    async def graph_operations_analyse(request: Request):
        content_length = request.headers.get("content-length")
        if content_length:
            try:
                declared_length = int(content_length)
            except ValueError:
                return JSONResponse(
                    {"ok": False, "error": "invalid Content-Length header"},
                    status_code=400,
                )
            if declared_length > MAX_BODY_BYTES:
                return JSONResponse(
                    {"ok": False, "error": "request body exceeds the 96 KiB limit"},
                    status_code=413,
                )
        chunks = []
        body_bytes = 0
        async for chunk in request.stream():
            body_bytes += len(chunk)
            if body_bytes > MAX_BODY_BYTES:
                return JSONResponse(
                    {"ok": False, "error": "request body exceeds the 96 KiB limit"},
                    status_code=413,
                )
            chunks.append(chunk)
        body = b"".join(chunks)
        try:
            payload = json.loads(body.decode("utf-8"))
            result = analyse_graph(payload)
        except (UnicodeDecodeError, json.JSONDecodeError):
            return JSONResponse({"ok": False, "error": "invalid JSON body"}, status_code=400)
        except GraphContractError as exc:
            return JSONResponse({"ok": False, "error": str(exc)}, status_code=422)
        return JSONResponse(result, headers={"cache-control": "no-store"})

    return {
        "ok": True,
        "state": IMPLEMENTATION_STATUS,
        "evidence_label": EVIDENCE_LABEL,
        "routes": [
            "/graph-operations",
            prefix + "/status",
            prefix + "/sample/{sample_id}",
            prefix + "/analyse",
        ],
    }