diff --git a/graph/inspection.py b/graph/inspection.py new file mode 100644 index 0000000..3a16827 --- /dev/null +++ b/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")), + } diff --git a/graph/migrations/0003_graphapproval.py b/graph/migrations/0003_graphapproval.py new file mode 100644 index 0000000..ab7c2a4 --- /dev/null +++ b/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")), + ], + ), + ] diff --git a/graph/models.py b/graph/models.py index e1212ed..327fdb0 100644 --- a/graph/models.py +++ b/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) diff --git a/graph/native_runtime.py b/graph/native_runtime.py index 8d68452..ef7d79c 100644 --- a/graph/native_runtime.py +++ b/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: diff --git a/tests/test_graph_pause_subgraph_inspection.py b/tests/test_graph_pause_subgraph_inspection.py new file mode 100644 index 0000000..174e3bb --- /dev/null +++ b/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": {}}]