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

Make chunk validation a framework contract

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-04 12:52:16 +00:00
parent b95af4f87d
commit e5eb0ba5c8
8 changed files with 249 additions and 4 deletions
Download Patch File Download Diff File Expand all files Collapse all files

60
internal/framework/pipeline/chunk_validation.go Normal file
View File

@@ -0,0 +1,60 @@
package pipeline
import (
"fmt"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func validateChunkResult(doc *source.SourceDocument, chunks []contracts.SourceChunk) error {
sourceUnitIndexes := make(map[string]int, len(doc.Units))
for index, unit := range doc.Units {
sourceUnitIndexes[unit.ID] = index
}
seenChunkIDs := make(map[string]struct{}, len(chunks))
for chunkIndex, chunk := range chunks {
if strings.TrimSpace(chunk.ID) == "" {
return fmt.Errorf("chunk[%d].id must not be empty", chunkIndex)
}
if _, ok := seenChunkIDs[chunk.ID]; ok {
return fmt.Errorf("chunk id %q is duplicated", chunk.ID)
}
seenChunkIDs[chunk.ID] = struct{}{}
if chunk.SourceID != doc.ID {
return fmt.Errorf("chunk %q source_id %q does not match source document id %q", chunk.ID, chunk.SourceID, doc.ID)
}
if chunk.Index != chunkIndex {
return fmt.Errorf("chunk %q index %d does not match returned order %d", chunk.ID, chunk.Index, chunkIndex)
}
if len(chunk.Units) == 0 {
return fmt.Errorf("chunk %q units must not be empty", chunk.ID)
}
seenUnitIDs := make(map[string]struct{}, len(chunk.Units))
previousSourceIndex := -1
for unitIndex, unit := range chunk.Units {
if strings.TrimSpace(unit.ID) == "" {
return fmt.Errorf("chunk %q unit[%d].id must not be empty", chunk.ID, unitIndex)
}
if _, ok := seenUnitIDs[unit.ID]; ok {
return fmt.Errorf("chunk %q repeats source unit %q", chunk.ID, unit.ID)
}
seenUnitIDs[unit.ID] = struct{}{}
sourceIndex, ok := sourceUnitIndexes[unit.ID]
if !ok {
return fmt.Errorf("chunk %q source unit %q was not found in source document %q", chunk.ID, unit.ID, doc.ID)
}
if sourceIndex <= previousSourceIndex {
return fmt.Errorf("chunk %q source units must appear in source document order", chunk.ID)
}
previousSourceIndex = sourceIndex
}
}
return nil
}

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

@@ -91,6 +91,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
}
chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{
Source: doc,
LLMClient: input.LLMClient,
LLMProfile: input.Pipeline.Chunk.LLMProfile,
Options: cloneOptions(input.Pipeline.Chunk.Options),
Metadata: input.Metadata,
@@ -102,6 +103,9 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
if len(chunkResult.Chunks) == 0 {
return failOutput(output), fmt.Errorf("chunker %q returned no chunks", chunker.Key())
}
if err := validateChunkResult(doc, chunkResult.Chunks); err != nil {
return failOutput(output), fmt.Errorf("validate chunks from chunker %q: %w", chunker.Key(), err)
}
nextCandidateIndex := 0
for _, lane := range input.Pipeline.ArtifactLanes {

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

@@ -256,6 +256,91 @@ func TestRunRejectsChunkerBuildChunkAndEmptyChunkErrors(t *testing.T) {
}
}
func TestRunRejectsInvalidChunks(t *testing.T) {
tests := []struct {
name string
chunks []contracts.SourceChunk
want string
}{
{
name: "empty chunk id",
chunks: []contracts.SourceChunk{{ID: "", SourceID: "source-1", Index: 0, Units: []source.SourceUnit{unitWithID("u1")}}},
want: "id must not be empty",
},
{
name: "duplicate chunk id",
chunks: []contracts.SourceChunk{
{ID: "chunk-0", SourceID: "source-1", Index: 0, Units: []source.SourceUnit{unitWithID("u1")}},
{ID: "chunk-0", SourceID: "source-1", Index: 1, Units: []source.SourceUnit{unitWithID("u2")}},
},
want: "duplicated",
},
{
name: "wrong source id",
chunks: []contracts.SourceChunk{{ID: "chunk-0", SourceID: "other-source", Index: 0, Units: []source.SourceUnit{unitWithID("u1")}}},
want: "source_id",
},
{
name: "wrong index",
chunks: []contracts.SourceChunk{{ID: "chunk-0", SourceID: "source-1", Index: 1, Units: []source.SourceUnit{unitWithID("u1")}}},
want: "index",
},
{
name: "empty units",
chunks: []contracts.SourceChunk{{ID: "chunk-0", SourceID: "source-1", Index: 0}},
want: "units must not be empty",
},
{
name: "repeated unit inside chunk",
chunks: []contracts.SourceChunk{{ID: "chunk-0", SourceID: "source-1", Index: 0, Units: []source.SourceUnit{unitWithID("u1"), unitWithID("u1")}}},
want: "repeats source unit",
},
{
name: "unknown unit",
chunks: []contracts.SourceChunk{{ID: "chunk-0", SourceID: "source-1", Index: 0, Units: []source.SourceUnit{unitWithID("u9")}}},
want: "was not found",
},
{
name: "units out of source order",
chunks: []contracts.SourceChunk{{ID: "chunk-0", SourceID: "source-1", Index: 0, Units: []source.SourceUnit{unitWithID("u2"), unitWithID("u1")}}},
want: "source document order",
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
modules := defaultRunnerModules()
modules.chunker.chunks = test.chunks
output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()})
assertRunError(t, err, test.want)
if output.Manifest.ValidationStatus != "failed" {
t.Fatalf("ValidationStatus = %q, want failed", output.Manifest.ValidationStatus)
}
if len(modules.extractors["extract-alpha"].requests) != 0 {
t.Fatalf("extractor calls = %d, want none after invalid chunks", len(modules.extractors["extract-alpha"].requests))
}
})
}
}
func TestRunAllowsPartialCoverageAndOverlappingChunks(t *testing.T) {
modules := defaultRunnerModules()
modules.chunker.chunks = []contracts.SourceChunk{
{ID: "chunk-0", SourceID: "source-1", Index: 0, Units: []source.SourceUnit{unitWithID("u1"), unitWithID("u2")}},
{ID: "chunk-1", SourceID: "source-1", Index: 1, Units: []source.SourceUnit{unitWithID("u2")}},
}
output, err := New(newRunnerRegistries(t, modules)).Run(context.Background(), RunInput{Pipeline: resolvedPipeline()})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
if len(output.Approved) != 2 {
t.Fatalf("len(Approved) = %d, want one candidate per accepted chunk", len(output.Approved))
}
}
func TestRunExecutesChunksAndPassesChunkAndLLMClient(t *testing.T) {
modules := defaultRunnerModules()
llmClient := fakeLLMClient{}
@@ -273,6 +358,9 @@ func TestRunExecutesChunksAndPassesChunkAndLLMClient(t *testing.T) {
if !reflect.DeepEqual(extractor.seenChunkIDs, []string{"chunk-0", "chunk-1"}) {
t.Fatalf("seen chunks = %#v, want both chunks", extractor.seenChunkIDs)
}
if len(modules.chunker.requests) != 1 || modules.chunker.requests[0].LLMClient == nil {
t.Fatalf("chunker LLM client = %#v, want client on chunk request", modules.chunker.requests)
}
if len(extractor.seenLLMClients) != 2 || extractor.seenLLMClients[0] == nil || extractor.seenLLMClients[1] == nil {
t.Fatalf("seen LLM clients = %#v, want client for each chunk", extractor.seenLLMClients)
}
@@ -1179,6 +1267,8 @@ func validSourceDocument() *source.SourceDocument {
Digest: "sha256:source",
Units: []source.SourceUnit{
{ID: "u1", Kind: "unit", Text: "Source unit."},
{ID: "u2", Kind: "unit", Text: "Second source unit."},
{ID: "u3", Kind: "unit", Text: "Third source unit."},
},
}
}
@@ -1194,6 +1284,19 @@ func sourceChunkWithID(id string, index int) contracts.SourceChunk {
}
}
func unitWithID(id string) source.SourceUnit {
switch id {
case "u1":
return source.SourceUnit{ID: "u1", Kind: "unit", Text: "Source unit."}
case "u2":
return source.SourceUnit{ID: "u2", Kind: "unit", Text: "Second source unit."}
case "u3":
return source.SourceUnit{ID: "u3", Kind: "unit", Text: "Third source unit."}
default:
return source.SourceUnit{ID: id, Kind: "unit", Text: "Unknown source unit."}
}
}
func warningReasons(warnings []contracts.Warning) []string {
reasons := make([]string, 0, len(warnings))
for _, warning := range warnings {