diff --git a/docs/cli.md b/docs/cli.md index 8f79706..c767586 100644 --- a/docs/cli.md +++ b/docs/cli.md @@ -59,8 +59,9 @@ the warning count is printed to stderr. Reference flags are resolved against selected artifact lanes before the run starts. Flat slot names are accepted only when exactly one selected lane -declares that slot. Binding paths are recorded in resolved pipeline metadata; -reference file content is not read by the current runner. +declares that slot. Bound reference files are read before extraction, validated +as UTF-8 text, and passed only to the lane extractor that declares the slot. +Reference content is not written to diagnostics, logs, errors, or manifests. For durable output, diagnostics, retention, and failure inspection, see [Operations](operations.md). diff --git a/docs/config.md b/docs/config.md index 4b947aa..202824a 100644 --- a/docs/config.md +++ b/docs/config.md @@ -142,8 +142,11 @@ module keys. Reference bindings are validated against extractor-declared slots during pipeline resolution. Required slots must be bound after config defaults, lane-level bindings, and run-time `--reference` or `--without-reference` -overrides are applied. Reference paths are recorded in resolved pipeline -metadata; the current runner does not read reference file content. +overrides are applied. Config-relative paths are resolved relative to the +config file; CLI reference paths are resolved relative to the current working +directory. Bound files must be UTF-8 text and are passed only to lane +extractors that declare the slot. Reference content is not written to +diagnostics, logs, errors, or manifests. ## Module Bindings diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index c9aefe8..0cc4f8c 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -38,6 +38,12 @@ Resolution validates bindings against extractor specs and records lane-scoped binding metadata. It does not read reference files or include reference bytes in source digests. +During run preparation, resolved file references are materialized before any +LLM-backed pipeline work. Config bindings resolve relative to the config file, +CLI bindings resolve relative to the current working directory, and materialized +reference content is passed to extractors through `ExtractionRequest`. +Reference content is omitted from diagnostics and manifests. + ## Registries And Module Specs `pipeline.Registries` holds concrete constructors for execution. A diff --git a/internal/cli/run.go b/internal/cli/run.go index 90703ff..9996b23 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -184,6 +184,18 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i if err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) } + workingDir, err := os.Getwd() + if err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("resolve working directory: %w", err)) + } + materialized, referenceWarnings, err := pipeline.MaterializeReferences(effective.ResolvedPipeline, catalog, pipeline.ReferenceMaterializationOptions{ + ConfigPath: loadedConfigPath, + WorkingDir: workingDir, + }) + if err != nil { + return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, err) + } + effective.ResolvedPipeline = materialized invocation.PipelineDigest = effective.ResolvedPipeline.Digest if err := runDir.WriteInvocationMetadata(invocation); err != nil { return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics invocation metadata: %w", err)) @@ -225,6 +237,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i StartedAt: startedAt, LLMProfiles: llmProfiles, Metadata: runMetadata(*outputDir, *diagnosticsDir), + Warnings: referenceWarnings, }) if err != nil { if output.Manifest.PipelineID != "" { diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index d26a967..4bb405a 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -741,6 +741,7 @@ func TestRunPipelineLLMProfileOverrideSelectsFactoryProfile(t *testing.T) { func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) { configPath := writeTestConfig(t, testConfigYAML("example", "events")) inputPath := filepath.Join(t.TempDir(), "missing.json") + referencePath := writeFile(t, "roster.yml", "Aria\n") diagnosticsDir := t.TempDir() var stdout bytes.Buffer var stderr bytes.Buffer @@ -750,7 +751,7 @@ func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) { "--config", configPath, "--input", inputPath, "--diagnostics-dir", diagnosticsDir, - "--reference", "roster=./roster.yml", + "--reference", "roster=" + referencePath, }, &stdout, &stderr, Options{ Catalog: fakeCatalog(t, pipeline.ModuleSpec{ Key: "fake/extract", @@ -769,7 +770,7 @@ func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) { resolved := readResolvedPipeline(t, diagnosticsDir) refs := resolved.ArtifactLanes[0].References want := []pipeline.ReferenceBinding{ - {LaneID: "events", SlotName: "roster", Source: "./roster.yml", BindingSource: contracts.ReferenceBindingSourceCLI}, + {LaneID: "events", SlotName: "roster", Source: referencePath, BindingSource: contracts.ReferenceBindingSourceCLI}, } if !reflect.DeepEqual(refs, want) { t.Fatalf("resolved references = %#v, want %#v", refs, want) @@ -779,6 +780,7 @@ func TestRunPipelineReferenceFlagBindsUnambiguousSlot(t *testing.T) { func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) { configPath := writeTestConfig(t, testConfigYAML("example", "events", "notes")) inputPath := filepath.Join(t.TempDir(), "missing.json") + referencePath := writeFile(t, "notes.yml", "Notes\n") diagnosticsDir := t.TempDir() var stdout bytes.Buffer var stderr bytes.Buffer @@ -789,7 +791,7 @@ func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) { "--input", inputPath, "--only", "events,notes", "--diagnostics-dir", diagnosticsDir, - "--reference", "notes.roster=./notes.yml", + "--reference", "notes.roster=" + referencePath, }, &stdout, &stderr, Options{ Catalog: fakeCatalog(t, pipeline.ModuleSpec{ Key: "fake/extract", @@ -811,7 +813,7 @@ func TestRunPipelineReferenceFlagBindsLaneQualifiedSlot(t *testing.T) { t.Fatalf("events references = %#v, want none", events.References) } notes := resolvedArtifactLane(t, resolved, "notes") - if len(notes.References) != 1 || notes.References[0].Source != "./notes.yml" { + if len(notes.References) != 1 || notes.References[0].Source != referencePath { t.Fatalf("notes references = %#v, want lane-qualified binding", notes.References) } } diff --git a/internal/core/config/redaction.go b/internal/core/config/redaction.go index dd9494c..ded3ae8 100644 --- a/internal/core/config/redaction.go +++ b/internal/core/config/redaction.go @@ -50,6 +50,7 @@ func cloneResolvedArtifactLane(in pipeline.ResolvedArtifactLane) pipeline.Resolv out.Merge = cloneModuleBinding(in.Merge) out.Normalize = cloneModuleBinding(in.Normalize) out.References = append([]pipeline.ReferenceBinding(nil), in.References...) + out.ReferenceSet = pipeline.CloneReferenceSet(in.ReferenceSet) if len(in.Validators) > 0 { out.Validators = make([]pipeline.ModuleBinding, len(in.Validators)) for i, binding := range in.Validators { diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index 785b2e3..14a3ec6 100644 --- a/internal/framework/pipeline/profile.go +++ b/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 { diff --git a/internal/framework/pipeline/references.go b/internal/framework/pipeline/references.go new file mode 100644 index 0000000..723c3c3 --- /dev/null +++ b/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 +} diff --git a/internal/framework/pipeline/references_test.go b/internal/framework/pipeline/references_test.go new file mode 100644 index 0000000..35ec15c --- /dev/null +++ b/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) + } +} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 08e7a6e..8ba1d52 100644 --- a/internal/framework/pipeline/runner.go +++ b/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 } diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index e4e6f31..a351686 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/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{