Materialize extraction reference files

internal/framework/pipeline/profile.go

@@ -61,12 +61,13 @@ type ReferenceUnbind struct {
 }
 
 type ResolvedArtifactLane struct {
-	ID         string
-	Extract    ModuleBinding
-	Merge      ModuleBinding
-	Normalize  ModuleBinding
-	Validators []ModuleBinding
-	References []ReferenceBinding `json:"references,omitempty"`
+	ID           string
+	Extract      ModuleBinding
+	Merge        ModuleBinding
+	Normalize    ModuleBinding
+	Validators   []ModuleBinding
+	References   []ReferenceBinding     `json:"references,omitempty"`
+	ReferenceSet contracts.ReferenceSet `json:"-"`
 }
 
 type ResolvedPipeline struct {

internal/framework/pipeline/references.go

@@ -0,0 +1,187 @@
+package pipeline
+
+import (
+	"crypto/sha256"
+	"encoding/hex"
+	"fmt"
+	"net/url"
+	"os"
+	"path/filepath"
+	"sort"
+	"strings"
+	"unicode/utf8"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+const (
+	referenceOriginFile = "file"
+	referenceMediaType  = "text/plain; charset=utf-8"
+)
+
+type ReferenceMaterializationOptions struct {
+	ConfigPath string
+	WorkingDir string
+}
+
+func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, options ReferenceMaterializationOptions) (ResolvedPipeline, []contracts.Warning, error) {
+	out := resolved
+	if len(resolved.ArtifactLanes) == 0 {
+		return out, nil, nil
+	}
+
+	warnings := []contracts.Warning(nil)
+	out.ArtifactLanes = make([]ResolvedArtifactLane, len(resolved.ArtifactLanes))
+	for i, lane := range resolved.ArtifactLanes {
+		materializedLane := lane
+		referenceSet, laneWarnings, err := materializeLaneReferences(resolved.ID, lane, catalog, options)
+		if err != nil {
+			return ResolvedPipeline{}, nil, err
+		}
+		materializedLane.ReferenceSet = referenceSet
+		out.ArtifactLanes[i] = materializedLane
+		warnings = append(warnings, laneWarnings...)
+	}
+	return out, warnings, nil
+}
+
+func materializeLaneReferences(
+	pipelineID string,
+	lane ResolvedArtifactLane,
+	catalog ModuleCatalog,
+	options ReferenceMaterializationOptions,
+) (contracts.ReferenceSet, []contracts.Warning, error) {
+	if len(lane.References) == 0 {
+		return contracts.ReferenceSet{}, nil, nil
+	}
+	if catalog.Extractors == nil {
+		return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", pipelineID, lane.ID, lane.Extract.Module, lane.Extract.Module)
+	}
+	spec, ok := catalog.Extractors.Spec(lane.Extract.Module)
+	if !ok {
+		return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q extract module %q: module %q is not registered", pipelineID, lane.ID, lane.Extract.Module, lane.Extract.Module)
+	}
+
+	slotByName := make(map[string]contracts.ReferenceSlot, len(spec.ReferenceSlots))
+	for _, slot := range spec.ReferenceSlots {
+		slotByName[slot.Name] = slot
+	}
+
+	set := contracts.ReferenceSet{Slots: make(map[string]contracts.ResolvedReferenceSlot, len(lane.References))}
+	var warnings []contracts.Warning
+	for _, binding := range lane.References {
+		slotName := strings.TrimSpace(binding.SlotName)
+		slot, ok := slotByName[slotName]
+		if !ok {
+			return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q is not declared by extractor %q", pipelineID, lane.ID, slotName, lane.Extract.Module)
+		}
+
+		path, err := referencePath(binding, options)
+		if err != nil {
+			return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q path %q: %w", pipelineID, lane.ID, slotName, binding.Source, err)
+		}
+		content, err := os.ReadFile(path)
+		if err != nil {
+			return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q read %q: %w", pipelineID, lane.ID, slotName, path, err)
+		}
+		if !utf8.Valid(content) {
+			return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q path %q must be UTF-8 text", pipelineID, lane.ID, slotName, path)
+		}
+		if slot.MaxBytes > 0 && int64(len(content)) > slot.MaxBytes {
+			return contracts.ReferenceSet{}, nil, fmt.Errorf("pipeline %q lane %q reference slot %q path %q is %d bytes, limit %d", pipelineID, lane.ID, slotName, path, len(content), slot.MaxBytes)
+		}
+		if len(content) == 0 {
+			warnings = append(warnings, contracts.Warning{
+				Scope:      fmt.Sprintf("pipeline.%s.lane.%s.reference.%s", pipelineID, lane.ID, slotName),
+				ReasonCode: "empty_reference",
+				Message:    fmt.Sprintf("reference slot %q for lane %q is bound to an empty file", slotName, lane.ID),
+			})
+		}
+
+		item := contracts.ReferenceItem{
+			SlotName:      slotName,
+			MediaType:     referenceMediaType,
+			Content:       append([]byte(nil), content...),
+			Digest:        referenceDigest(content),
+			Origin:        contracts.ReferenceOrigin{Type: referenceOriginFile, URI: fileURI(path)},
+			SizeBytes:     int64(len(content)),
+			BindingSource: strings.TrimSpace(binding.BindingSource),
+		}
+		set.Slots[slotName] = contracts.ResolvedReferenceSlot{
+			Slot:  cloneReferenceSlot(slot),
+			Items: []contracts.ReferenceItem{item},
+		}
+	}
+	return set, warnings, nil
+}
+
+func referencePath(binding ReferenceBinding, options ReferenceMaterializationOptions) (string, error) {
+	source := strings.TrimSpace(binding.Source)
+	if source == "" {
+		return "", fmt.Errorf("must not be empty")
+	}
+	if filepath.IsAbs(source) {
+		return filepath.Clean(source), nil
+	}
+
+	base := strings.TrimSpace(options.WorkingDir)
+	if strings.TrimSpace(binding.BindingSource) == contracts.ReferenceBindingSourceConfig {
+		base = filepath.Dir(strings.TrimSpace(options.ConfigPath))
+	}
+	if base == "" {
+		var err error
+		base, err = os.Getwd()
+		if err != nil {
+			return "", fmt.Errorf("resolve working directory: %w", err)
+		}
+	}
+	return filepath.Clean(filepath.Join(base, source)), nil
+}
+
+func referenceDigest(content []byte) string {
+	sum := sha256.Sum256(content)
+	return "sha256:" + hex.EncodeToString(sum[:])
+}
+
+func fileURI(path string) string {
+	absolute, err := filepath.Abs(path)
+	if err != nil {
+		absolute = path
+	}
+	absolute = filepath.ToSlash(filepath.Clean(absolute))
+	if strings.HasPrefix(absolute, "/") {
+		return "file://" + (&url.URL{Path: absolute}).EscapedPath()
+	}
+	return "file:///" + (&url.URL{Path: absolute}).EscapedPath()
+}
+
+func cloneReferenceSlot(slot contracts.ReferenceSlot) contracts.ReferenceSlot {
+	slot.AcceptedMediaTypes = append([]string(nil), slot.AcceptedMediaTypes...)
+	return slot
+}
+
+func CloneReferenceSet(in contracts.ReferenceSet) contracts.ReferenceSet {
+	if len(in.Slots) == 0 {
+		return contracts.ReferenceSet{}
+	}
+	out := contracts.ReferenceSet{Slots: make(map[string]contracts.ResolvedReferenceSlot, len(in.Slots))}
+	keys := make([]string, 0, len(in.Slots))
+	for key := range in.Slots {
+		keys = append(keys, key)
+	}
+	sort.Strings(keys)
+	for _, key := range keys {
+		slot := in.Slots[key]
+		slot.Slot = cloneReferenceSlot(slot.Slot)
+		if len(slot.Items) > 0 {
+			items := make([]contracts.ReferenceItem, len(slot.Items))
+			for i, item := range slot.Items {
+				item.Content = append([]byte(nil), item.Content...)
+				items[i] = item
+			}
+			slot.Items = items
+		}
+		out.Slots[key] = slot
+	}
+	return out
+}

internal/framework/pipeline/references_test.go

@@ -0,0 +1,178 @@
+package pipeline
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+)
+
+func TestMaterializeReferencesResolvesPathsAndDigestsContent(t *testing.T) {
+	configDir := t.TempDir()
+	workingDir := t.TempDir()
+	configReference := filepath.Join(configDir, "config-reference.txt")
+	cliReference := filepath.Join(workingDir, "cli-reference.txt")
+	writeReferenceFile(t, configReference, []byte("config text"))
+	writeReferenceFile(t, cliReference, []byte("cli text"))
+
+	pipeline := baselineProfile()
+	pipeline.References = map[string]string{"roster": "config-reference.txt"}
+	lane := pipeline.Artifacts["events"]
+	lane.References = map[string]string{"glossary": "cli-reference.txt"}
+	pipeline.Artifacts["events"] = lane
+	catalog := referenceCatalog(t, []contracts.ReferenceSlot{
+		{Name: "roster"},
+		{Name: "glossary"},
+	})
+	resolved, err := ResolvePipeline(pipeline, ResolveOptions{
+		ReferenceOverrides: []ReferenceBinding{
+			{LaneID: "events", SlotName: "glossary", Source: "cli-reference.txt", BindingSource: contracts.ReferenceBindingSourceCLI},
+		},
+	}, catalog)
+	if err != nil {
+		t.Fatalf("ResolvePipeline() error = %v, want nil", err)
+	}
+
+	first, warnings, err := MaterializeReferences(resolved, catalog, ReferenceMaterializationOptions{
+		ConfigPath: filepath.Join(configDir, "config.yml"),
+		WorkingDir: workingDir,
+	})
+	if err != nil {
+		t.Fatalf("MaterializeReferences() error = %v, want nil", err)
+	}
+	if len(warnings) != 0 {
+		t.Fatalf("warnings = %#v, want none", warnings)
+	}
+	second, _, err := MaterializeReferences(resolved, catalog, ReferenceMaterializationOptions{
+		ConfigPath: filepath.Join(configDir, "config.yml"),
+		WorkingDir: workingDir,
+	})
+	if err != nil {
+		t.Fatalf("MaterializeReferences(second) error = %v, want nil", err)
+	}
+
+	referenceSet := first.ArtifactLanes[0].ReferenceSet
+	roster := referenceSet.Slots["roster"].Items[0]
+	if string(roster.Content) != "config text" {
+		t.Fatalf("roster content = %q, want config text", roster.Content)
+	}
+	if roster.Digest != referenceDigest([]byte("config text")) || roster.Digest != second.ArtifactLanes[0].ReferenceSet.Slots["roster"].Items[0].Digest {
+		t.Fatalf("roster digest = %q, want stable digest", roster.Digest)
+	}
+	if roster.BindingSource != contracts.ReferenceBindingSourceConfig {
+		t.Fatalf("roster binding source = %q, want config", roster.BindingSource)
+	}
+	if roster.MediaType != referenceMediaType || roster.Origin.Type != referenceOriginFile || roster.SizeBytes != int64(len("config text")) {
+		t.Fatalf("roster metadata = %#v, want text file metadata", roster)
+	}
+	if !strings.Contains(roster.Origin.URI, "config-reference.txt") {
+		t.Fatalf("roster origin URI = %q, want config reference path", roster.Origin.URI)
+	}
+
+	glossary := referenceSet.Slots["glossary"].Items[0]
+	if string(glossary.Content) != "cli text" {
+		t.Fatalf("glossary content = %q, want cli text", glossary.Content)
+	}
+	if glossary.BindingSource != contracts.ReferenceBindingSourceCLI {
+		t.Fatalf("glossary binding source = %q, want cli", glossary.BindingSource)
+	}
+	if !strings.Contains(glossary.Origin.URI, "cli-reference.txt") {
+		t.Fatalf("glossary origin URI = %q, want cli reference path", glossary.Origin.URI)
+	}
+
+	encoded, err := json.Marshal(first)
+	if err != nil {
+		t.Fatalf("json.Marshal(materialized) error = %v, want nil", err)
+	}
+	if strings.Contains(string(encoded), "config text") || strings.Contains(string(encoded), "cli text") {
+		t.Fatalf("materialized pipeline JSON contains reference content: %s", encoded)
+	}
+}
+
+func TestMaterializeReferencesRejectsNonUTF8Content(t *testing.T) {
+	configDir := t.TempDir()
+	path := filepath.Join(configDir, "bad.txt")
+	writeReferenceFile(t, path, []byte{0xff, 0xfe})
+
+	resolved := resolvedPipelineWithReference(t, "roster", "bad.txt", contracts.ReferenceBindingSourceConfig, contracts.ReferenceSlot{Name: "roster"})
+	_, _, err := MaterializeReferences(resolved, referenceCatalog(t, []contracts.ReferenceSlot{{Name: "roster"}}), ReferenceMaterializationOptions{
+		ConfigPath: filepath.Join(configDir, "config.yml"),
+	})
+	if err == nil || !strings.Contains(err.Error(), "UTF-8") || !strings.Contains(err.Error(), "roster") || !strings.Contains(err.Error(), path) {
+		t.Fatalf("error = %v, want UTF-8 path error", err)
+	}
+}
+
+func TestMaterializeReferencesWarnsForEmptyFiles(t *testing.T) {
+	configDir := t.TempDir()
+	path := filepath.Join(configDir, "empty.txt")
+	writeReferenceFile(t, path, nil)
+
+	resolved := resolvedPipelineWithReference(t, "roster", "empty.txt", contracts.ReferenceBindingSourceConfig, contracts.ReferenceSlot{Name: "roster"})
+	materialized, warnings, err := MaterializeReferences(resolved, referenceCatalog(t, []contracts.ReferenceSlot{{Name: "roster"}}), ReferenceMaterializationOptions{
+		ConfigPath: filepath.Join(configDir, "config.yml"),
+	})
+	if err != nil {
+		t.Fatalf("MaterializeReferences() error = %v, want nil", err)
+	}
+	if len(warnings) != 1 || warnings[0].ReasonCode != "empty_reference" {
+		t.Fatalf("warnings = %#v, want empty reference warning", warnings)
+	}
+	item := materialized.ArtifactLanes[0].ReferenceSet.Slots["roster"].Items[0]
+	if item.SizeBytes != 0 || item.Digest != referenceDigest(nil) {
+		t.Fatalf("empty item = %#v, want zero size and empty digest", item)
+	}
+}
+
+func TestMaterializeReferencesEnforcesMaxBytes(t *testing.T) {
+	configDir := t.TempDir()
+	path := filepath.Join(configDir, "large.txt")
+	writeReferenceFile(t, path, []byte("too large"))
+
+	slot := contracts.ReferenceSlot{Name: "roster", MaxBytes: 3}
+	resolved := resolvedPipelineWithReference(t, "roster", "large.txt", contracts.ReferenceBindingSourceConfig, slot)
+	_, _, err := MaterializeReferences(resolved, referenceCatalog(t, []contracts.ReferenceSlot{slot}), ReferenceMaterializationOptions{
+		ConfigPath: filepath.Join(configDir, "config.yml"),
+	})
+	if err == nil || !strings.Contains(err.Error(), "9 bytes") || !strings.Contains(err.Error(), "limit 3") || !strings.Contains(err.Error(), "roster") {
+		t.Fatalf("error = %v, want max bytes error", err)
+	}
+}
+
+func resolvedPipelineWithReference(t *testing.T, slotName, source, bindingSource string, slot contracts.ReferenceSlot) ResolvedPipeline {
+	t.Helper()
+	profile := baselineProfile()
+	lane := profile.Artifacts["events"]
+	lane.References = map[string]string{slotName: source}
+	profile.Artifacts["events"] = lane
+	catalog := referenceCatalog(t, []contracts.ReferenceSlot{slot})
+	resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
+	if err != nil {
+		t.Fatalf("ResolvePipeline() error = %v, want nil", err)
+	}
+	if bindingSource != contracts.ReferenceBindingSourceConfig {
+		resolved.ArtifactLanes[0].References[0].BindingSource = bindingSource
+	}
+	return resolved
+}
+
+func referenceCatalog(t *testing.T, slots []contracts.ReferenceSlot) ModuleCatalog {
+	t.Helper()
+	return newProfileCatalogWithOverride(t, ModuleSpec{
+		Key:            "event-extractor",
+		Stage:          StageExtract,
+		Requires:       []string{"chunk"},
+		Provides:       []string{"candidate"},
+		ReferenceSlots: slots,
+	})
+}
+
+func writeReferenceFile(t *testing.T, path string, content []byte) {
+	t.Helper()
+	if err := os.WriteFile(path, content, 0o644); err != nil {
+		t.Fatalf("write reference %q: %v", path, err)
+	}
+}

internal/framework/pipeline/runner.go

@@ -41,6 +41,7 @@ type RunInput struct {
 	StartedAt   time.Time
 	LLMProfiles []artifacts.LLMProfileManifest
 	Metadata    map[string]any
+	Warnings    []contracts.Warning
 }
 
 type RunOutput struct {
@@ -63,6 +64,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
 		return output, err
 	}
 
+	output.Warnings = append(output.Warnings, cloneWarnings(input.Warnings)...)
 	output.Manifest = manifestFromPipeline(input)
 
 	adapter, err := r.registries.Inputs.Build(input.Pipeline.Input.Module)
@@ -184,6 +186,7 @@ func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.Source
 		result, err := extractor.Extract(ctx, contracts.ExtractionRequest{
 			Source:     doc,
 			Chunk:      &chunk,
+			References: CloneReferenceSet(lane.ReferenceSet),
 			LLMClient:  input.LLMClient,
 			LLMProfile: lane.Extract.LLMProfile,
 			Options:    cloneOptions(lane.Extract.Options),
@@ -500,6 +503,13 @@ func cloneLLMProfiles(profiles []artifacts.LLMProfileManifest) []artifacts.LLMPr
 	return append([]artifacts.LLMProfileManifest(nil), profiles...)
 }
 
+func cloneWarnings(warnings []contracts.Warning) []contracts.Warning {
+	if len(warnings) == 0 {
+		return nil
+	}
+	return append([]contracts.Warning(nil), warnings...)
+}
+
 func timePtr(t time.Time) *time.Time {
 	return &t
 }

internal/framework/pipeline/runner_test.go

@@ -559,6 +559,60 @@ func TestRunPassesModuleBindingConfigToStageRequests(t *testing.T) {
 	}
 }
 
+func TestRunPassesLaneReferencesToExtractorRequests(t *testing.T) {
+	modules := defaultRunnerModules()
+	pipeline := resolvedPipeline()
+	pipeline.ArtifactLanes[0].ReferenceSet = contracts.ReferenceSet{
+		Slots: map[string]contracts.ResolvedReferenceSlot{
+			"roster": {
+				Slot: contracts.ReferenceSlot{Name: "roster"},
+				Items: []contracts.ReferenceItem{
+					{
+						SlotName:      "roster",
+						MediaType:     "text/plain; charset=utf-8",
+						Content:       []byte("reference text"),
+						Digest:        "sha256:test",
+						Origin:        contracts.ReferenceOrigin{Type: "file", URI: "file:///tmp/reference.txt"},
+						SizeBytes:     int64(len("reference text")),
+						BindingSource: contracts.ReferenceBindingSourceConfig,
+					},
+				},
+			},
+		},
+	}
+
+	_, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: pipeline})
+	if err != nil {
+		t.Fatalf("Run() error = %v, want nil", err)
+	}
+
+	req := modules.extractors["extract-alpha"].requests[0]
+	item := req.References.Slots["roster"].Items[0]
+	if string(item.Content) != "reference text" {
+		t.Fatalf("reference content = %q, want reference text", item.Content)
+	}
+	item.Content[0] = 'R'
+	if got := string(pipeline.ArtifactLanes[0].ReferenceSet.Slots["roster"].Items[0].Content); got != "reference text" {
+		t.Fatalf("runner mutated reference set content = %q", got)
+	}
+}
+
+func TestRunIncludesInputWarnings(t *testing.T) {
+	modules := defaultRunnerModules()
+	warning := contracts.Warning{Scope: "reference", ReasonCode: "empty_reference", Message: "empty reference"}
+
+	output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{
+		Pipeline: resolvedPipeline(),
+		Warnings: []contracts.Warning{warning},
+	})
+	if err != nil {
+		t.Fatalf("Run() error = %v, want nil", err)
+	}
+	if len(output.Warnings) != 1 || output.Warnings[0] != warning {
+		t.Fatalf("warnings = %#v, want input warning", output.Warnings)
+	}
+}
+
 func TestRunRecordsTopLevelModuleMetadataForSingletonModules(t *testing.T) {
 	modules := defaultRunnerModules()
 	modules.input.manifestMetadata = map[string]any{