Artifex/tests/test_graph_pause_subgraph_inspection.py

from __future__ import annotations

from graph.inspection import graph_run_inspection
from graph.models import ExecutionGraphDefinition, ExecutionGraphVersion, ExecutionGraphVersionStatus, GraphApproval, GraphApprovalStatus, GraphRun, GraphRunStatus
from graph.native_runtime import GraphExecutionContext, NativeGraphRuntime
from graph.registry import NodeHandlerRegistry, NodeResult
from graph.spec import ExecutionGraphSpec, GraphEdgeSpec, GraphNodeSpec


def persist_spec(spec: ExecutionGraphSpec) -> GraphRun:
    definition = ExecutionGraphDefinition.objects.create(name=spec.name, graph_type=spec.graph_type)
    version = ExecutionGraphVersion.objects.create(graph=definition, version=spec.version, status=ExecutionGraphVersionStatus.CHAMPION, graph_spec=spec.to_dict())
    return GraphRun.objects.create(execution_graph_version=version, current_node=spec.entry)


class ApprovalGate:
    node_type = "approval_gate"
    idempotent = True
    replay_safe = "checkpointed"
    destructive = False

    def run(self, context: GraphExecutionContext) -> NodeResult:
        node_run = context.graph_run.node_runs.filter(node_id=context.graph_run.current_node).order_by("-visit_index").first()
        if GraphApproval.objects.filter(graph_run=context.graph_run, status=GraphApprovalStatus.APPROVED).exists():
            return NodeResult("COMPLETE", "approved")
        GraphApproval.objects.get_or_create(graph_run=context.graph_run, node_run=node_run, reason="AWAITING_APPROVAL")
        return NodeResult("PAUSED", "awaiting", pause_reason="AWAITING_APPROVAL")


class DoneNode:
    node_type = "done_node"
    idempotent = True
    replay_safe = True
    destructive = False

    def run(self, context: GraphExecutionContext) -> NodeResult:
        return NodeResult("COMPLETE", "success")


def test_human_approval_pause_signal_resume() -> None:
    spec = ExecutionGraphSpec(
        name="approval_fixture",
        version=1,
        graph_type="FIXTURE",
        entry="approval",
        nodes={"approval": GraphNodeSpec("approval", "approval_gate"), "done": GraphNodeSpec("done", "done_node")},
        edges=[GraphEdgeSpec("approval", "done", "approved")],
        terminal_nodes=["done"],
    )
    graph_run = persist_spec(spec)
    registry = NodeHandlerRegistry()
    registry.register(ApprovalGate())
    registry.register(DoneNode())
    runtime = NativeGraphRuntime(registry)

    runtime.run_until_terminal_or_paused(graph_run)
    graph_run.refresh_from_db()
    assert graph_run.status == GraphRunStatus.PAUSED
    assert graph_run.approvals.get().status == GraphApprovalStatus.PENDING

    runtime.signal_now(str(graph_run.id), {"action": "approve", "actor": "tester"})
    runtime.run_until_terminal_or_paused(graph_run)
    graph_run.refresh_from_db()
    assert graph_run.status == GraphRunStatus.COMPLETE
    assert graph_run.approvals.get().status == GraphApprovalStatus.APPROVED


def test_subgraph_can_be_represented_in_runtime_neutral_spec() -> None:
    spec = ExecutionGraphSpec(
        name="subgraph_fixture",
        version=1,
        graph_type="FIXTURE",
        entry="parent",
        nodes={
            "parent": GraphNodeSpec("parent", "subgraph", {"subgraph": {"name": "child_graph", "version": 1}}),
            "done": GraphNodeSpec("done", "done_node"),
        },
        edges=[GraphEdgeSpec("parent", "done", "success")],
        terminal_nodes=["done"],
    )

    restored = ExecutionGraphSpec.from_dict(spec.to_dict())

    assert restored.nodes["parent"].metadata["subgraph"]["name"] == "child_graph"


def test_graph_run_inspection_exposes_ui_ready_shape() -> None:
    spec = ExecutionGraphSpec(
        name="inspect_fixture",
        version=1,
        graph_type="FIXTURE",
        entry="done",
        nodes={"done": GraphNodeSpec("done", "done_node")},
        edges=[],
        terminal_nodes=["done"],
    )
    graph_run = persist_spec(spec)
    inspection = graph_run_inspection(graph_run)

    assert inspection["graph"] == "inspect_fixture"
    assert inspection["current_node"] == "done"
    assert inspection["version_status"] == ExecutionGraphVersionStatus.CHAMPION
    assert inspection["final_failure_reason"] is None
    assert inspection["historical_failures"] == []
    assert inspection["nodes"] == [{"id": "done", "type": "done_node", "status": "PENDING", "visit_index": 0, "duration_ms": None, "failure": {}, "metadata": {}}]


def test_graph_run_inspection_separates_final_and_historical_failures() -> None:
    spec = ExecutionGraphSpec(
        name="inspect_failures_fixture",
        version=1,
        graph_type="FIXTURE",
        entry="done",
        nodes={"done": GraphNodeSpec("done", "done_node")},
        edges=[],
        terminal_nodes=["done"],
    )
    graph_run = persist_spec(spec)
    graph_run.metadata = {
        "final_failure_reason": None,
        "historical_failures": [{"node_id": "review", "reason": "review_failed", "evidence": [{"type": "missing_route"}]}],
    }
    graph_run.save(update_fields=["metadata", "updated_at"])

    inspection = graph_run_inspection(graph_run)

    assert inspection["final_failure_reason"] is None
    assert inspection["historical_failures"] == [{"node_id": "review", "reason": "review_failed", "evidence": [{"type": "missing_route"}]}]