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

Record reference provenance in manifests

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-05 14:44:44 +00:00
parent 9e89b88efc
commit 2f97895732
17 changed files with 447 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
internal/cli/run.go
View File

@@ -206,6 +206,9 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i
if err := runDir.WriteResolvedPipeline(effective.ResolvedPipeline); err != nil {
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics resolved pipeline: %w", err))
}
if err := runDir.WriteResolvedReferences(pipeline.ReferenceProvenance(effective.ResolvedPipeline)); err != nil {
return failPipelineCommand(stderr, runDir, cfg.Diagnostics.Retention, fmt.Errorf("write diagnostics resolved references: %w", err))
}
profileIDs := effectiveLLMProfileIDs(effective.ResolvedPipeline)
if len(profileIDs) != 1 {

240
internal/cli/run_test.go
View File

@@ -15,6 +15,7 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/config"
"gitea.maximumdirect.net/eric/notarius/internal/core/diagnostics"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
"gitea.maximumdirect.net/eric/notarius/internal/modules/chunk/dnd/scenes"
@@ -984,6 +985,74 @@ func TestRunPipelineWritesDurableOutputFiles(t *testing.T) {
assertNoTemporaryFiles(t, runOutputDir)
}
func TestRunPipelineReferenceBytesProduceDistinctManifests(t *testing.T) {
run := func(t *testing.T, referenceText string) artifacts.RunManifest {
t.Helper()
diagnosticsDir := t.TempDir()
outputDir := t.TempDir()
configDir := t.TempDir()
referencePath := filepath.Join(configDir, "roster.txt")
if err := os.WriteFile(referencePath, []byte(referenceText), 0o644); err != nil {
t.Fatalf("write reference: %v", err)
}
configPath := filepath.Join(configDir, "config.yml")
if err := os.WriteFile(configPath, []byte(testConfigYAMLWithReferencesAndDiagnostics("example", "events", diagnosticsDir, map[string]string{"roster": "roster.txt"})), 0o644); err != nil {
t.Fatalf("write config: %v", err)
}
inputPath := writeFile(t, "source.txt", "source text")
var stdout bytes.Buffer
var stderr bytes.Buffer
code := RunWithOptions([]string{"run", "example", "--config", configPath, "--input", inputPath, "--output-dir", outputDir}, &stdout, &stderr, Options{
Catalog: fakeCatalog(t, pipeline.ModuleSpec{
Key: "fake/extract",
Stage: pipeline.StageExtract,
Requires: []string{"chunks"},
Provides: []string{"artifact"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "roster"},
},
}),
Registries: fakeExecutionRegistries(t),
LLMClientFactory: fakeLLMFactory(newFakeRunLLMClient(false), nil),
})
if code != 0 {
t.Fatalf("RunWithOptions() code = %d, stderr=%q", code, stderr.String())
}
var manifest artifacts.RunManifest
readJSONFile(t, filepath.Join(onlyChildDir(t, outputDir), "manifest.json"), &manifest)
if len(manifest.References) != 1 {
t.Fatalf("manifest references = %#v, want one entry", manifest.References)
}
if !reflect.DeepEqual(manifest.SourceDigests, []string{"sha256:source"}) {
t.Fatalf("source digests = %#v, want source-only digest", manifest.SourceDigests)
}
var resolvedReferences []artifacts.ReferenceProvenance
readJSONFile(t, filepath.Join(onlyChildDir(t, diagnosticsDir), diagnostics.ArtifactResolvedReferences), &resolvedReferences)
if !reflect.DeepEqual(resolvedReferences, manifest.References) {
t.Fatalf("resolved references = %#v, want manifest references %#v", resolvedReferences, manifest.References)
}
resolvedReferenceJSON := string(readFile(t, filepath.Join(onlyChildDir(t, diagnosticsDir), diagnostics.ArtifactResolvedReferences)))
if strings.Contains(resolvedReferenceJSON, referenceText) || strings.Contains(resolvedReferenceJSON, "content") {
t.Fatalf("resolved references diagnostics contains content: %s", resolvedReferenceJSON)
}
return manifest
}
first := run(t, "first roster")
second := run(t, "second roster")
if first.References[0].Digest == second.References[0].Digest {
t.Fatalf("reference digests match for different bytes: %q", first.References[0].Digest)
}
if first.PipelineDigest != second.PipelineDigest {
t.Fatalf("pipeline digests differ = %q vs %q, want reference bytes outside pipeline identity", first.PipelineDigest, second.PipelineDigest)
}
}
func TestRunPipelineRejectsUnsafeOutputFileName(t *testing.T) {
diagnosticsDir := t.TempDir()
outputDir := t.TempDir()
@@ -1540,6 +1609,30 @@ func testConfigYAMLWithReferences(pipelineID string, laneID string, references m
return b.String()
}
func testConfigYAMLWithReferencesAndDiagnostics(pipelineID string, laneID string, diagnosticsDir string, references map[string]string) string {
var b strings.Builder
b.WriteString("version: 1\n")
b.WriteString("diagnostics:\n")
b.WriteString(" work_dir: " + diagnosticsDir + "\n")
b.WriteString(" retention: always\n")
b.WriteString("pipelines:\n")
b.WriteString(" " + pipelineID + ":\n")
b.WriteString(" input: fake/input\n")
b.WriteString(" artifacts:\n")
b.WriteString(" " + laneID + ":\n")
b.WriteString(" extract: fake/extract\n")
b.WriteString(" references:\n")
keys := make([]string, 0, len(references))
for key := range references {
keys = append(keys, key)
}
sort.Strings(keys)
for _, key := range keys {
b.WriteString(" " + key + ": " + references[key] + "\n")
}
return b.String()
}
func mvpConfigYAML(pipelineID string, extractor string) string {
return `version: 1
pipelines:
@@ -1788,6 +1881,153 @@ func registriesWithOutput(t *testing.T, encoder contracts.OutputEncoder) pipelin
return registries
}
func fakeExecutionRegistries(t *testing.T) pipeline.Registries {
t.Helper()
inputs := pipeline.NewInputAdapterRegistry()
chunkers := pipeline.NewChunkerRegistry()
extractors := pipeline.NewExtractorRegistry()
mergers := pipeline.NewMergerRegistry()
normalizers := pipeline.NewNormalizerRegistry()
outputs := pipeline.NewOutputEncoderRegistry()
if err := inputs.RegisterWithSpec(pipeline.ModuleSpec{Key: "fake/input", Stage: pipeline.StageInput, Provides: []string{"source"}}, func() (contracts.InputAdapter, error) {
return fakeRunInputAdapter{}, nil
}); err != nil {
t.Fatalf("register fake input: %v", err)
}
if err := chunkers.RegisterWithSpec(pipeline.ModuleSpec{Key: "generic", Stage: pipeline.StageChunk, Requires: []string{"source"}, Provides: []string{"chunks"}}, func() (contracts.Chunker, error) {
return fakeRunChunker{}, nil
}); err != nil {
t.Fatalf("register fake chunker: %v", err)
}
if err := extractors.RegisterWithSpec(pipeline.ModuleSpec{
Key: "fake/extract",
Stage: pipeline.StageExtract,
Requires: []string{"chunks"},
Provides: []string{"artifact"},
ReferenceSlots: []contracts.ReferenceSlot{
{Name: "roster"},
},
}, func() (contracts.Extractor, error) {
return fakeRunExtractor{}, nil
}); err != nil {
t.Fatalf("register fake extractor: %v", err)
}
if err := mergers.RegisterWithSpec(pipeline.ModuleSpec{Key: "appendorder", Stage: pipeline.StageMerge, Requires: []string{"artifact"}, Provides: []string{"merged"}}, func() (contracts.Merger, error) {
return fakeRunMerger{}, nil
}); err != nil {
t.Fatalf("register fake merger: %v", err)
}
if err := normalizers.RegisterWithSpec(pipeline.ModuleSpec{Key: "noop", Stage: pipeline.StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}}, func() (contracts.Normalizer, error) {
return fakeRunNormalizer{}, nil
}); err != nil {
t.Fatalf("register fake normalizer: %v", err)
}
if err := jsonoutput.Register(outputs); err != nil {
t.Fatalf("register json output: %v", err)
}
return pipeline.Registries{
Inputs: inputs,
Chunkers: chunkers,
Extractors: extractors,
Mergers: mergers,
Normalizers: normalizers,
Outputs: outputs,
}
}
type fakeRunInputAdapter struct{}
func (fakeRunInputAdapter) Key() string {
return "fake/input"
}
func (fakeRunInputAdapter) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
return &source.SourceDocument{
ID: "source",
Kind: "text",
Format: "test",
Digest: "sha256:source",
Units: []source.SourceUnit{
{ID: "unit-1", Kind: "text", Text: string(req.Raw)},
},
}, nil
}
type fakeRunChunker struct{}
func (fakeRunChunker) Key() string {
return "generic"
}
func (fakeRunChunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) {
return contracts.ChunkResult{
Chunks: []contracts.SourceChunk{
{ID: "chunk-1", SourceID: req.Source.ID, Index: 0, Units: req.Source.Units},
},
}, nil
}
type fakeRunExtractor struct{}
func (fakeRunExtractor) Key() string {
return "fake/extract"
}
func (fakeRunExtractor) ArtifactType() string {
return "fake.artifact"
}
func (fakeRunExtractor) SchemaVersion() string {
return "v1"
}
func (fakeRunExtractor) ReferenceSlots() []contracts.ReferenceSlot {
return []contracts.ReferenceSlot{{Name: "roster"}}
}
func (fakeRunExtractor) Validators() []contracts.Validator {
return nil
}
func (fakeRunExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) {
return contracts.ExtractionResult{
Candidates: []artifacts.ArtifactCandidate{
{
Payload: []byte(`{"value":true}`),
SourceRefs: []source.SourceRef{
{SourceID: "source", StartUnitID: "unit-1", EndUnitID: "unit-1"},
},
},
},
}, nil
}
type fakeRunMerger struct{}
func (fakeRunMerger) Key() string {
return "appendorder"
}
func (fakeRunMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) {
var candidates []artifacts.ArtifactCandidate
for _, chunkArtifacts := range req.ChunkArtifacts {
candidates = append(candidates, chunkArtifacts.Candidates...)
}
return contracts.MergeResult{Candidates: candidates}, nil
}
type fakeRunNormalizer struct{}
func (fakeRunNormalizer) Key() string {
return "noop"
}
func (fakeRunNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
return contracts.NormalizeResult{Candidates: append([]artifacts.ArtifactCandidate(nil), req.Candidates...)}, nil
}
func onlyChildDir(t *testing.T, root string) string {
t.Helper()
children := childDirs(t, root)

12
internal/core/artifacts/artifacts.go
View File

@@ -48,6 +48,17 @@ type LLMProfileManifest struct {
Model string `json:"model,omitempty"`
}
type ReferenceProvenance struct {
LaneID string `json:"lane_id"`
SlotName string `json:"slot_name"`
OriginType string `json:"origin_type"`
OriginURI string `json:"origin_uri,omitempty"`
Digest string `json:"digest,omitempty"`
MediaType string `json:"media_type,omitempty"`
SizeBytes int64 `json:"size_bytes,omitempty"`
BindingSource string `json:"binding_source,omitempty"`
}
type RunManifest struct {
RunID string `json:"run_id,omitempty"`
PipelineID string `json:"pipeline_id,omitempty"`
@@ -61,6 +72,7 @@ type RunManifest struct {
OutputEncoder string `json:"output_encoder,omitempty"`
ModuleMetadata map[string]map[string]any `json:"module_metadata,omitempty"`
ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"`
References []ReferenceProvenance `json:"references,omitempty"`
LLMProfiles []LLMProfileManifest `json:"llm_profiles,omitempty"`
SchemaVersion string `json:"schema_version,omitempty"`
ValidationStatus string `json:"validation_status,omitempty"`

37
internal/core/artifacts/artifacts_test.go
View File

@@ -183,6 +183,43 @@ func TestRunManifestIncludesPipelineAndArtifactLaneFields(t *testing.T) {
assertHasKeys(t, lane, "id", "extractor", "merger", "normalizer", "validators", "metadata")
}
func TestRunManifestIncludesReferenceProvenance(t *testing.T) {
manifest := RunManifest{
References: []ReferenceProvenance{
{
LaneID: "events",
SlotName: "roster",
OriginType: "file",
OriginURI: "file:///tmp/roster.txt",
Digest: "sha256:reference",
MediaType: "text/plain; charset=utf-8",
SizeBytes: 12,
BindingSource: "config",
},
},
}
gotJSON, err := json.Marshal(manifest)
if err != nil {
t.Fatalf("json.Marshal() error = %v", err)
}
var got RunManifest
if err := json.Unmarshal(gotJSON, &got); err != nil {
t.Fatalf("json.Unmarshal() error = %v", err)
}
if len(got.References) != 1 {
t.Fatalf("len(References) = %d, want 1", len(got.References))
}
reference := got.References[0]
if reference.LaneID != "events" || reference.SlotName != "roster" || reference.OriginType != "file" || reference.OriginURI != "file:///tmp/roster.txt" {
t.Fatalf("reference provenance = %#v, want lane-scoped origin details", reference)
}
if reference.Digest != "sha256:reference" || reference.MediaType != "text/plain; charset=utf-8" || reference.SizeBytes != 12 || reference.BindingSource != "config" {
t.Fatalf("reference provenance = %#v, want digest/media/size/source details", reference)
}
}
func TestRunManifestIncludesTopLevelModuleMetadata(t *testing.T) {
manifest := RunManifest{
ModuleMetadata: map[string]map[string]any{

1
internal/core/diagnostics/artifacts.go
View File

@@ -4,6 +4,7 @@ const (
ArtifactInvocationMetadata = "invocation.json"
ArtifactEffectiveConfig = "effective-config.json"
ArtifactResolvedPipeline = "resolved-pipeline.json"
ArtifactResolvedReferences = "resolved-references.json"
ArtifactSourceDocument = "source-document.json"
ArtifactRunManifest = "run-manifest.json"
ArtifactRunReport = "run-report.json"

1
internal/core/diagnostics/artifacts_test.go
View File

@@ -7,6 +7,7 @@ func TestArtifactNamesUseExtractionOrientedNames(t *testing.T) {
ArtifactInvocationMetadata,
ArtifactEffectiveConfig,
ArtifactResolvedPipeline,
ArtifactResolvedReferences,
ArtifactSourceDocument,
ArtifactRunManifest,
ArtifactRunReport,

4
internal/core/diagnostics/run_dir.go
View File

@@ -149,6 +149,10 @@ func (r *RunDirectory) WriteResolvedPipeline(payload any) error {
return r.WriteJSONArtifact(ArtifactResolvedPipeline, payload)
}
func (r *RunDirectory) WriteResolvedReferences(payload any) error {
return r.WriteJSONArtifact(ArtifactResolvedReferences, payload)
}
func (r *RunDirectory) WriteSourceDocument(payload any) error {
return r.WriteJSONArtifact(ArtifactSourceDocument, payload)
}

4
internal/core/diagnostics/run_dir_test.go
View File

@@ -191,6 +191,9 @@ func TestWriteTypedArtifacts(t *testing.T) {
if err := runDir.WriteResolvedPipeline(map[string]any{"pipeline": "test"}); err != nil {
t.Fatalf("WriteResolvedPipeline: %v", err)
}
if err := runDir.WriteResolvedReferences([]artifacts.ReferenceProvenance{{LaneID: "events", SlotName: "roster"}}); err != nil {
t.Fatalf("WriteResolvedReferences: %v", err)
}
if err := runDir.WriteSourceDocument(map[string]any{"source_id": "source-1"}); err != nil {
t.Fatalf("WriteSourceDocument: %v", err)
}
@@ -207,6 +210,7 @@ func TestWriteTypedArtifacts(t *testing.T) {
for _, name := range []string{
ArtifactEffectiveConfig,
ArtifactResolvedPipeline,
ArtifactResolvedReferences,
ArtifactSourceDocument,
ArtifactRunManifest,
ArtifactRunReport,

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

@@ -11,6 +11,7 @@ import (
"strings"
"unicode/utf8"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
@@ -185,3 +186,33 @@ func CloneReferenceSet(in contracts.ReferenceSet) contracts.ReferenceSet {
}
return out
}
func ReferenceProvenance(resolved ResolvedPipeline) []artifacts.ReferenceProvenance {
provenance := []artifacts.ReferenceProvenance{}
for _, lane := range resolved.ArtifactLanes {
if len(lane.ReferenceSet.Slots) == 0 {
continue
}
slotNames := make([]string, 0, len(lane.ReferenceSet.Slots))
for slotName := range lane.ReferenceSet.Slots {
slotNames = append(slotNames, slotName)
}
sort.Strings(slotNames)
for _, slotName := range slotNames {
slot := lane.ReferenceSet.Slots[slotName]
for _, item := range slot.Items {
provenance = append(provenance, artifacts.ReferenceProvenance{
LaneID: lane.ID,
SlotName: item.SlotName,
OriginType: item.Origin.Type,
OriginURI: item.Origin.URI,
Digest: item.Digest,
MediaType: item.MediaType,
SizeBytes: item.SizeBytes,
BindingSource: item.BindingSource,
})
}
}
}
return provenance
}

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

@@ -83,6 +83,17 @@ func TestMaterializeReferencesResolvesPathsAndDigestsContent(t *testing.T) {
t.Fatalf("glossary origin URI = %q, want cli reference path", glossary.Origin.URI)
}
provenance := ReferenceProvenance(first)
if len(provenance) != 2 {
t.Fatalf("ReferenceProvenance() = %#v, want two entries", provenance)
}
if provenance[0].LaneID != "events" || provenance[0].SlotName != "glossary" || provenance[0].Digest != glossary.Digest {
t.Fatalf("ReferenceProvenance()[0] = %#v, want sorted glossary provenance", provenance[0])
}
if provenance[1].LaneID != "events" || provenance[1].SlotName != "roster" || provenance[1].Digest != roster.Digest {
t.Fatalf("ReferenceProvenance()[1] = %#v, want roster provenance", provenance[1])
}
encoded, err := json.Marshal(first)
if err != nil {
t.Fatalf("json.Marshal(materialized) error = %v, want nil", err)

4
internal/framework/pipeline/runner.go
View File

@@ -356,8 +356,12 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest {
ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)),
RunID: runID,
StartedAt: timePtr(startedAt),
References: ReferenceProvenance(pipeline),
LLMProfiles: cloneLLMProfiles(input.LLMProfiles),
}
// The runner does not currently maintain a cache or idempotency key. Reference
// digests are recorded in manifest provenance and intentionally kept separate
// from source_digests.
for _, lane := range pipeline.ArtifactLanes {
laneManifest := artifacts.ArtifactLaneManifest{

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

@@ -960,7 +960,27 @@ func TestRunReturnsFailedManifestWhenOutputEncoderFails(t *testing.T) {
}
func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) {
output, err := New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: resolvedPipelineWithValidators("configured")})
resolved := resolvedPipelineWithValidators("configured")
resolved.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 content"),
Digest: "sha256:reference",
Origin: contracts.ReferenceOrigin{Type: "file", URI: "file:///tmp/roster.txt"},
SizeBytes: int64(len("reference content")),
BindingSource: contracts.ReferenceBindingSourceConfig,
},
},
},
},
}
output, err := New(newRunnerRegistries(t, nil)).Run(context.Background(), RunInput{Pipeline: resolved})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
@@ -975,6 +995,16 @@ 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)
}
reference := manifest.References[0]
if reference.LaneID != "alpha" || reference.SlotName != "roster" || reference.Digest != "sha256:reference" {
t.Fatalf("reference provenance = %#v, want lane slot digest", reference)
}
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 manifest.ValidationStatus != "approved" {
t.Fatalf("ValidationStatus = %q, want approved", manifest.ValidationStatus)
}

36
internal/modules/output/json/encoder_test.go
View File

@@ -142,6 +142,42 @@ func TestEncodePrettyPrintsJSON(t *testing.T) {
}
}
func TestEncodeIncludesManifestReferences(t *testing.T) {
result, err := New().Encode(context.Background(), contracts.OutputRequest{
Manifest: artifacts.RunManifest{
RunID: "run-1",
References: []artifacts.ReferenceProvenance{
{
LaneID: "events",
SlotName: "roster",
OriginType: "file",
OriginURI: "file:///tmp/roster.txt",
Digest: "sha256:reference",
MediaType: "text/plain; charset=utf-8",
SizeBytes: 12,
BindingSource: "config",
},
},
},
})
if err != nil {
t.Fatalf("Encode() error = %v, want nil", err)
}
manifest := decodeObject(t, fileBytes(t, result.Files, "manifest.json"))
references := manifest["references"].([]any)
if len(references) != 1 {
t.Fatalf("references = %#v, want one entry", references)
}
reference := references[0].(map[string]any)
if reference["lane_id"] != "events" || reference["slot_name"] != "roster" || reference["digest"] != "sha256:reference" {
t.Fatalf("reference manifest = %#v, want lane slot digest", reference)
}
if _, ok := reference["content"]; ok {
t.Fatalf("reference manifest = %#v, want no content field", reference)
}
}
func TestEncodeRejectsArtifactTypeWithoutSafeFileName(t *testing.T) {
_, err := New().Encode(context.Background(), contracts.OutputRequest{
Approved: []artifacts.Artifact{artifact("///", "unsafe")},