From bd75819afb64de0edcadb410f0e362247f7b24f8 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:20:41 +0200
Subject: [PATCH 01/15] feat(pipeline): persist artifact provenance and control
 decisions

---
 internal/pipeline/provenance.go | 194 ++++++++++++++++++++++++++++++++
 1 file changed, 194 insertions(+)
 create mode 100644 internal/pipeline/provenance.go

diff --git a/internal/pipeline/provenance.go b/internal/pipeline/provenance.go
new file mode 100644
index 0000000..c641ec5
--- /dev/null
+++ b/internal/pipeline/provenance.go
@@ -0,0 +1,194 @@
+package pipeline
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"time"
+
+	"github.com/MKlolbullen/termaid/internal/graph"
+)
+
+// ArtifactRef is an immutable description of a file that participated in a
+// workflow step. Hashes make later evidence/reporting code able to prove which
+// exact bytes were consumed or produced without copying the raw artifact.
+type ArtifactRef struct {
+	Path      string `json:"path"`
+	Role      string `json:"role"`
+	SHA256    string `json:"sha256,omitempty"`
+	Size      int64  `json:"size"`
+	LineCount int    `json:"line_count"`
+	Format    string `json:"format,omitempty"`
+}
+
+// ControlDecision records why a control-only dependency allowed or pruned a
+// branch. Control edges affect dispatch but never become data input.
+type ControlDecision struct {
+	From        string     `json:"from"`
+	To          string     `json:"to"`
+	Condition   string     `json:"condition,omitempty"`
+	Label       string     `json:"label,omitempty"`
+	Matched     bool       `json:"matched"`
+	ParentState NodeStatus `json:"parent_state"`
+}
+
+// ApprovalDecision records the authorization state used when dispatching a
+// node. This is deliberately persisted with the run rather than inferred later.
+type ApprovalDecision struct {
+	Key                string `json:"key,omitempty"`
+	RequiresApproval   bool   `json:"requires_approval"`
+	Intrusive          bool   `json:"intrusive"`
+	Approved           bool   `json:"approved"`
+	BroadIntrusiveMode bool   `json:"broad_intrusive_mode"`
+}
+
+// NodeProvenance links a node to exact input/output artifacts, data parents,
+// control-edge decisions, and approval state.
+type NodeProvenance struct {
+	RecordedAt       time.Time         `json:"recorded_at"`
+	Parents          []string          `json:"parents,omitempty"`
+	DataParents      []string          `json:"data_parents,omitempty"`
+	InputArtifacts   []ArtifactRef     `json:"input_artifacts,omitempty"`
+	PreparedInput    *ArtifactRef      `json:"prepared_input,omitempty"`
+	OutputArtifacts  []ArtifactRef     `json:"output_artifacts,omitempty"`
+	ControlDecisions []ControlDecision `json:"control_decisions,omitempty"`
+	Approval         ApprovalDecision  `json:"approval"`
+}
+
+func attachNodeProvenance(df *DataFlow, node *graph.Node, dag *graph.DAG, cfg RunConfig, dataParents []string, preparedInput string) error {
+	if df == nil || node == nil || dag == nil {
+		return fmt.Errorf("cannot attach provenance to nil runtime object")
+	}
+	output := df.NodeOutputs[node.ID]
+	if output == nil {
+		return fmt.Errorf("node %s has no recorded output for provenance", node.ID)
+	}
+
+	parents := dag.Parents(node.ID)
+	inputPaths := append([]string(nil), df.GlobalState.DataLinks[node.ID]...)
+	if len(inputPaths) == 0 {
+		for _, parentID := range dataParents {
+			inputPaths = append(inputPaths, outputFiles(df, parentID)...)
+		}
+	}
+	inputPaths = uniqueSortedStrings(inputPaths)
+
+	prov := &NodeProvenance{
+		RecordedAt:  time.Now().UTC(),
+		Parents:     parents,
+		DataParents: append([]string(nil), dataParents...),
+		Approval:    approvalDecision(node, cfg),
+	}
+	sort.Strings(prov.DataParents)
+
+	for _, path := range inputPaths {
+		ref, err := artifactRef(df, path, "raw-input")
+		if err != nil {
+			continue
+		}
+		prov.InputArtifacts = append(prov.InputArtifacts, ref)
+	}
+	if strings.TrimSpace(preparedInput) != "" {
+		if ref, err := artifactRef(df, preparedInput, "prepared-input"); err == nil {
+			prov.PreparedInput = &ref
+		}
+	}
+	for _, path := range output.OutputFiles {
+		ref, err := artifactRef(df, path, "output")
+		if err != nil {
+			continue
+		}
+		prov.OutputArtifacts = append(prov.OutputArtifacts, ref)
+	}
+	for _, edge := range dag.IncomingEdges(node.ID) {
+		if !edge.Control {
+			continue
+		}
+		state := df.GlobalState.NodeStates[edge.From]
+		matched := false
+		if state == NodeCompleted {
+			matched = conditionMatches(edge.Condition, outputFiles(df, edge.From), df, cfg)
+		}
+		prov.ControlDecisions = append(prov.ControlDecisions, ControlDecision{
+			From: edge.From, To: edge.To, Condition: edge.Condition, Label: edge.Label,
+			Matched: matched, ParentState: state,
+		})
+	}
+	sort.Slice(prov.ControlDecisions, func(i, j int) bool {
+		return prov.ControlDecisions[i].From < prov.ControlDecisions[j].From
+	})
+
+	output.Provenance = prov
+	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-provenance.json", node.ID))
+	data, err := json.MarshalIndent(prov, "", "  ")
+	if err != nil {
+		return fmt.Errorf("marshal provenance for %s: %w", node.ID, err)
+	}
+	if err := os.WriteFile(path, data, 0o600); err != nil {
+		return fmt.Errorf("write provenance for %s: %w", node.ID, err)
+	}
+	output.ProvenanceFile = path
+	return nil
+}
+
+func approvalDecision(node *graph.Node, cfg RunConfig) ApprovalDecision {
+	key := strings.TrimSpace(node.Policy.Approval)
+	if key == "" && node.Policy.RequiresApproval {
+		key = node.ID
+	}
+	approved := false
+	if key != "" {
+		approved = cfg.Approvals[key] || cfg.Approvals[node.ID]
+	}
+	if node.Policy.Intrusive && (cfg.AllowIntrusive || cfg.Approvals["intrusive"]) {
+		approved = true
+	}
+	return ApprovalDecision{
+		Key: key, RequiresApproval: node.Policy.RequiresApproval, Intrusive: node.Policy.Intrusive,
+		Approved: approved, BroadIntrusiveMode: cfg.AllowIntrusive || cfg.Approvals["intrusive"],
+	}
+}
+
+func artifactRef(df *DataFlow, path, role string) (ArtifactRef, error) {
+	stat, err := os.Stat(path)
+	if err != nil {
+		return ArtifactRef{}, err
+	}
+	file, err := os.Open(path)
+	if err != nil {
+		return ArtifactRef{}, err
+	}
+	defer file.Close()
+	h := sha256.New()
+	if _, err := io.Copy(h, file); err != nil {
+		return ArtifactRef{}, err
+	}
+	lines, _ := df.countLines(path)
+	return ArtifactRef{
+		Path: path, Role: role, SHA256: hex.EncodeToString(h.Sum(nil)), Size: stat.Size(),
+		LineCount: lines, Format: df.detectFormat(path),
+	}, nil
+}
+
+func uniqueSortedStrings(items []string) []string {
+	seen := make(map[string]struct{}, len(items))
+	out := make([]string, 0, len(items))
+	for _, item := range items {
+		if item == "" {
+			continue
+		}
+		if _, ok := seen[item]; ok {
+			continue
+		}
+		seen[item] = struct{}{}
+		out = append(out, item)
+	}
+	sort.Strings(out)
+	return out
+}

From b08131edd9110002917ee277529e8f03a20a9890 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:21:12 +0200
Subject: [PATCH 02/15] feat(pipeline): preserve correlation groups and report
 lineage

---
 internal/pipeline/correlation.go | 200 +++++++++++++++++++++++++++----
 1 file changed, 174 insertions(+), 26 deletions(-)

diff --git a/internal/pipeline/correlation.go b/internal/pipeline/correlation.go
index 11363d9..a1e74f4 100644
--- a/internal/pipeline/correlation.go
+++ b/internal/pipeline/correlation.go
@@ -3,8 +3,16 @@ package pipeline
 import (
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
 	"sort"
+	"strconv"
 	"strings"
+	"time"
+
+	"github.com/MKlolbullen/termaid/internal/graph"
 )
 
 // CorrelationKey groups tool observations that describe the same underlying
@@ -31,54 +39,194 @@ func CorrelationKey(record DataRecord) string {
 	return hex.EncodeToString(sum[:16])
 }
 
-// CorrelateRecords de-duplicates observations while retaining provenance from
-// all contributing tools in metadata.sources. The highest-confidence record is
-// used as the representative observation.
-func CorrelateRecords(records []DataRecord) []DataRecord {
-	type group struct {
-		best    DataRecord
-		sources map[string]struct{}
+// CorrelationGroup preserves every raw observation contributing to one logical
+// finding/evidence item while still selecting a representative record for
+// reporting. This prevents de-duplication from destroying source evidence.
+type CorrelationGroup struct {
+	Key            string       `json:"key"`
+	Representative DataRecord   `json:"representative"`
+	Sources        []string     `json:"sources"`
+	Observations   []DataRecord `json:"observations"`
+}
+
+// CorrelateGroups returns stable correlation groups with complete observations.
+func CorrelateGroups(records []DataRecord) []CorrelationGroup {
+	type mutableGroup struct {
+		best         DataRecord
+		observations []DataRecord
+		sources      map[string]struct{}
 	}
-	groups := make(map[string]*group)
+	groups := make(map[string]*mutableGroup)
 	for _, record := range records {
+		if record.Metadata == nil {
+			record.Metadata = make(map[string]string)
+		}
 		key := CorrelationKey(record)
 		g, ok := groups[key]
 		if !ok {
-			copyRecord := record
-			if copyRecord.Metadata == nil {
-				copyRecord.Metadata = make(map[string]string)
-			}
-			g = &group{best: copyRecord, sources: make(map[string]struct{})}
+			g = &mutableGroup{best: cloneRecord(record), sources: make(map[string]struct{})}
 			groups[key] = g
 		}
+		g.observations = append(g.observations, cloneRecord(record))
 		if record.Source != "" {
 			g.sources[record.Source] = struct{}{}
 		}
 		if record.Confidence > g.best.Confidence {
-			meta := g.best.Metadata
-			g.best = record
-			if g.best.Metadata == nil {
-				g.best.Metadata = meta
-			}
+			g.best = cloneRecord(record)
 		}
 	}
 
-	out := make([]DataRecord, 0, len(groups))
+	out := make([]CorrelationGroup, 0, len(groups))
 	for key, g := range groups {
-		if g.best.Metadata == nil {
-			g.best.Metadata = make(map[string]string)
-		}
 		var sources []string
 		for source := range g.sources {
 			sources = append(sources, source)
 		}
 		sort.Strings(sources)
+		sort.SliceStable(g.observations, func(i, j int) bool {
+			if g.observations[i].Source != g.observations[j].Source {
+				return g.observations[i].Source < g.observations[j].Source
+			}
+			return g.observations[i].Value < g.observations[j].Value
+		})
+		if g.best.Metadata == nil {
+			g.best.Metadata = make(map[string]string)
+		}
 		g.best.Metadata["correlation_key"] = key
 		g.best.Metadata["sources"] = strings.Join(sources, ",")
-		out = append(out, g.best)
+		g.best.Metadata["observation_count"] = strconv.Itoa(len(g.observations))
+		out = append(out, CorrelationGroup{
+			Key: key, Representative: g.best, Sources: sources, Observations: g.observations,
+		})
+	}
+	sort.Slice(out, func(i, j int) bool { return out[i].Key < out[j].Key })
+	return out
+}
+
+// CorrelateRecords is the compact compatibility view used by existing callers.
+func CorrelateRecords(records []DataRecord) []DataRecord {
+	groups := CorrelateGroups(records)
+	out := make([]DataRecord, 0, len(groups))
+	for _, group := range groups {
+		out = append(out, group.Representative)
 	}
-	sort.Slice(out, func(i, j int) bool {
-		return CorrelationKey(out[i]) < CorrelationKey(out[j])
-	})
 	return out
 }
+
+func cloneRecord(record DataRecord) DataRecord {
+	copyRecord := record
+	if record.Metadata != nil {
+		copyRecord.Metadata = make(map[string]string, len(record.Metadata))
+		for k, v := range record.Metadata {
+			copyRecord.Metadata[k] = v
+		}
+	}
+	return copyRecord
+}
+
+// writeCorrelationSnapshot correlates a merge node from its raw parent outputs,
+// not from the already-normalized merged file. That retains source identity and
+// gives report generation a candidate/evidence snapshot created before the sink.
+func writeCorrelationSnapshot(df *DataFlow, node *graph.Node, parents []string) error {
+	output := df.NodeOutputs[node.ID]
+	if output == nil {
+		return fmt.Errorf("node %s has no recorded output for correlation", node.ID)
+	}
+	var records []DataRecord
+	for _, parentID := range parents {
+		for _, file := range outputFiles(df, parentID) {
+			parsed, err := df.parseFile(file, parentID)
+			if err != nil {
+				continue
+			}
+			records = append(records, parsed...)
+		}
+	}
+	groups := CorrelateGroups(records)
+	payload := struct {
+		NodeID      string             `json:"node_id"`
+		Artifact    string             `json:"artifact"`
+		GeneratedAt time.Time          `json:"generated_at"`
+		Groups      []CorrelationGroup `json:"groups"`
+	}{
+		NodeID: node.ID, Artifact: artifactTypes(node.Outputs), GeneratedAt: time.Now().UTC(), Groups: groups,
+	}
+	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-correlation.json", node.ID))
+	data, err := json.MarshalIndent(payload, "", "  ")
+	if err != nil {
+		return err
+	}
+	if err := os.WriteFile(path, data, 0o600); err != nil {
+		return err
+	}
+	if output.Metadata == nil {
+		output.Metadata = make(map[string]string)
+	}
+	output.Metadata["correlation_file"] = path
+	output.Metadata["correlation_groups"] = strconv.Itoa(len(groups))
+	return nil
+}
+
+// CorrelationReport is the report sink's structured output. Candidate findings
+// and verification evidence stay separate lifecycle stages while each retains
+// its raw observations.
+type CorrelationReport struct {
+	Version     string             `json:"version"`
+	GeneratedAt time.Time          `json:"generated_at"`
+	Candidates  []CorrelationGroup `json:"candidates"`
+	Evidence    []CorrelationGroup `json:"evidence"`
+}
+
+func writeCorrelationReport(df *DataFlow, node *graph.Node, fallbackInput string) (string, error) {
+	var candidateRecords []DataRecord
+	var evidenceRecords []DataRecord
+	for _, output := range df.NodeOutputs {
+		path := output.Metadata["correlation_file"]
+		if path == "" {
+			continue
+		}
+		groups, err := readCorrelationGroups(path)
+		if err != nil {
+			continue
+		}
+		for _, group := range groups {
+			if strings.Contains(output.Metadata["outputs"], string(graph.ArtifactFinding)) {
+				candidateRecords = append(candidateRecords, group.Observations...)
+			}
+			if strings.Contains(output.Metadata["outputs"], string(graph.ArtifactEvidence)) {
+				evidenceRecords = append(evidenceRecords, group.Observations...)
+			}
+		}
+	}
+	if len(evidenceRecords) == 0 && fallbackInput != "" {
+		records, err := df.parseFile(fallbackInput, node.ID)
+		if err == nil {
+			evidenceRecords = append(evidenceRecords, records...)
+		}
+	}
+
+	report := CorrelationReport{
+		Version: "1.0", GeneratedAt: time.Now().UTC(),
+		Candidates: CorrelateGroups(candidateRecords), Evidence: CorrelateGroups(evidenceRecords),
+	}
+	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-correlated-report.json", node.ID))
+	data, err := json.MarshalIndent(report, "", "  ")
+	if err != nil {
+		return "", err
+	}
+	return path, os.WriteFile(path, data, 0o600)
+}
+
+func readCorrelationGroups(path string) ([]CorrelationGroup, error) {
+	data, err := os.ReadFile(path)
+	if err != nil {
+		return nil, err
+	}
+	var payload struct {
+		Groups []CorrelationGroup `json:"groups"`
+	}
+	if err := json.Unmarshal(data, &payload); err != nil {
+		return nil, err
+	}
+	return payload.Groups, nil
+}

From 451eba64c89acee6dcb233d9b015ea3de5b13d02 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:21:33 +0200
Subject: [PATCH 03/15] fix(pipeline): persist provenance via existing node
 metadata

---
 internal/pipeline/provenance.go | 7 +++++--
 1 file changed, 5 insertions(+), 2 deletions(-)

diff --git a/internal/pipeline/provenance.go b/internal/pipeline/provenance.go
index c641ec5..2fa8a38 100644
--- a/internal/pipeline/provenance.go
+++ b/internal/pipeline/provenance.go
@@ -124,7 +124,6 @@ func attachNodeProvenance(df *DataFlow, node *graph.Node, dag *graph.DAG, cfg Ru
 		return prov.ControlDecisions[i].From < prov.ControlDecisions[j].From
 	})
 
-	output.Provenance = prov
 	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-provenance.json", node.ID))
 	data, err := json.MarshalIndent(prov, "", "  ")
 	if err != nil {
@@ -133,7 +132,11 @@ func attachNodeProvenance(df *DataFlow, node *graph.Node, dag *graph.DAG, cfg Ru
 	if err := os.WriteFile(path, data, 0o600); err != nil {
 		return fmt.Errorf("write provenance for %s: %w", node.ID, err)
 	}
-	output.ProvenanceFile = path
+	if output.Metadata == nil {
+		output.Metadata = make(map[string]string)
+	}
+	output.Metadata["provenance_file"] = path
+	output.Metadata["provenance_recorded_at"] = prov.RecordedAt.Format(time.RFC3339Nano)
 	return nil
 }
 

From 8afe877e681bb2f005fb42e2f5d73c0a6c6fe35b Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:21:59 +0200
Subject: [PATCH 04/15] feat(graph): enforce control inputs and approval
 boundaries

---
 internal/graph/semantics.go | 60 ++++++++++++++++++++++++++++++++++++-
 1 file changed, 59 insertions(+), 1 deletion(-)

diff --git a/internal/graph/semantics.go b/internal/graph/semantics.go
index 905ac6c..1e88643 100644
--- a/internal/graph/semantics.go
+++ b/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) {
@@ -150,7 +209,6 @@ func (g *DAG) TopologicalOrder() ([]string, error) {
 				ready = append(ready, e.To)
 				sort.Strings(ready)
 			}
-		}
 	}
 	if len(order) != len(g.Nodes) {
 		return nil, fmt.Errorf("workflow is not acyclic")

From 4f7dce0b1b0f9ca69bf5b51204c797e0c93a7765 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:24:22 +0200
Subject: [PATCH 05/15] feat(pipeline): persist immutable artifact provenance

---
 internal/pipeline/provenance.go | 158 +++++++++++---------------------
 1 file changed, 51 insertions(+), 107 deletions(-)

diff --git a/internal/pipeline/provenance.go b/internal/pipeline/provenance.go
index 2fa8a38..b96a2c5 100644
--- a/internal/pipeline/provenance.go
+++ b/internal/pipeline/provenance.go
@@ -9,128 +9,76 @@ import (
 	"os"
 	"path/filepath"
 	"sort"
-	"strings"
 	"time"
-
-	"github.com/MKlolbullen/termaid/internal/graph"
 )
 
 // ArtifactRef is an immutable description of a file that participated in a
-// workflow step. Hashes make later evidence/reporting code able to prove which
-// exact bytes were consumed or produced without copying the raw artifact.
+// workflow step. Hashes let later evidence/reporting code prove which exact
+// bytes were consumed or produced without replacing the original artifact.
 type ArtifactRef struct {
-	Path      string `json:"path"`
-	Role      string `json:"role"`
-	SHA256    string `json:"sha256,omitempty"`
-	Size      int64  `json:"size"`
-	LineCount int    `json:"line_count"`
-	Format    string `json:"format,omitempty"`
-}
-
-// ControlDecision records why a control-only dependency allowed or pruned a
-// branch. Control edges affect dispatch but never become data input.
-type ControlDecision struct {
-	From        string     `json:"from"`
-	To          string     `json:"to"`
-	Condition   string     `json:"condition,omitempty"`
-	Label       string     `json:"label,omitempty"`
-	Matched     bool       `json:"matched"`
-	ParentState NodeStatus `json:"parent_state"`
+	Path       string `json:"path"`
+	Role       string `json:"role"`
+	SourceNode string `json:"source_node,omitempty"`
+	SHA256     string `json:"sha256,omitempty"`
+	Size       int64  `json:"size"`
+	LineCount  int    `json:"line_count"`
+	Format     string `json:"format,omitempty"`
 }
 
-// ApprovalDecision records the authorization state used when dispatching a
-// node. This is deliberately persisted with the run rather than inferred later.
-type ApprovalDecision struct {
-	Key                string `json:"key,omitempty"`
-	RequiresApproval   bool   `json:"requires_approval"`
-	Intrusive          bool   `json:"intrusive"`
-	Approved           bool   `json:"approved"`
-	BroadIntrusiveMode bool   `json:"broad_intrusive_mode"`
-}
-
-// NodeProvenance links a node to exact input/output artifacts, data parents,
-// control-edge decisions, and approval state.
+// NodeProvenance links one execution result to its exact file inputs and
+// outputs. Graph-level parent/control/approval metadata remains in NodeOutput
+// metadata and the checkpoint; this file preserves the byte-level evidence.
 type NodeProvenance struct {
-	RecordedAt       time.Time         `json:"recorded_at"`
-	Parents          []string          `json:"parents,omitempty"`
-	DataParents      []string          `json:"data_parents,omitempty"`
-	InputArtifacts   []ArtifactRef     `json:"input_artifacts,omitempty"`
-	PreparedInput    *ArtifactRef      `json:"prepared_input,omitempty"`
-	OutputArtifacts  []ArtifactRef     `json:"output_artifacts,omitempty"`
-	ControlDecisions []ControlDecision `json:"control_decisions,omitempty"`
-	Approval         ApprovalDecision  `json:"approval"`
+	NodeID          string        `json:"node_id"`
+	Tool            string        `json:"tool"`
+	RecordedAt      time.Time     `json:"recorded_at"`
+	InputArtifacts  []ArtifactRef `json:"input_artifacts,omitempty"`
+	OutputArtifacts []ArtifactRef `json:"output_artifacts,omitempty"`
 }
 
-func attachNodeProvenance(df *DataFlow, node *graph.Node, dag *graph.DAG, cfg RunConfig, dataParents []string, preparedInput string) error {
-	if df == nil || node == nil || dag == nil {
-		return fmt.Errorf("cannot attach provenance to nil runtime object")
+// persistNodeProvenance writes a sidecar under analysis/ and links it from the
+// NodeOutput metadata. Raw files themselves are never rewritten or deleted.
+func persistNodeProvenance(df *DataFlow, output *NodeOutput) error {
+	if df == nil || output == nil {
+		return fmt.Errorf("cannot persist provenance for nil output")
 	}
-	output := df.NodeOutputs[node.ID]
-	if output == nil {
-		return fmt.Errorf("node %s has no recorded output for provenance", node.ID)
+	prov := NodeProvenance{
+		NodeID: output.NodeID, Tool: output.Tool, RecordedAt: time.Now().UTC(),
 	}
 
-	parents := dag.Parents(node.ID)
-	inputPaths := append([]string(nil), df.GlobalState.DataLinks[node.ID]...)
-	if len(inputPaths) == 0 {
-		for _, parentID := range dataParents {
-			inputPaths = append(inputPaths, outputFiles(df, parentID)...)
+	inputPaths := append([]string(nil), df.GlobalState.DataLinks[output.NodeID]...)
+	// Root-level nodes receive the seed directly and historically had no
+	// DataLinks entry. Preserve that seed relationship explicitly.
+	if len(inputPaths) == 0 && output.NodeID != "seed" {
+		if seed := df.NodeOutputs["seed"]; seed != nil {
+			inputPaths = append(inputPaths, seed.OutputFiles...)
 		}
 	}
 	inputPaths = uniqueSortedStrings(inputPaths)
-
-	prov := &NodeProvenance{
-		RecordedAt:  time.Now().UTC(),
-		Parents:     parents,
-		DataParents: append([]string(nil), dataParents...),
-		Approval:    approvalDecision(node, cfg),
-	}
-	sort.Strings(prov.DataParents)
-
 	for _, path := range inputPaths {
-		ref, err := artifactRef(df, path, "raw-input")
+		ref, err := artifactRef(df, path, "input")
 		if err != nil {
 			continue
 		}
+		ref.SourceNode = sourceNodeForArtifact(df, path)
 		prov.InputArtifacts = append(prov.InputArtifacts, ref)
 	}
-	if strings.TrimSpace(preparedInput) != "" {
-		if ref, err := artifactRef(df, preparedInput, "prepared-input"); err == nil {
-			prov.PreparedInput = &ref
-		}
-	}
-	for _, path := range output.OutputFiles {
+	for _, path := range uniqueSortedStrings(output.OutputFiles) {
 		ref, err := artifactRef(df, path, "output")
 		if err != nil {
 			continue
 		}
+		ref.SourceNode = output.NodeID
 		prov.OutputArtifacts = append(prov.OutputArtifacts, ref)
 	}
-	for _, edge := range dag.IncomingEdges(node.ID) {
-		if !edge.Control {
-			continue
-		}
-		state := df.GlobalState.NodeStates[edge.From]
-		matched := false
-		if state == NodeCompleted {
-			matched = conditionMatches(edge.Condition, outputFiles(df, edge.From), df, cfg)
-		}
-		prov.ControlDecisions = append(prov.ControlDecisions, ControlDecision{
-			From: edge.From, To: edge.To, Condition: edge.Condition, Label: edge.Label,
-			Matched: matched, ParentState: state,
-		})
-	}
-	sort.Slice(prov.ControlDecisions, func(i, j int) bool {
-		return prov.ControlDecisions[i].From < prov.ControlDecisions[j].From
-	})
 
-	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-provenance.json", node.ID))
+	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-provenance.json", output.NodeID))
 	data, err := json.MarshalIndent(prov, "", "  ")
 	if err != nil {
-		return fmt.Errorf("marshal provenance for %s: %w", node.ID, err)
+		return fmt.Errorf("marshal provenance for %s: %w", output.NodeID, err)
 	}
 	if err := os.WriteFile(path, data, 0o600); err != nil {
-		return fmt.Errorf("write provenance for %s: %w", node.ID, err)
+		return fmt.Errorf("write provenance for %s: %w", output.NodeID, err)
 	}
 	if output.Metadata == nil {
 		output.Metadata = make(map[string]string)
@@ -140,24 +88,6 @@ func attachNodeProvenance(df *DataFlow, node *graph.Node, dag *graph.DAG, cfg Ru
 	return nil
 }
 
-func approvalDecision(node *graph.Node, cfg RunConfig) ApprovalDecision {
-	key := strings.TrimSpace(node.Policy.Approval)
-	if key == "" && node.Policy.RequiresApproval {
-		key = node.ID
-	}
-	approved := false
-	if key != "" {
-		approved = cfg.Approvals[key] || cfg.Approvals[node.ID]
-	}
-	if node.Policy.Intrusive && (cfg.AllowIntrusive || cfg.Approvals["intrusive"]) {
-		approved = true
-	}
-	return ApprovalDecision{
-		Key: key, RequiresApproval: node.Policy.RequiresApproval, Intrusive: node.Policy.Intrusive,
-		Approved: approved, BroadIntrusiveMode: cfg.AllowIntrusive || cfg.Approvals["intrusive"],
-	}
-}
-
 func artifactRef(df *DataFlow, path, role string) (ArtifactRef, error) {
 	stat, err := os.Stat(path)
 	if err != nil {
@@ -179,6 +109,20 @@ func artifactRef(df *DataFlow, path, role string) (ArtifactRef, error) {
 	}, nil
 }
 
+func sourceNodeForArtifact(df *DataFlow, path string) string {
+	for nodeID, output := range df.NodeOutputs {
+		if output == nil {
+			continue
+		}
+		for _, candidate := range output.OutputFiles {
+			if candidate == path {
+				return nodeID
+			}
+		}
+	}
+	return ""
+}
+
 func uniqueSortedStrings(items []string) []string {
 	seen := make(map[string]struct{}, len(items))
 	out := make([]string, 0, len(items))

From 6efc999eac06ad8feed6ef8672e94858bc13c8c9 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:24:49 +0200
Subject: [PATCH 06/15] feat(pipeline): persist correlation before reporting

---
 internal/pipeline/correlation.go | 105 ++++++++++++++++++++-----------
 1 file changed, 68 insertions(+), 37 deletions(-)

diff --git a/internal/pipeline/correlation.go b/internal/pipeline/correlation.go
index a1e74f4..2f4b584 100644
--- a/internal/pipeline/correlation.go
+++ b/internal/pipeline/correlation.go
@@ -11,8 +11,6 @@ import (
 	"strconv"
 	"strings"
 	"time"
-
-	"github.com/MKlolbullen/termaid/internal/graph"
 )
 
 // CorrelationKey groups tool observations that describe the same underlying
@@ -41,7 +39,7 @@ func CorrelationKey(record DataRecord) string {
 
 // CorrelationGroup preserves every raw observation contributing to one logical
 // finding/evidence item while still selecting a representative record for
-// reporting. This prevents de-duplication from destroying source evidence.
+// reporting. De-duplication therefore never destroys source evidence.
 type CorrelationGroup struct {
 	Key            string       `json:"key"`
 	Representative DataRecord   `json:"representative"`
@@ -124,34 +122,33 @@ func cloneRecord(record DataRecord) DataRecord {
 	return copyRecord
 }
 
-// writeCorrelationSnapshot correlates a merge node from its raw parent outputs,
-// not from the already-normalized merged file. That retains source identity and
-// gives report generation a candidate/evidence snapshot created before the sink.
-func writeCorrelationSnapshot(df *DataFlow, node *graph.Node, parents []string) error {
-	output := df.NodeOutputs[node.ID]
-	if output == nil {
-		return fmt.Errorf("node %s has no recorded output for correlation", node.ID)
+// persistMergeCorrelation correlates directly from a merge node's raw parent
+// files before the normalized merge artifact moves downstream. The sidecar is
+// linked from NodeOutput metadata; raw parent files remain untouched.
+func persistMergeCorrelation(df *DataFlow, output *NodeOutput) error {
+	if df == nil || output == nil {
+		return fmt.Errorf("cannot correlate nil merge output")
 	}
 	var records []DataRecord
-	for _, parentID := range parents {
-		for _, file := range outputFiles(df, parentID) {
-			parsed, err := df.parseFile(file, parentID)
-			if err != nil {
-				continue
-			}
-			records = append(records, parsed...)
+	for _, file := range uniqueSortedStrings(df.GlobalState.DataLinks[output.NodeID]) {
+		source := sourceNodeForArtifact(df, file)
+		if source == "" {
+			source = filepath.Base(file)
+		}
+		parsed, err := df.parseFile(file, source)
+		if err != nil {
+			continue
 		}
+		records = append(records, parsed...)
 	}
 	groups := CorrelateGroups(records)
 	payload := struct {
 		NodeID      string             `json:"node_id"`
-		Artifact    string             `json:"artifact"`
 		GeneratedAt time.Time          `json:"generated_at"`
 		Groups      []CorrelationGroup `json:"groups"`
-	}{
-		NodeID: node.ID, Artifact: artifactTypes(node.Outputs), GeneratedAt: time.Now().UTC(), Groups: groups,
-	}
-	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-correlation.json", node.ID))
+	}{NodeID: output.NodeID, GeneratedAt: time.Now().UTC(), Groups: groups}
+
+	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-correlation.json", output.NodeID))
 	data, err := json.MarshalIndent(payload, "", "  ")
 	if err != nil {
 		return err
@@ -168,8 +165,8 @@ func writeCorrelationSnapshot(df *DataFlow, node *graph.Node, parents []string)
 }
 
 // CorrelationReport is the report sink's structured output. Candidate findings
-// and verification evidence stay separate lifecycle stages while each retains
-// its raw observations.
+// and verification evidence remain separate lifecycle stages while every group
+// retains all raw observations.
 type CorrelationReport struct {
 	Version     string             `json:"version"`
 	GeneratedAt time.Time          `json:"generated_at"`
@@ -177,10 +174,20 @@ type CorrelationReport struct {
 	Evidence    []CorrelationGroup `json:"evidence"`
 }
 
-func writeCorrelationReport(df *DataFlow, node *graph.Node, fallbackInput string) (string, error) {
+// persistCorrelationReport is called only after upstream merge nodes have been
+// decorated with their declared output type, so candidate and evidence
+// snapshots can be classified without guessing from filenames.
+func persistCorrelationReport(df *DataFlow, sinkOutput *NodeOutput) error {
+	if df == nil || sinkOutput == nil {
+		return fmt.Errorf("cannot create correlation report for nil sink")
+	}
 	var candidateRecords []DataRecord
 	var evidenceRecords []DataRecord
-	for _, output := range df.NodeOutputs {
+
+	for nodeID, output := range df.NodeOutputs {
+		if output == nil || nodeID == sinkOutput.NodeID {
+			continue
+		}
 		path := output.Metadata["correlation_file"]
 		if path == "" {
 			continue
@@ -190,31 +197,46 @@ func writeCorrelationReport(df *DataFlow, node *graph.Node, fallbackInput string
 			continue
 		}
 		for _, group := range groups {
-			if strings.Contains(output.Metadata["outputs"], string(graph.ArtifactFinding)) {
+			switch {
+			case metadataHasArtifact(output.Metadata, "finding"):
 				candidateRecords = append(candidateRecords, group.Observations...)
-			}
-			if strings.Contains(output.Metadata["outputs"], string(graph.ArtifactEvidence)) {
+			case metadataHasArtifact(output.Metadata, "evidence"):
 				evidenceRecords = append(evidenceRecords, group.Observations...)
 			}
 		}
 	}
-	if len(evidenceRecords) == 0 && fallbackInput != "" {
-		records, err := df.parseFile(fallbackInput, node.ID)
-		if err == nil {
-			evidenceRecords = append(evidenceRecords, records...)
+
+	// A legacy/untyped workflow may not have an evidence-typed merge. Preserve a
+	// useful report by correlating the sink's actual input as a fallback.
+	if len(evidenceRecords) == 0 {
+		for _, file := range df.GlobalState.DataLinks[sinkOutput.NodeID] {
+			source := sourceNodeForArtifact(df, file)
+			parsed, err := df.parseFile(file, source)
+			if err == nil {
+				evidenceRecords = append(evidenceRecords, parsed...)
+			}
 		}
 	}
 
 	report := CorrelationReport{
 		Version: "1.0", GeneratedAt: time.Now().UTC(),
-		Candidates: CorrelateGroups(candidateRecords), Evidence: CorrelateGroups(evidenceRecords),
+		Candidates: CorrelateGroups(candidateRecords),
+		Evidence:   CorrelateGroups(evidenceRecords),
 	}
-	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-correlated-report.json", node.ID))
+	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-correlation-report.json", sinkOutput.NodeID))
 	data, err := json.MarshalIndent(report, "", "  ")
 	if err != nil {
-		return "", err
+		return err
+	}
+	if err := os.WriteFile(path, data, 0o600); err != nil {
+		return err
 	}
-	return path, os.WriteFile(path, data, 0o600)
+	if sinkOutput.Metadata == nil {
+		sinkOutput.Metadata = make(map[string]string)
+	}
+	sinkOutput.Metadata["correlation_report_file"] = path
+	sinkOutput.OutputFiles = append(sinkOutput.OutputFiles, path)
+	return nil
 }
 
 func readCorrelationGroups(path string) ([]CorrelationGroup, error) {
@@ -230,3 +252,12 @@ func readCorrelationGroups(path string) ([]CorrelationGroup, error) {
 	}
 	return payload.Groups, nil
 }
+
+func metadataHasArtifact(metadata map[string]string, want string) bool {
+	for _, item := range strings.Split(metadata["outputs"], ",") {
+		if strings.EqualFold(strings.TrimSpace(item), want) {
+			return true
+		}
+	}
+	return false
+}

From 08ba9e0684de9387d25f6aaf555cccbe0974319c Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:25:33 +0200
Subject: [PATCH 07/15] refactor(pipeline): correlate only finding and evidence
 merges

---
 internal/pipeline/correlation.go | 52 +++++++++++++++++++++++++-------
 1 file changed, 41 insertions(+), 11 deletions(-)

diff --git a/internal/pipeline/correlation.go b/internal/pipeline/correlation.go
index 2f4b584..b160402 100644
--- a/internal/pipeline/correlation.go
+++ b/internal/pipeline/correlation.go
@@ -122,13 +122,18 @@ func cloneRecord(record DataRecord) DataRecord {
 	return copyRecord
 }
 
-// persistMergeCorrelation correlates directly from a merge node's raw parent
-// files before the normalized merge artifact moves downstream. The sidecar is
-// linked from NodeOutput metadata; raw parent files remain untouched.
+// persistMergeCorrelation correlates directly from a finding/evidence merge's
+// raw parent files before the normalized artifact moves downstream. URL/host
+// merge stages are intentionally ignored to keep the analysis directory small.
 func persistMergeCorrelation(df *DataFlow, output *NodeOutput) error {
 	if df == nil || output == nil {
 		return fmt.Errorf("cannot correlate nil merge output")
 	}
+	artifact := correlationInputArtifact(df, output.NodeID)
+	if artifact == "" {
+		return nil
+	}
+
 	var records []DataRecord
 	for _, file := range uniqueSortedStrings(df.GlobalState.DataLinks[output.NodeID]) {
 		source := sourceNodeForArtifact(df, file)
@@ -144,9 +149,10 @@ func persistMergeCorrelation(df *DataFlow, output *NodeOutput) error {
 	groups := CorrelateGroups(records)
 	payload := struct {
 		NodeID      string             `json:"node_id"`
+		Artifact    string             `json:"artifact"`
 		GeneratedAt time.Time          `json:"generated_at"`
 		Groups      []CorrelationGroup `json:"groups"`
-	}{NodeID: output.NodeID, GeneratedAt: time.Now().UTC(), Groups: groups}
+	}{NodeID: output.NodeID, Artifact: artifact, GeneratedAt: time.Now().UTC(), Groups: groups}
 
 	path := filepath.Join(df.WorkDir, df.RunID, "analysis", fmt.Sprintf("%s-correlation.json", output.NodeID))
 	data, err := json.MarshalIndent(payload, "", "  ")
@@ -160,6 +166,7 @@ func persistMergeCorrelation(df *DataFlow, output *NodeOutput) error {
 		output.Metadata = make(map[string]string)
 	}
 	output.Metadata["correlation_file"] = path
+	output.Metadata["correlation_artifact"] = artifact
 	output.Metadata["correlation_groups"] = strconv.Itoa(len(groups))
 	return nil
 }
@@ -174,9 +181,6 @@ type CorrelationReport struct {
 	Evidence    []CorrelationGroup `json:"evidence"`
 }
 
-// persistCorrelationReport is called only after upstream merge nodes have been
-// decorated with their declared output type, so candidate and evidence
-// snapshots can be classified without guessing from filenames.
 func persistCorrelationReport(df *DataFlow, sinkOutput *NodeOutput) error {
 	if df == nil || sinkOutput == nil {
 		return fmt.Errorf("cannot create correlation report for nil sink")
@@ -197,10 +201,10 @@ func persistCorrelationReport(df *DataFlow, sinkOutput *NodeOutput) error {
 			continue
 		}
 		for _, group := range groups {
-			switch {
-			case metadataHasArtifact(output.Metadata, "finding"):
+			switch output.Metadata["correlation_artifact"] {
+			case "finding":
 				candidateRecords = append(candidateRecords, group.Observations...)
-			case metadataHasArtifact(output.Metadata, "evidence"):
+			case "evidence":
 				evidenceRecords = append(evidenceRecords, group.Observations...)
 			}
 		}
@@ -235,7 +239,9 @@ func persistCorrelationReport(df *DataFlow, sinkOutput *NodeOutput) error {
 		sinkOutput.Metadata = make(map[string]string)
 	}
 	sinkOutput.Metadata["correlation_report_file"] = path
-	sinkOutput.OutputFiles = append(sinkOutput.OutputFiles, path)
+	if ref, err := artifactRef(df, path, "analysis-report"); err == nil {
+		sinkOutput.Metadata["correlation_report_sha256"] = ref.SHA256
+	}
 	return nil
 }
 
@@ -253,6 +259,30 @@ func readCorrelationGroups(path string) ([]CorrelationGroup, error) {
 	return payload.Groups, nil
 }
 
+func correlationInputArtifact(df *DataFlow, nodeID string) string {
+	kind := ""
+	for _, file := range df.GlobalState.DataLinks[nodeID] {
+		source := sourceNodeForArtifact(df, file)
+		output := df.NodeOutputs[source]
+		if output == nil {
+			continue
+		}
+		if metadataHasArtifact(output.Metadata, "evidence") {
+			if kind != "" && kind != "evidence" {
+				return ""
+			}
+			kind = "evidence"
+		}
+		if metadataHasArtifact(output.Metadata, "finding") {
+			if kind != "" && kind != "finding" {
+				return ""
+			}
+			kind = "finding"
+		}
+	}
+	return kind
+}
+
 func metadataHasArtifact(metadata map[string]string, want string) bool {
 	for _, item := range strings.Split(metadata["outputs"], ",") {
 		if strings.EqualFold(strings.TrimSpace(item), want) {

From b6959bb427fb898ee24e755698e491ffa7a91000 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:26:16 +0200
Subject: [PATCH 08/15] feat(pipeline): persist provenance and pre-report
 correlation

---
 internal/pipeline/dataflow.go | 22 +++++++++++++++++++++-
 1 file changed, 21 insertions(+), 1 deletion(-)

diff --git a/internal/pipeline/dataflow.go b/internal/pipeline/dataflow.go
index afc5b30..c262553 100644
--- a/internal/pipeline/dataflow.go
+++ b/internal/pipeline/dataflow.go
@@ -283,7 +283,8 @@ func (df *DataFlow) RecordNodeOutput(nodeID, tool string, startTime, endTime tim
 		Metadata:    make(map[string]string),
 	}
 
-	// Store node output
+	// Store node output before analysis hooks so they can resolve provenance
+	// from already-completed parent outputs.
 	df.NodeOutputs[nodeID] = nodeOutput
 
 	// Update global state
@@ -295,6 +296,25 @@ func (df *DataFlow) RecordNodeOutput(nodeID, tool string, startTime, endTime tim
 		df.GlobalState.Statistics.FailedNodes++
 	}
 
+	// Candidate/evidence correlation is produced at merge time, before a later
+	// report sink consumes normalized data. This preserves all raw observations.
+	if exitCode == 0 && strings.HasPrefix(tool, "builtin:merge") {
+		if err := persistMergeCorrelation(df, nodeOutput); err != nil {
+			return fmt.Errorf("persist correlation for %s: %w", nodeID, err)
+		}
+	}
+	if exitCode == 0 && strings.HasPrefix(tool, "builtin:sink") {
+		if err := persistCorrelationReport(df, nodeOutput); err != nil {
+			return fmt.Errorf("persist correlation report for %s: %w", nodeID, err)
+		}
+	}
+
+	// Every recorded result gets a byte-level provenance sidecar, including
+	// failed executions that may have partial output worth preserving.
+	if err := persistNodeProvenance(df, nodeOutput); err != nil {
+		return fmt.Errorf("persist provenance for %s: %w", nodeID, err)
+	}
+
 	// Create analysis summary
 	if err := df.createNodeAnalysis(nodeOutput); err != nil {
 		return fmt.Errorf("failed to create node analysis: %w", err)

From 2d3b6d7055e9ceafd1542fe661547cb059fa8c64 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:26:44 +0200
Subject: [PATCH 09/15] test(graph): enforce control and approval invariants

---
 internal/graph/policy_validation_test.go | 111 +++++++++++++++++++++++
 1 file changed, 111 insertions(+)
 create mode 100644 internal/graph/policy_validation_test.go

diff --git a/internal/graph/policy_validation_test.go b/internal/graph/policy_validation_test.go
new file mode 100644
index 0000000..5249a11
--- /dev/null
+++ b/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)
+	}
+}

From ab91c05287311d2b39a0211bfa5a3440af03717b Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:27:03 +0200
Subject: [PATCH 10/15] test(pipeline): preserve raw evidence through
 correlation

---
 .../pipeline/provenance_correlation_test.go   | 178 ++++++++++++++++++
 1 file changed, 178 insertions(+)
 create mode 100644 internal/pipeline/provenance_correlation_test.go

diff --git a/internal/pipeline/provenance_correlation_test.go b/internal/pipeline/provenance_correlation_test.go
new file mode 100644
index 0000000..84f00a2
--- /dev/null
+++ b/internal/pipeline/provenance_correlation_test.go
@@ -0,0 +1,178 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+)
+
+func TestCorrelateGroupsPreservesEveryObservation(t *testing.T) {
+	records := []DataRecord{
+		{Value: "scanner A observation", Source: "scanner-a", Confidence: 0.7, Metadata: map[string]string{"asset": "example.com", "location": "/admin", "weakness": "exposure", "evidence": "200"}},
+		{Value: "scanner B richer observation", Source: "scanner-b", Confidence: 0.9, Metadata: map[string]string{"asset": "example.com", "location": "/admin", "weakness": "exposure", "evidence": "200"}},
+	}
+	groups := CorrelateGroups(records)
+	if len(groups) != 1 {
+		t.Fatalf("groups = %d, want 1", len(groups))
+	}
+	if len(groups[0].Observations) != 2 {
+		t.Fatalf("observations = %d, want 2", len(groups[0].Observations))
+	}
+	if len(groups[0].Sources) != 2 {
+		t.Fatalf("sources = %v, want both scanners", groups[0].Sources)
+	}
+	if groups[0].Representative.Source != "scanner-b" {
+		t.Fatalf("representative source = %q, want scanner-b", groups[0].Representative.Source)
+	}
+}
+
+func TestRecordedMergeKeepsRawProvenanceAndCorrelation(t *testing.T) {
+	df, err := NewDataFlow(t.TempDir(), "example.com")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := df.CreateSeedFile(); err != nil {
+		t.Fatal(err)
+	}
+
+	a := filepath.Join(df.WorkDir, df.RunID, "raw", "scanner-a.txt")
+	b := filepath.Join(df.WorkDir, df.RunID, "raw", "scanner-b.txt")
+	if err := os.WriteFile(a, []byte("same finding\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	if err := os.WriteFile(b, []byte("same finding\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now()
+	if err := df.RecordNodeOutput("scanner-a", "scanner-a", now, now, 0, []string{a}, ""); err != nil {
+		t.Fatal(err)
+	}
+	df.NodeOutputs["scanner-a"].Metadata["outputs"] = "finding"
+	if err := df.RecordNodeOutput("scanner-b", "scanner-b", now, now, 0, []string{b}, ""); err != nil {
+		t.Fatal(err)
+	}
+	df.NodeOutputs["scanner-b"].Metadata["outputs"] = "finding"
+
+	merged := filepath.Join(df.WorkDir, df.RunID, "processed", "finding-merge.txt")
+	if err := os.WriteFile(merged, []byte("same finding\n"), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	df.GlobalState.DataLinks["finding-merge"] = []string{a, b}
+	if err := df.RecordNodeOutput("finding-merge", "builtin:merge", now, now, 0, []string{merged}, ""); err != nil {
+		t.Fatal(err)
+	}
+	df.NodeOutputs["finding-merge"].Metadata["outputs"] = "finding"
+
+	correlationPath := df.NodeOutputs["finding-merge"].Metadata["correlation_file"]
+	if correlationPath == "" {
+		t.Fatal("finding merge did not produce a correlation sidecar")
+	}
+	groups, err := readCorrelationGroups(correlationPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if len(groups) != 1 || len(groups[0].Observations) != 2 {
+		t.Fatalf("correlation groups = %#v, want one group with two raw observations", groups)
+	}
+	if !strings.Contains(strings.Join(groups[0].Sources, ","), "scanner-a") || !strings.Contains(strings.Join(groups[0].Sources, ","), "scanner-b") {
+		t.Fatalf("correlation sources lost: %v", groups[0].Sources)
+	}
+
+	provenancePath := df.NodeOutputs["finding-merge"].Metadata["provenance_file"]
+	data, err := os.ReadFile(provenancePath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var prov NodeProvenance
+	if err := json.Unmarshal(data, &prov); err != nil {
+		t.Fatal(err)
+	}
+	if len(prov.InputArtifacts) != 2 {
+		t.Fatalf("provenance inputs = %d, want 2", len(prov.InputArtifacts))
+	}
+	for _, ref := range prov.InputArtifacts {
+		if ref.SHA256 == "" || ref.SourceNode == "" {
+			t.Fatalf("incomplete provenance ref: %#v", ref)
+		}
+	}
+}
+
+func TestSinkReportSeparatesCandidatesAndEvidence(t *testing.T) {
+	df, err := NewDataFlow(t.TempDir(), "example.com")
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := df.CreateSeedFile(); err != nil {
+		t.Fatal(err)
+	}
+	now := time.Now()
+
+	candidateA := writeTestArtifact(t, df, "candidate-a.txt", "candidate issue\n")
+	candidateB := writeTestArtifact(t, df, "candidate-b.txt", "candidate issue\n")
+	recordTypedOutput(t, df, "candidate-a", "scanner-a", candidateA, "finding", now)
+	recordTypedOutput(t, df, "candidate-b", "scanner-b", candidateB, "finding", now)
+	candidateMerge := writeTestArtifact(t, df, "candidate-merge.txt", "candidate issue\n")
+	df.GlobalState.DataLinks["finding-merge"] = []string{candidateA, candidateB}
+	if err := df.RecordNodeOutput("finding-merge", "builtin:merge", now, now, 0, []string{candidateMerge}, ""); err != nil {
+		t.Fatal(err)
+	}
+	df.NodeOutputs["finding-merge"].Metadata["outputs"] = "finding"
+
+	evidenceA := writeTestArtifact(t, df, "evidence-a.txt", "verified issue\n")
+	evidenceB := writeTestArtifact(t, df, "evidence-b.txt", "verified issue\n")
+	recordTypedOutput(t, df, "evidence-a", "verify-a", evidenceA, "evidence", now)
+	recordTypedOutput(t, df, "evidence-b", "verify-b", evidenceB, "evidence", now)
+	evidenceMerge := writeTestArtifact(t, df, "evidence-merge.txt", "verified issue\n")
+	df.GlobalState.DataLinks["evidence-merge"] = []string{evidenceA, evidenceB}
+	if err := df.RecordNodeOutput("evidence-merge", "builtin:merge", now, now, 0, []string{evidenceMerge}, ""); err != nil {
+		t.Fatal(err)
+	}
+	df.NodeOutputs["evidence-merge"].Metadata["outputs"] = "evidence"
+
+	sinkFile := writeTestArtifact(t, df, "report-sink.txt", "verified issue\n")
+	df.GlobalState.DataLinks["report-sink"] = []string{evidenceMerge}
+	if err := df.RecordNodeOutput("report-sink", "builtin:sink", now, now, 0, []string{sinkFile}, ""); err != nil {
+		t.Fatal(err)
+	}
+	reportPath := df.NodeOutputs["report-sink"].Metadata["correlation_report_file"]
+	if reportPath == "" {
+		t.Fatal("sink did not produce correlation report")
+	}
+	data, err := os.ReadFile(reportPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var report CorrelationReport
+	if err := json.Unmarshal(data, &report); err != nil {
+		t.Fatal(err)
+	}
+	if len(report.Candidates) != 1 || len(report.Candidates[0].Observations) != 2 {
+		t.Fatalf("candidate correlation lost observations: %#v", report.Candidates)
+	}
+	if len(report.Evidence) != 1 || len(report.Evidence[0].Observations) != 2 {
+		t.Fatalf("evidence correlation lost observations: %#v", report.Evidence)
+	}
+	if df.NodeOutputs["report-sink"].Metadata["correlation_report_sha256"] == "" {
+		t.Fatal("correlation report hash missing")
+	}
+}
+
+func writeTestArtifact(t *testing.T, df *DataFlow, name, content string) string {
+	t.Helper()
+	path := filepath.Join(df.WorkDir, df.RunID, "raw", name)
+	if err := os.WriteFile(path, []byte(content), 0o644); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}
+
+func recordTypedOutput(t *testing.T, df *DataFlow, nodeID, tool, path, outputType string, now time.Time) {
+	t.Helper()
+	if err := df.RecordNodeOutput(nodeID, tool, now, now, 0, []string{path}, ""); err != nil {
+		t.Fatal(err)
+	}
+	df.NodeOutputs[nodeID].Metadata["outputs"] = outputType
+}

From 978dc8bdf16bc9428969c3d09e18fd9203c2ba7a Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:27:17 +0200
Subject: [PATCH 11/15] temp

---
 docs/INVALID | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 docs/INVALID

diff --git a/docs/INVALID b/docs/INVALID
new file mode 100644
index 0000000..c1b0730
--- /dev/null
+++ b/docs/INVALID
@@ -0,0 +1 @@
+x
\ No newline at end of file

From 66424d596759d6a6e252c91dbd7f6fb2dac462c2 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:27:26 +0200
Subject: [PATCH 12/15] chore: remove accidental placeholder

---
 docs/INVALID | 1 -
 1 file changed, 1 deletion(-)
 delete mode 100644 docs/INVALID

diff --git a/docs/INVALID b/docs/INVALID
deleted file mode 100644
index c1b0730..0000000
--- a/docs/INVALID
+++ /dev/null
@@ -1 +0,0 @@
-x
\ No newline at end of file

From 17833e0c906b0698e8b1850703367c628ac844a8 Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:27:59 +0200
Subject: [PATCH 13/15] temp

---
 docs/INVALID | 1 +
 1 file changed, 1 insertion(+)
 create mode 100644 docs/INVALID

diff --git a/docs/INVALID b/docs/INVALID
new file mode 100644
index 0000000..c1b0730
--- /dev/null
+++ b/docs/INVALID
@@ -0,0 +1 @@
+x
\ No newline at end of file

From b72f14db5c02764adf420a69efd3939b4e376bef Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:28:07 +0200
Subject: [PATCH 14/15] chore: remove accidental placeholder

---
 docs/INVALID | 1 -
 1 file changed, 1 deletion(-)
 delete mode 100644 docs/INVALID

diff --git a/docs/INVALID b/docs/INVALID
deleted file mode 100644
index c1b0730..0000000
--- a/docs/INVALID
+++ /dev/null
@@ -1 +0,0 @@
-x
\ No newline at end of file

From 98184fa92ad468cf6c2395fdaa3337701fb6ed5c Mon Sep 17 00:00:00 2001
From: 0xlolbullen <53312929+MKlolbullen@users.noreply.github.com>
Date: Sat, 22 Aug 2026 11:29:14 +0200
Subject: [PATCH 15/15] fix(graph): close topological ready loop

---
 internal/graph/semantics.go | 1 +
 1 file changed, 1 insertion(+)

diff --git a/internal/graph/semantics.go b/internal/graph/semantics.go
index 1e88643..aed1862 100644
--- a/internal/graph/semantics.go
+++ b/internal/graph/semantics.go
@@ -209,6 +209,7 @@ func (g *DAG) TopologicalOrder() ([]string, error) {
 				ready = append(ready, e.To)
 				sort.Strings(ready)
 			}
+		}
 	}
 	if len(order) != len(g.Nodes) {
 		return nil, fmt.Errorf("workflow is not acyclic")