eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Materialize references for all eligible targets

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-05 16:35:24 +00:00
parent 8c623b7ad8
commit 4cafde2502
8 changed files with 316 additions and 52 deletions
Download Patch File Download Diff File Expand all files Collapse all files

22
internal/framework/pipeline/references.go
View File

@@ -30,23 +30,35 @@ type ReferenceMaterializationOptions struct {
func MaterializeReferences(resolved ResolvedPipeline, catalog ModuleCatalog, options ReferenceMaterializationOptions) (ResolvedPipeline, []contracts.Warning, error) {
out := resolved
out.ChunkReferences = CloneReferenceTarget(resolved.ChunkReferences)
chunkReferenceSet, chunkWarnings, err := materializeReferenceTarget(resolved.ID, resolved.ChunkReferences, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
out.ChunkReferences.ReferenceSet = chunkReferenceSet
warnings := append([]contracts.Warning(nil), chunkWarnings...)
if len(resolved.ArtifactLanes) == 0 {
return out, nil, nil
return out, warnings, nil
}
warnings := []contracts.Warning(nil)
out.ArtifactLanes = make([]ResolvedArtifactLane, len(resolved.ArtifactLanes))
for i, lane := range resolved.ArtifactLanes {
materializedLane := lane
materializedLane.ExtractReferences = CloneReferenceTarget(lane.ExtractReferences)
materializedLane.NormalizeReferences = CloneReferenceTarget(lane.NormalizeReferences)
referenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options)
extractReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.ExtractReferences, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
materializedLane.ExtractReferences.ReferenceSet = referenceSet
out.ArtifactLanes[i] = materializedLane
materializedLane.ExtractReferences.ReferenceSet = extractReferenceSet
warnings = append(warnings, laneWarnings...)
normalizeReferenceSet, laneWarnings, err := materializeReferenceTarget(resolved.ID, lane.NormalizeReferences, catalog, options)
if err != nil {
return ResolvedPipeline{}, nil, err
}
materializedLane.NormalizeReferences.ReferenceSet = normalizeReferenceSet
warnings = append(warnings, laneWarnings...)
out.ArtifactLanes[i] = materializedLane
}
return out, warnings, nil
}

222
internal/framework/pipeline/references_test.go
View File

@@ -4,6 +4,8 @@ import (
"encoding/json"
"os"
"path/filepath"
"reflect"
"sort"
"strings"
"testing"
@@ -103,6 +105,66 @@ func TestMaterializeReferencesResolvesPathsAndDigestsContent(t *testing.T) {
}
}
func TestMaterializeReferencesStoresSetsAndProvenanceForAllTargets(t *testing.T) {
configDir := t.TempDir()
writeReferenceFile(t, filepath.Join(configDir, "chunk.txt"), []byte("chunk text"))
writeReferenceFile(t, filepath.Join(configDir, "extract.txt"), []byte("extract text"))
writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), []byte("normalize text"))
profile := baselineProfile()
profile.References = map[string]string{
"scene_guide": "chunk.txt",
"roster": "extract.txt",
"normalization_notes": "normalize.txt",
}
catalog := referenceCatalogForTargets(t,
[]contracts.ReferenceSlot{{Name: "scene_guide"}},
[]contracts.ReferenceSlot{{Name: "roster"}},
[]contracts.ReferenceSlot{{Name: "normalization_notes"}},
)
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
materialized, warnings, err := MaterializeReferences(resolved, catalog, ReferenceMaterializationOptions{
ConfigPath: filepath.Join(configDir, "config.yml"),
})
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
if len(warnings) != 0 {
t.Fatalf("warnings = %#v, want none", warnings)
}
chunkItem := materialized.ChunkReferences.ReferenceSet.Slots["scene_guide"].Items[0]
if string(chunkItem.Content) != "chunk text" {
t.Fatalf("chunk content = %q, want chunk text", chunkItem.Content)
}
extractItem := materialized.ArtifactLanes[0].ExtractReferences.ReferenceSet.Slots["roster"].Items[0]
if string(extractItem.Content) != "extract text" {
t.Fatalf("extract content = %q, want extract text", extractItem.Content)
}
normalizeItem := materialized.ArtifactLanes[0].NormalizeReferences.ReferenceSet.Slots["normalization_notes"].Items[0]
if string(normalizeItem.Content) != "normalize text" {
t.Fatalf("normalize content = %q, want normalize text", normalizeItem.Content)
}
provenance := ReferenceProvenance(materialized)
if len(provenance) != 3 {
t.Fatalf("ReferenceProvenance() = %#v, want three entries", provenance)
}
if provenance[0].Stage != string(StageChunk) || provenance[0].LaneID != "" || provenance[0].SlotName != "scene_guide" || provenance[0].Digest != chunkItem.Digest {
t.Fatalf("ReferenceProvenance()[0] = %#v, want chunk scene guide provenance", provenance[0])
}
if provenance[1].Stage != string(StageExtract) || provenance[1].LaneID != "events" || provenance[1].SlotName != "roster" || provenance[1].Digest != extractItem.Digest {
t.Fatalf("ReferenceProvenance()[1] = %#v, want extract roster provenance", provenance[1])
}
if provenance[2].Stage != string(StageNormalize) || provenance[2].LaneID != "events" || provenance[2].SlotName != "normalization_notes" || provenance[2].Digest != normalizeItem.Digest {
t.Fatalf("ReferenceProvenance()[2] = %#v, want normalize notes provenance", provenance[2])
}
}
func TestMaterializeReferencesRejectsNonUTF8Content(t *testing.T) {
configDir := t.TempDir()
path := filepath.Join(configDir, "bad.txt")
@@ -117,6 +179,20 @@ func TestMaterializeReferencesRejectsNonUTF8Content(t *testing.T) {
}
}
func TestMaterializeReferencesRejectsNonUTF8ContentForChunkTarget(t *testing.T) {
configDir := t.TempDir()
path := filepath.Join(configDir, "bad.txt")
writeReferenceFile(t, path, []byte{0xff, 0xfe})
resolved := resolvedPipelineWithTargetReference(t, StageChunk, "", "scene_guide", "bad.txt", contracts.ReferenceBindingSourceConfig, contracts.ReferenceSlot{Name: "scene_guide"})
_, _, err := MaterializeReferences(resolved, referenceCatalogForTargets(t, []contracts.ReferenceSlot{{Name: "scene_guide"}}, nil, nil), ReferenceMaterializationOptions{
ConfigPath: filepath.Join(configDir, "config.yml"),
})
if err == nil || !strings.Contains(err.Error(), "chunk") || !strings.Contains(err.Error(), "UTF-8") || !strings.Contains(err.Error(), "scene_guide") || !strings.Contains(err.Error(), path) {
t.Fatalf("error = %v, want chunk UTF-8 path error", err)
}
}
func TestMaterializeReferencesAllowsAnyMediaTypeWhenSlotDoesNotRestrictIt(t *testing.T) {
configDir := t.TempDir()
path := filepath.Join(configDir, "roster.reference")
@@ -207,6 +283,21 @@ func TestMaterializeReferencesMatchesAcceptedMediaTypesIgnoringParameters(t *tes
}
}
func TestMaterializeReferencesRejectsUnacceptedMediaTypeForNormalizeTarget(t *testing.T) {
configDir := t.TempDir()
path := filepath.Join(configDir, "notes.json")
writeReferenceFile(t, path, []byte(`{"notes":true}`))
slot := contracts.ReferenceSlot{Name: "normalization_notes", AcceptedMediaTypes: []string{"text/markdown"}}
resolved := resolvedPipelineWithTargetReference(t, StageNormalize, "events", "normalization_notes", "notes.json", contracts.ReferenceBindingSourceConfig, slot)
_, _, err := MaterializeReferences(resolved, referenceCatalogForTargets(t, nil, nil, []contracts.ReferenceSlot{slot}), ReferenceMaterializationOptions{
ConfigPath: filepath.Join(configDir, "config.yml"),
})
if err == nil || !strings.Contains(err.Error(), "normalize") || !strings.Contains(err.Error(), "media type") || !strings.Contains(err.Error(), "application/json") || !strings.Contains(err.Error(), "normalization_notes") {
t.Fatalf("error = %v, want normalize media type rejection", err)
}
}
func TestMaterializeReferencesWarnsForEmptyFiles(t *testing.T) {
configDir := t.TempDir()
path := filepath.Join(configDir, "empty.txt")
@@ -228,6 +319,45 @@ func TestMaterializeReferencesWarnsForEmptyFiles(t *testing.T) {
}
}
func TestMaterializeReferencesWarningScopesIncludeTargetContext(t *testing.T) {
configDir := t.TempDir()
writeReferenceFile(t, filepath.Join(configDir, "chunk.txt"), nil)
writeReferenceFile(t, filepath.Join(configDir, "extract.txt"), nil)
writeReferenceFile(t, filepath.Join(configDir, "normalize.txt"), nil)
profile := baselineProfile()
profile.References = map[string]string{
"scene_guide": "chunk.txt",
"roster": "extract.txt",
"normalization_notes": "normalize.txt",
}
catalog := referenceCatalogForTargets(t,
[]contracts.ReferenceSlot{{Name: "scene_guide"}},
[]contracts.ReferenceSlot{{Name: "roster"}},
[]contracts.ReferenceSlot{{Name: "normalization_notes"}},
)
resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
_, warnings, err := MaterializeReferences(resolved, catalog, ReferenceMaterializationOptions{
ConfigPath: filepath.Join(configDir, "config.yml"),
})
if err != nil {
t.Fatalf("MaterializeReferences() error = %v, want nil", err)
}
got := warningScopes(warnings)
want := []string{
"pipeline.baseline.chunk.reference.scene_guide",
"pipeline.baseline.lane.events.extract.reference.roster",
"pipeline.baseline.lane.events.normalize.reference.normalization_notes",
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("warning scopes = %#v, want %#v", got, want)
}
}
func TestMaterializeReferencesEnforcesMaxBytes(t *testing.T) {
configDir := t.TempDir()
path := filepath.Join(configDir, "large.txt")
@@ -244,31 +374,99 @@ func TestMaterializeReferencesEnforcesMaxBytes(t *testing.T) {
}
func resolvedPipelineWithReference(t *testing.T, slotName, source, bindingSource string, slot contracts.ReferenceSlot) ResolvedPipeline {
t.Helper()
return resolvedPipelineWithTargetReference(t, StageExtract, "events", slotName, source, bindingSource, slot)
}
func resolvedPipelineWithTargetReference(t *testing.T, stage ModuleStage, laneID string, 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})
switch stage {
case StageChunk:
profile.Chunk.References = map[string]string{slotName: source}
case StageExtract:
lane := profile.Artifacts[laneID]
lane.References = map[string]string{slotName: source}
profile.Artifacts[laneID] = lane
case StageNormalize:
lane := profile.Artifacts[laneID]
lane.Normalize.References = map[string]string{slotName: source}
profile.Artifacts[laneID] = lane
default:
t.Fatalf("unsupported reference target stage %q", stage)
}
catalog := referenceCatalogForStage(t, stage, []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].ExtractReferences.Bindings[0].BindingSource = bindingSource
switch stage {
case StageChunk:
resolved.ChunkReferences.Bindings[0].BindingSource = bindingSource
case StageExtract:
resolved.ArtifactLanes[0].ExtractReferences.Bindings[0].BindingSource = bindingSource
case StageNormalize:
resolved.ArtifactLanes[0].NormalizeReferences.Bindings[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,
})
return referenceCatalogForStage(t, StageExtract, slots)
}
func referenceCatalogForStage(t *testing.T, stage ModuleStage, slots []contracts.ReferenceSlot) ModuleCatalog {
t.Helper()
switch stage {
case StageChunk:
return referenceCatalogForTargets(t, slots, nil, nil)
case StageExtract:
return referenceCatalogForTargets(t, nil, slots, nil)
case StageNormalize:
return referenceCatalogForTargets(t, nil, nil, slots)
default:
t.Fatalf("unsupported reference target stage %q", stage)
return ModuleCatalog{}
}
}
func referenceCatalogForTargets(t *testing.T, chunkSlots, extractSlots, normalizeSlots []contracts.ReferenceSlot) ModuleCatalog {
t.Helper()
return newProfileCatalogWithOverrides(t,
ModuleSpec{
Key: "generic",
Stage: StageChunk,
Requires: []string{"source"},
Provides: []string{"chunk"},
ReferenceSlots: chunkSlots,
},
ModuleSpec{
Key: "event-extractor",
Stage: StageExtract,
Requires: []string{"chunk"},
Provides: []string{"candidate"},
ReferenceSlots: extractSlots,
},
ModuleSpec{
Key: "noop",
Stage: StageNormalize,
Requires: []string{"merged"},
Provides: []string{"normalized"},
ReferenceSlots: normalizeSlots,
},
)
}
func warningScopes(warnings []contracts.Warning) []string {
scopes := make([]string, 0, len(warnings))
for _, warning := range warnings {
scopes = append(scopes, warning.Scope)
}
sort.Strings(scopes)
return scopes
}
func writeReferenceFile(t *testing.T, path string, content []byte) {

61
internal/framework/pipeline/runner_test.go
View File

@@ -965,6 +965,24 @@ func TestRunReturnsFailedManifestWhenOutputEncoderFails(t *testing.T) {
func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
resolved := resolvedPipelineWithValidators("configured")
resolved.ChunkReferences.ReferenceSet = contracts.ReferenceSet{
Slots: map[string]contracts.ResolvedReferenceSlot{
"scene_guide": {
Slot: contracts.ReferenceSlot{Name: "scene_guide"},
Items: []contracts.ReferenceItem{
{
SlotName: "scene_guide",
MediaType: "text/plain; charset=utf-8",
Content: []byte("chunk reference content"),
Digest: "sha256:chunk-reference",
Origin: contracts.ReferenceOrigin{Type: "file", URI: "file:///tmp/scene-guide.txt"},
SizeBytes: int64(len("chunk reference content")),
BindingSource: contracts.ReferenceBindingSourceCLI,
},
},
},
},
}
resolved.ArtifactLanes[0].ExtractReferences.ReferenceSet = contracts.ReferenceSet{
Slots: map[string]contracts.ResolvedReferenceSlot{
"roster": {
@@ -983,6 +1001,24 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
},
},
}
resolved.ArtifactLanes[0].NormalizeReferences.ReferenceSet = contracts.ReferenceSet{
Slots: map[string]contracts.ResolvedReferenceSlot{
"normalization_notes": {
Slot: contracts.ReferenceSlot{Name: "normalization_notes"},
Items: []contracts.ReferenceItem{
{
SlotName: "normalization_notes",
MediaType: "text/plain; charset=utf-8",
Content: []byte("normalize reference content"),
Digest: "sha256:normalize-reference",
Origin: contracts.ReferenceOrigin{Type: "file", URI: "file:///tmp/normalize.txt"},
SizeBytes: int64(len("normalize reference content")),
BindingSource: contracts.ReferenceBindingSourceConfig,
},
},
},
},
}
output, err := New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: resolved})
if err != nil {
@@ -999,15 +1035,26 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
if !reflect.DeepEqual(manifest.SourceDigests, []string{"sha256:source"}) {
t.Fatalf("SourceDigests = %#v, want source digest", manifest.SourceDigests)
}
if len(manifest.References) != 1 {
t.Fatalf("References = %#v, want one reference provenance entry", manifest.References)
if len(manifest.References) != 3 {
t.Fatalf("References = %#v, want three reference provenance entries", manifest.References)
}
reference := manifest.References[0]
if reference.Stage != string(StageExtract) || reference.LaneID != "alpha" || reference.SlotName != "roster" || reference.Digest != "sha256:reference" {
t.Fatalf("reference provenance = %#v, want lane slot digest", reference)
chunkReference := manifest.References[0]
if chunkReference.Stage != string(StageChunk) || chunkReference.LaneID != "" || chunkReference.SlotName != "scene_guide" || chunkReference.Digest != "sha256:chunk-reference" {
t.Fatalf("chunk reference provenance = %#v, want chunk slot digest", chunkReference)
}
if reference.OriginType != "file" || reference.OriginURI != "file:///tmp/roster.txt" || reference.MediaType != "text/plain; charset=utf-8" || reference.SizeBytes != int64(len("reference content")) || reference.BindingSource != contracts.ReferenceBindingSourceConfig {
t.Fatalf("reference provenance = %#v, want origin/media/size/source", reference)
if chunkReference.OriginType != "file" || chunkReference.OriginURI != "file:///tmp/scene-guide.txt" || chunkReference.MediaType != "text/plain; charset=utf-8" || chunkReference.SizeBytes != int64(len("chunk reference content")) || chunkReference.BindingSource != contracts.ReferenceBindingSourceCLI {
t.Fatalf("chunk reference provenance = %#v, want origin/media/size/source", chunkReference)
}
extractReference := manifest.References[1]
if extractReference.Stage != string(StageExtract) || extractReference.LaneID != "alpha" || extractReference.SlotName != "roster" || extractReference.Digest != "sha256:reference" {
t.Fatalf("extract reference provenance = %#v, want lane slot digest", extractReference)
}
if extractReference.OriginType != "file" || extractReference.OriginURI != "file:///tmp/roster.txt" || extractReference.MediaType != "text/plain; charset=utf-8" || extractReference.SizeBytes != int64(len("reference content")) || extractReference.BindingSource != contracts.ReferenceBindingSourceConfig {
t.Fatalf("extract reference provenance = %#v, want origin/media/size/source", extractReference)
}
normalizeReference := manifest.References[2]
if normalizeReference.Stage != string(StageNormalize) || normalizeReference.LaneID != "alpha" || normalizeReference.SlotName != "normalization_notes" || normalizeReference.Digest != "sha256:normalize-reference" {
t.Fatalf("normalize reference provenance = %#v, want lane slot digest", normalizeReference)
}
if manifest.ValidationStatus != "approved" {
t.Fatalf("ValidationStatus = %q, want approved", manifest.ValidationStatus)