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)
+	}
+}
diff --git a/internal/graph/semantics.go b/internal/graph/semantics.go
index 905ac6c..aed1862 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) {
diff --git a/internal/pipeline/correlation.go b/internal/pipeline/correlation.go
index 11363d9..b160402 100644
--- a/internal/pipeline/correlation.go
+++ b/internal/pipeline/correlation.go
@@ -3,8 +3,14 @@ package pipeline
 import (
 	"crypto/sha256"
 	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"os"
+	"path/filepath"
 	"sort"
+	"strconv"
 	"strings"
+	"time"
 )
 
 // CorrelationKey groups tool observations that describe the same underlying
@@ -31,54 +37,257 @@ 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. De-duplication therefore never destroys 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
+}
+
+// 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)
+		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: 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, "", "  ")
+	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_artifact"] = artifact
+	output.Metadata["correlation_groups"] = strconv.Itoa(len(groups))
+	return nil
+}
+
+// CorrelationReport is the report sink's structured output. Candidate findings
+// 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"`
+	Candidates  []CorrelationGroup `json:"candidates"`
+	Evidence    []CorrelationGroup `json:"evidence"`
+}
+
+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 nodeID, output := range df.NodeOutputs {
+		if output == nil || nodeID == sinkOutput.NodeID {
+			continue
+		}
+		path := output.Metadata["correlation_file"]
+		if path == "" {
+			continue
+		}
+		groups, err := readCorrelationGroups(path)
+		if err != nil {
+			continue
+		}
+		for _, group := range groups {
+			switch output.Metadata["correlation_artifact"] {
+			case "finding":
+				candidateRecords = append(candidateRecords, group.Observations...)
+			case "evidence":
+				evidenceRecords = append(evidenceRecords, group.Observations...)
+			}
+		}
+	}
+
+	// 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),
+	}
+	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
+	}
+	if err := os.WriteFile(path, data, 0o600); err != nil {
+		return err
+	}
+	if sinkOutput.Metadata == nil {
+		sinkOutput.Metadata = make(map[string]string)
+	}
+	sinkOutput.Metadata["correlation_report_file"] = path
+	if ref, err := artifactRef(df, path, "analysis-report"); err == nil {
+		sinkOutput.Metadata["correlation_report_sha256"] = ref.SHA256
+	}
+	return nil
+}
+
+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
+}
+
+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) {
+			return true
+		}
+	}
+	return false
+}
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)
diff --git a/internal/pipeline/provenance.go b/internal/pipeline/provenance.go
new file mode 100644
index 0000000..b96a2c5
--- /dev/null
+++ b/internal/pipeline/provenance.go
@@ -0,0 +1,141 @@
+package pipeline
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"fmt"
+	"io"
+	"os"
+	"path/filepath"
+	"sort"
+	"time"
+)
+
+// ArtifactRef is an immutable description of a file that participated in a
+// 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"`
+	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"`
+}
+
+// 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 {
+	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"`
+}
+
+// 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")
+	}
+	prov := NodeProvenance{
+		NodeID: output.NodeID, Tool: output.Tool, RecordedAt: time.Now().UTC(),
+	}
+
+	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)
+	for _, path := range inputPaths {
+		ref, err := artifactRef(df, path, "input")
+		if err != nil {
+			continue
+		}
+		ref.SourceNode = sourceNodeForArtifact(df, path)
+		prov.InputArtifacts = append(prov.InputArtifacts, ref)
+	}
+	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)
+	}
+
+	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", output.NodeID, err)
+	}
+	if err := os.WriteFile(path, data, 0o600); err != nil {
+		return fmt.Errorf("write provenance for %s: %w", output.NodeID, err)
+	}
+	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
+}
+
+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 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))
+	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
+}
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
+}