feat: enforce workflow control, approval, correlation, and provenance invariants

internal/graph/policy_validation_test.go

@@ -0,0 +1,111 @@
+package graph
+
+import (
+	"strings"
+	"testing"
+)
+
+func TestValidateRejectsTypedNodeWithOnlyControlInput(t *testing.T) {
+	g := NewDAG()
+	if err := g.AddNode("input", "tech", "fingerprint", "", 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := g.AddNode("tech", "adaptive", "scanner", "", 2); err != nil {
+		t.Fatal(err)
+	}
+	g.Nodes["tech"].Outputs = []ArtifactType{ArtifactTechnology}
+	g.Nodes["adaptive"].Inputs = []ArtifactType{ArtifactURL}
+	g.Nodes["adaptive"].Outputs = []ArtifactType{ArtifactFinding}
+	for i := range g.Edges {
+		if g.Edges[i].From == "tech" && g.Edges[i].To == "adaptive" {
+			g.Edges[i].Control = true
+			g.Edges[i].Condition = "contains:wordpress"
+		}
+	}
+
+	err := g.Validate()
+	if err == nil || !strings.Contains(err.Error(), "only control edges") {
+		t.Fatalf("expected control-only typed input error, got %v", err)
+	}
+}
+
+func TestValidateAllowsDataAndControlInputsTogether(t *testing.T) {
+	g := NewDAG()
+	if err := g.AddNode("input", "surface", "surface", "", 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := g.AddNode("input", "tech", "fingerprint", "", 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := g.AddNode("surface", "adaptive", "scanner", "", 2); err != nil {
+		t.Fatal(err)
+	}
+	g.Nodes["surface"].Outputs = []ArtifactType{ArtifactURL}
+	g.Nodes["tech"].Outputs = []ArtifactType{ArtifactTechnology}
+	g.Nodes["adaptive"].Inputs = []ArtifactType{ArtifactURL}
+	g.Nodes["adaptive"].Outputs = []ArtifactType{ArtifactFinding}
+	g.Edges = append(g.Edges, Edge{
+		From: "tech", To: "adaptive", Condition: "contains:wordpress",
+		Label: "WordPress detected", Control: true,
+	})
+
+	if err := g.Validate(); err != nil {
+		t.Fatalf("valid data+control adaptive branch rejected: %v", err)
+	}
+}
+
+func TestValidateRejectsUnguardedIntrusiveNode(t *testing.T) {
+	g := NewDAG()
+	if err := g.AddNode("input", "active", "active-tool", "", 1); err != nil {
+		t.Fatal(err)
+	}
+	g.Nodes["active"].Policy = NodePolicy{Intrusive: true, Approval: "active-validation"}
+
+	err := g.Validate()
+	if err == nil || !strings.Contains(err.Error(), "not behind an approval gate") {
+		t.Fatalf("expected approval-boundary error, got %v", err)
+	}
+}
+
+func TestValidateAcceptsMatchingApprovalBoundary(t *testing.T) {
+	g := NewDAG()
+	if err := g.AddNode("input", "approval-gate", "approval", "", 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := g.AddNode("approval-gate", "active", "active-tool", "", 2); err != nil {
+		t.Fatal(err)
+	}
+	gate := g.Nodes["approval-gate"]
+	gate.Kind = NodeKindGate
+	gate.Inputs = []ArtifactType{ArtifactDomain}
+	gate.Outputs = []ArtifactType{ArtifactParameter}
+	gate.Policy = NodePolicy{RequiresApproval: true, Approval: "active-validation"}
+	active := g.Nodes["active"]
+	active.Inputs = []ArtifactType{ArtifactParameter}
+	active.Outputs = []ArtifactType{ArtifactFinding}
+	active.Policy = NodePolicy{Intrusive: true, Approval: "active-validation"}
+
+	if err := g.Validate(); err != nil {
+		t.Fatalf("matching approval boundary rejected: %v", err)
+	}
+}
+
+func TestValidateRejectsMismatchedApprovalBoundary(t *testing.T) {
+	g := NewDAG()
+	if err := g.AddNode("input", "approval-gate", "approval", "", 1); err != nil {
+		t.Fatal(err)
+	}
+	if err := g.AddNode("approval-gate", "active", "active-tool", "", 2); err != nil {
+		t.Fatal(err)
+	}
+	gate := g.Nodes["approval-gate"]
+	gate.Kind = NodeKindGate
+	gate.Policy = NodePolicy{RequiresApproval: true, Approval: "manual-review"}
+	active := g.Nodes["active"]
+	active.Policy = NodePolicy{Intrusive: true, Approval: "active-validation"}
+
+	err := g.Validate()
+	if err == nil || !strings.Contains(err.Error(), "active-validation") {
+		t.Fatalf("expected mismatched approval error, got %v", err)
+	}
+}

internal/graph/semantics.go

@@ -112,14 +112,73 @@ func (g *DAG) Validate() error {
 		if n.Execution.TimeoutSeconds < 0 || n.Execution.Retries < 0 || n.Execution.RetryBackoffMS < 0 {
 			return fmt.Errorf("node %q has negative execution controls", id)
 		}
+		if n.Policy.RequiresApproval && strings.TrimSpace(n.Policy.Approval) == "" {
+			return fmt.Errorf("node %q requires approval but has no explicit approval key", id)
+		}
+
+		dataIncoming, controlIncoming := 0, 0
+		for _, edge := range g.IncomingEdges(id) {
+			if edge.Control {
+				controlIncoming++
+			} else {
+				dataIncoming++
+			}
+		}
+		if controlIncoming > 0 && dataIncoming == 0 && len(n.Inputs) > 0 {
+			return fmt.Errorf("node %q has typed inputs but only control edges; add a data edge for its artifact input", id)
+		}
 	}
 
 	if cycle := g.findCycle(); len(cycle) > 0 {
 		return fmt.Errorf("workflow contains a cycle: %s", strings.Join(cycle, " -> "))
 	}
+	if err := g.validateApprovalBoundaries(); err != nil {
+		return err
+	}
+	return nil
+}
+
+// validateApprovalBoundaries makes the workflow structure prove where active
+// validation is authorized. Runtime flags can approve a boundary, but they do
+// not turn an unguarded intrusive worker into a valid workflow.
+func (g *DAG) validateApprovalBoundaries() error {
+	for id, node := range g.Nodes {
+		if id == g.Root || !node.Policy.Intrusive {
+			continue
+		}
+		key := strings.TrimSpace(node.Policy.Approval)
+		if key == "" {
+			return fmt.Errorf("intrusive node %q must declare an explicit approval key", id)
+		}
+		if !g.hasApprovalGateAncestor(id, key) {
+			return fmt.Errorf("intrusive node %q is not behind an approval gate for %q", id, key)
+		}
+	}
 	return nil
 }
 
+func (g *DAG) hasApprovalGateAncestor(nodeID, approval string) bool {
+	queue := append([]string(nil), g.Parents(nodeID)...)
+	seen := make(map[string]struct{}, len(queue))
+	for len(queue) > 0 {
+		id := queue[0]
+		queue = queue[1:]
+		if _, ok := seen[id]; ok {
+			continue
+		}
+		seen[id] = struct{}{}
+		node := g.Nodes[id]
+		if node == nil {
+			continue
+		}
+		if node.EffectiveKind() == NodeKindGate && node.Policy.RequiresApproval && strings.TrimSpace(node.Policy.Approval) == approval {
+			return true
+		}
+		queue = append(queue, g.Parents(id)...)
+	}
+	return false
+}
+
 // TopologicalOrder returns deterministic dependency order independent of the
 // visual matrix. It is useful for validation, reporting and schedulers.
 func (g *DAG) TopologicalOrder() ([]string, error) {