From e5eb0ba5c8793babcb271be305e969b3022f64de Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Sat, 4 Jul 2026 12:52:16 +0000 Subject: [PATCH] Make chunk validation a framework contract --- docs/internal/modules.md | 5 + docs/internal/pipeline.md | 30 ++++- .../framework/contracts/composition_test.go | 15 ++- internal/framework/contracts/contracts.go | 1 + .../framework/contracts/contracts_test.go | 35 ++++++ .../framework/pipeline/chunk_validation.go | 60 ++++++++++ internal/framework/pipeline/runner.go | 4 + internal/framework/pipeline/runner_test.go | 103 ++++++++++++++++++ 8 files changed, 249 insertions(+), 4 deletions(-) create mode 100644 internal/framework/pipeline/chunk_validation.go diff --git a/docs/internal/modules.md b/docs/internal/modules.md index 1f62371..67f1269 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -20,6 +20,11 @@ A production module package should provide: Module specs should describe capabilities accurately. Resolution uses specs to reject incompatible pipelines before execution. +Chunk modules receive the structured LLM client through `contracts.ChunkRequest` +when they need model-backed chunking. The pipeline runner validates generic +chunk result invariants before extraction; module-owned policies may be stricter +but must stay within the module package. + ## `seriatim` Input Package: `internal/modules/input/seriatim` diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 96b8267..f624e10 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -73,8 +73,34 @@ The runner: 2. builds the input adapter and parses the raw input into a source document; 3. validates the source document; 4. builds the chunker and produces source chunks; -5. runs each selected artifact lane in sorted resolved order; -6. builds the output encoder and validates logical output file names. +5. validates source chunks against framework invariants; +6. runs each selected artifact lane in sorted resolved order; +7. builds the output encoder and validates logical output file names. + +## Chunk Results + +Chunkers implement `contracts.Chunker` and receive a `contracts.ChunkRequest` +with the validated source document, the structured LLM client, the configured +LLM profile, module options, and run metadata. Deterministic and LLM-backed +chunkers use the same contract; provider construction stays outside chunk +modules. + +After `Chunk` returns, the runner appends chunker warnings before returning any +chunker error. When chunking succeeds, the runner validates generic chunk +invariants before running extractors: + +- chunk IDs must be non-empty and unique in the chunk result; +- each chunk `SourceID` must match the source document ID; +- each chunk `Index` must match its zero-based returned order; +- each chunk must contain at least one source unit; +- a chunk must not repeat a source unit; +- every chunk source unit must exist in the source document; +- source units inside each chunk must appear in source-document order. + +The framework does not require complete source-unit coverage and does not reject +overlap between different chunks. Stricter policies, such as full coverage or +non-overlap, belong to individual chunk modules when they are part of that +module's contract. Within an artifact lane, the runner: diff --git a/internal/framework/contracts/composition_test.go b/internal/framework/contracts/composition_test.go index fb59166..29dbb60 100644 --- a/internal/framework/contracts/composition_test.go +++ b/internal/framework/contracts/composition_test.go @@ -17,6 +17,7 @@ var _ contracts.Extractor = compositionExtractor{} var _ contracts.Merger = compositionMerger{} var _ contracts.Normalizer = compositionNormalizer{} var _ contracts.Validator = compositionValidator{} +var _ contracts.StructuredLLMClient = compositionLLMClient{} var _ contracts.OutputEncoder = compositionOutputEncoder{} func TestContractsComposeAcrossPackages(t *testing.T) { @@ -38,8 +39,9 @@ func TestContractsComposeAcrossPackages(t *testing.T) { } chunking, err := chunker.Chunk(ctx, contracts.ChunkRequest{ - Source: doc, - Metadata: map[string]any{"max_units": 2}, + Source: doc, + LLMClient: compositionLLMClient{}, + Metadata: map[string]any{"max_units": 2}, }) if err != nil { t.Fatalf("Chunk() error = %v, want nil", err) @@ -164,6 +166,9 @@ func (chunker compositionChunker) Chunk(ctx context.Context, req contracts.Chunk if req.Source == nil { return contracts.ChunkResult{}, errors.New("source document is required") } + if req.LLMClient == nil { + return contracts.ChunkResult{}, errors.New("structured llm client is required") + } return contracts.ChunkResult{ Chunks: []contracts.SourceChunk{ @@ -178,6 +183,12 @@ func (chunker compositionChunker) Chunk(ctx context.Context, req contracts.Chunk }, nil } +type compositionLLMClient struct{} + +func (client compositionLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) { + return contracts.StructuredCompletionResponse{}, nil +} + type compositionExtractor struct{} func (extractor compositionExtractor) Key() string { diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index a4b8ecc..a81f40d 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -58,6 +58,7 @@ type SourceChunk struct { type ChunkRequest struct { Source *source.SourceDocument `json:"-"` + LLMClient StructuredLLMClient `json:"-"` LLMProfile string `json:"llm_profile,omitempty"` Options map[string]any `json:"options,omitempty"` Metadata map[string]any `json:"metadata,omitempty"` diff --git a/internal/framework/contracts/contracts_test.go b/internal/framework/contracts/contracts_test.go index 867b3bd..316df6e 100644 --- a/internal/framework/contracts/contracts_test.go +++ b/internal/framework/contracts/contracts_test.go @@ -117,6 +117,27 @@ func TestFakeChunkerReturnsSourceChunks(t *testing.T) { } } +func TestFakeChunkerReceivesLLMClient(t *testing.T) { + doc := &source.SourceDocument{ + ID: "source-1", + Kind: "document", + Format: "text/plain", + Digest: "sha256:abc123", + Units: []source.SourceUnit{ + {ID: "u1", Kind: "section", Text: "Source text."}, + }, + } + client := fakeLLMClient{} + chunker := &recordingChunker{key: "llm-chunker"} + + if _, err := chunker.Chunk(context.Background(), ChunkRequest{Source: doc, LLMClient: client}); err != nil { + t.Fatalf("Chunk() error = %v, want nil", err) + } + if chunker.request.LLMClient == nil { + t.Fatal("ChunkRequest.LLMClient = nil, want structured LLM client") + } +} + func TestFakeExtractorReceivesChunkAndAmbientContext(t *testing.T) { extractor := fakeExtractor{ key: "generic-extractor", @@ -305,6 +326,20 @@ func (chunker fakeChunker) Chunk(ctx context.Context, req ChunkRequest) (ChunkRe }, nil } +type recordingChunker struct { + key string + request ChunkRequest +} + +func (chunker *recordingChunker) Key() string { + return chunker.key +} + +func (chunker *recordingChunker) Chunk(ctx context.Context, req ChunkRequest) (ChunkResult, error) { + chunker.request = req + return fakeChunker{key: chunker.key}.Chunk(ctx, req) +} + type fakeExtractor struct { key string artifactType string diff --git a/internal/framework/pipeline/chunk_validation.go b/internal/framework/pipeline/chunk_validation.go new file mode 100644 index 0000000..effc1cb --- /dev/null +++ b/internal/framework/pipeline/chunk_validation.go @@ -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 +} diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index c65f72a..45379b6 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -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 { diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 7fc4e48..edaa7d5 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -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 {