Add graph pause and inspection support

graph/inspection.py

@@ -0,0 +1,37 @@
+from __future__ import annotations
+
+from graph.models import GraphRun
+
+
+def graph_run_inspection(graph_run: GraphRun) -> dict[str, object]:
+    spec = graph_run.execution_graph_version.graph_spec
+    node_runs = {
+        node.node_id: node
+        for node in graph_run.node_runs.order_by("node_id", "-visit_index")
+    }
+    nodes = []
+    for node_id, node_spec in spec.get("nodes", {}).items():
+        run = node_runs.get(node_id)
+        nodes.append(
+            {
+                "id": node_id,
+                "type": node_spec.get("type"),
+                "status": run.status if run else "PENDING",
+                "visit_index": run.visit_index if run else 0,
+                "duration_ms": (run.telemetry or {}).get("duration_ms") if run else None,
+                "failure": run.failure_evidence if run else {},
+                "metadata": node_spec.get("metadata", {}),
+            }
+        )
+    return {
+        "graph": graph_run.execution_graph_version.graph.name,
+        "version": graph_run.execution_graph_version.version,
+        "status": graph_run.status,
+        "current_node": graph_run.current_node,
+        "nodes": nodes,
+        "edges": spec.get("edges", []),
+        "edge_traversals": list(
+            graph_run.edge_traversals.order_by("created_at").values("source_node", "target_node", "condition", "result", "metadata", "created_at")
+        ),
+        "failures": list(graph_run.node_runs.exclude(failure_evidence={}).values("node_id", "failure_evidence")),
+    }

graph/migrations/0003_graphapproval.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("graph", "0002_graphnoderun_visit_index"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="GraphApproval",
+            fields=[
+                ("id", models.BigAutoField(auto_created=True, primary_key=True, serialize=False, verbose_name="ID")),
+                ("status", models.CharField(choices=[("PENDING", "Pending"), ("APPROVED", "Approved"), ("REJECTED", "Rejected")], default="PENDING", max_length=32)),
+                ("reason", models.CharField(max_length=160)),
+                ("payload", models.JSONField(blank=True, default=dict)),
+                ("requested_by", models.CharField(default="graph_runtime", max_length=120)),
+                ("decided_by", models.CharField(blank=True, max_length=120)),
+                ("decided_at", models.DateTimeField(blank=True, null=True)),
+                ("created_at", models.DateTimeField(auto_now_add=True)),
+                ("updated_at", models.DateTimeField(auto_now=True)),
+                ("graph_run", models.ForeignKey(on_delete=django.db.models.deletion.CASCADE, related_name="approvals", to="graph.graphrun")),
+                ("node_run", models.ForeignKey(blank=True, null=True, on_delete=django.db.models.deletion.CASCADE, related_name="approvals", to="graph.graphnoderun")),
+            ],
+        ),
+    ]

graph/models.py

@@ -108,3 +108,22 @@ class GraphEdgeTraversal(models.Model):
 
     class Meta:
         indexes = [models.Index(fields=["graph_run", "created_at"])]
+
+
+class GraphApprovalStatus(models.TextChoices):
+    PENDING = "PENDING"
+    APPROVED = "APPROVED"
+    REJECTED = "REJECTED"
+
+
+class GraphApproval(models.Model):
+    graph_run = models.ForeignKey(GraphRun, on_delete=models.CASCADE, related_name="approvals")
+    node_run = models.ForeignKey(GraphNodeRun, on_delete=models.CASCADE, null=True, blank=True, related_name="approvals")
+    status = models.CharField(max_length=32, choices=GraphApprovalStatus.choices, default=GraphApprovalStatus.PENDING)
+    reason = models.CharField(max_length=160)
+    payload = models.JSONField(default=dict, blank=True)
+    requested_by = models.CharField(max_length=120, default="graph_runtime")
+    decided_by = models.CharField(max_length=120, blank=True)
+    decided_at = models.DateTimeField(null=True, blank=True)
+    created_at = models.DateTimeField(auto_now_add=True)
+    updated_at = models.DateTimeField(auto_now=True)

graph/native_runtime.py

@@ -5,10 +5,11 @@ from dataclasses import dataclass, field
 from typing import Any
 from uuid import UUID
 
+from asgiref.sync import sync_to_async
 from django.utils import timezone
 
 from control_plane.events.bus import EventBus
-from graph.models import GraphEdgeTraversal, GraphNodeRun, GraphNodeRunStatus, GraphRun, GraphRunStatus
+from graph.models import GraphApprovalStatus, GraphEdgeTraversal, GraphNodeRun, GraphNodeRunStatus, GraphRun, GraphRunStatus
 from graph.registry import NodeHandlerRegistry, NodeResult
 from graph.runtime import GraphRuntime
 from graph.spec import ExecutionGraphSpec, outgoing_edges
@@ -60,16 +61,25 @@ class NativeGraphRuntime(GraphRuntime):
         self.bus.publish("GRAPH_RUN_CANCELLED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id})
 
     async def signal(self, run_id: str, event: dict[str, Any]) -> None:
+        await sync_to_async(self.signal_now)(run_id, event)
+
+    def signal_now(self, run_id: str, event: dict[str, Any]) -> None:
         graph_run = GraphRun.objects.get(id=run_id)
         metadata = dict(graph_run.metadata)
         metadata["last_signal"] = self._bounded(event)
         if event.get("action") == "approve" and graph_run.status == GraphRunStatus.PAUSED:
+            for approval in graph_run.approvals.filter(status=GraphApprovalStatus.PENDING):
+                approval.status = GraphApprovalStatus.APPROVED
+                approval.decided_by = str(event.get("actor", "system"))
+                approval.decided_at = timezone.now()
+                approval.save(update_fields=["status", "decided_by", "decided_at", "updated_at"])
             graph_run.status = GraphRunStatus.RUNNING
         graph_run.metadata = metadata
         graph_run.save(update_fields=["metadata", "status", "updated_at"])
         self.bus.publish("GRAPH_RUN_SIGNALED", project=graph_run.project, task=graph_run.task, payload={"graph_run_id": graph_run.id, "signal": self._bounded(event)})
 
     def run_until_terminal_or_paused(self, graph_run: GraphRun, *, interrupt_after: str | None = None) -> GraphRun:
+        graph_run.refresh_from_db()
         spec = ExecutionGraphSpec.from_dict(graph_run.execution_graph_version.graph_spec)
         spec.validate()
         if graph_run.status == GraphRunStatus.PENDING:

tests/test_graph_pause_subgraph_inspection.py

@@ -0,0 +1,102 @@
+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["nodes"] == [{"id": "done", "type": "done_node", "status": "PENDING", "visit_index": 0, "duration_ms": None, "failure": {}, "metadata": {}}]