Compare commits

..

7 Commits

30 changed files with 4991 additions and 390 deletions

View File

@@ -33,8 +33,18 @@ type RejectedArtifact struct {
Message string `json:"message"`
}
type ArtifactLaneManifest struct {
ID string `json:"id"`
Extractor string `json:"extractor"`
Merger string `json:"merger"`
Normalizer string `json:"normalizer"`
Validators []string `json:"validators,omitempty"`
}
type RunManifest struct {
RunID string `json:"run_id,omitempty"`
PipelineID string `json:"pipeline_id,omitempty"`
PipelineDigest string `json:"pipeline_digest,omitempty"`
InputModule string `json:"input_module,omitempty"`
Chunker string `json:"chunker,omitempty"`
SourceDigests []string `json:"source_digests,omitempty"`
@@ -42,6 +52,7 @@ type RunManifest struct {
Merger string `json:"merger,omitempty"`
Normalizer string `json:"normalizer,omitempty"`
OutputEncoder string `json:"output_encoder,omitempty"`
ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"`
SchemaVersion string `json:"schema_version,omitempty"`
ValidationStatus string `json:"validation_status,omitempty"`
StartedAt *time.Time `json:"started_at,omitempty"`

View File

@@ -123,6 +123,47 @@ func TestRunManifestOmitsEmptyOptionalFields(t *testing.T) {
}
}
func TestRunManifestIncludesPipelineAndArtifactLaneFields(t *testing.T) {
manifest := RunManifest{
PipelineID: "pipeline-1",
PipelineDigest: "sha256:abc123",
ArtifactLanes: []ArtifactLaneManifest{
{
ID: "events",
Extractor: "event-extractor",
Merger: "appendorder",
Normalizer: "noop",
Validators: []string{"grounded"},
},
},
}
gotJSON, err := json.Marshal(manifest)
if err != nil {
t.Fatalf("json.Marshal() error = %v", err)
}
var got map[string]any
if err := json.Unmarshal(gotJSON, &got); err != nil {
t.Fatalf("json.Unmarshal() error = %v", err)
}
assertHasKeys(t, got, "pipeline_id", "pipeline_digest", "artifact_lanes")
lanes, ok := got["artifact_lanes"].([]any)
if !ok {
t.Fatalf("artifact_lanes = %#v, want array", got["artifact_lanes"])
}
if len(lanes) != 1 {
t.Fatalf("len(artifact_lanes) = %d, want 1", len(lanes))
}
lane, ok := lanes[0].(map[string]any)
if !ok {
t.Fatalf("artifact_lanes[0] = %#v, want object", lanes[0])
}
assertHasKeys(t, lane, "id", "extractor", "merger", "normalizer", "validators")
}
func assertHasKeys(t *testing.T, values map[string]any, keys ...string) {
t.Helper()

View File

@@ -12,14 +12,22 @@ import (
)
var _ contracts.InputAdapter = compositionAdapter{}
var _ contracts.Chunker = compositionChunker{}
var _ contracts.Extractor = compositionExtractor{}
var _ contracts.Merger = compositionMerger{}
var _ contracts.Normalizer = compositionNormalizer{}
var _ contracts.Validator = compositionValidator{}
var _ contracts.OutputEncoder = compositionOutputEncoder{}
func TestContractsComposeAcrossPackages(t *testing.T) {
ctx := context.Background()
adapter := compositionAdapter{}
chunker := compositionChunker{}
extractor := compositionExtractor{}
merger := compositionMerger{}
normalizer := compositionNormalizer{}
validator := compositionValidator{}
encoder := compositionOutputEncoder{}
doc, err := adapter.Parse(ctx, contracts.ParseRequest{SourceID: "source-1"})
if err != nil {
@@ -29,7 +37,22 @@ func TestContractsComposeAcrossPackages(t *testing.T) {
t.Fatalf("ValidateDocument() error = %v, want nil", err)
}
extraction, err := extractor.Extract(ctx, contracts.ExtractionRequest{Source: doc})
chunking, err := chunker.Chunk(ctx, contracts.ChunkRequest{
Source: doc,
Metadata: map[string]any{"max_units": 2},
})
if err != nil {
t.Fatalf("Chunk() error = %v, want nil", err)
}
if len(chunking.Chunks) != 1 {
t.Fatalf("len(Chunks) = %d, want 1", len(chunking.Chunks))
}
extraction, err := extractor.Extract(ctx, contracts.ExtractionRequest{
Source: doc,
Chunk: &chunking.Chunks[0],
AmbientContext: map[string]any{"synopsis": "example synopsis"},
})
if err != nil {
t.Fatalf("Extract() error = %v, want nil", err)
}
@@ -44,9 +67,38 @@ func TestContractsComposeAcrossPackages(t *testing.T) {
}
}
merge, err := merger.Merge(ctx, contracts.MergeRequest{
Source: doc,
LaneID: candidate.ArtifactType,
ChunkArtifacts: []contracts.ChunkArtifacts{
{
Chunk: chunking.Chunks[0],
Candidates: extraction.Candidates,
},
},
})
if err != nil {
t.Fatalf("Merge() error = %v, want nil", err)
}
if len(merge.Candidates) != 1 {
t.Fatalf("len(merge.Candidates) = %d, want 1", len(merge.Candidates))
}
normalize, err := normalizer.Normalize(ctx, contracts.NormalizeRequest{
Source: doc,
LaneID: candidate.ArtifactType,
Candidates: merge.Candidates,
})
if err != nil {
t.Fatalf("Normalize() error = %v, want nil", err)
}
if len(normalize.Candidates) != 1 {
t.Fatalf("len(normalize.Candidates) = %d, want 1", len(normalize.Candidates))
}
validation, err := validator.Validate(ctx, contracts.ValidationRequest{
Source: doc,
Candidates: extraction.Candidates,
Candidates: normalize.Candidates,
})
if err != nil {
t.Fatalf("Validate() error = %v, want nil", err)
@@ -62,6 +114,22 @@ func TestContractsComposeAcrossPackages(t *testing.T) {
if decision.CandidateIndex != candidate.Index {
t.Fatalf("CandidateIndex = %d, want %d", decision.CandidateIndex, candidate.Index)
}
output, err := encoder.Encode(ctx, contracts.OutputRequest{
Manifest: artifacts.RunManifest{RunID: "run-1"},
Approved: []artifacts.Artifact{
artifacts.ArtifactFromCandidate(normalize.Candidates[0]),
},
})
if err != nil {
t.Fatalf("Encode() error = %v, want nil", err)
}
if output.ContentType != "application/json" {
t.Fatalf("ContentType = %q, want application/json", output.ContentType)
}
if len(output.Bytes) == 0 {
t.Fatal("len(Bytes) = 0, want encoded bytes")
}
}
type compositionAdapter struct{}
@@ -83,6 +151,30 @@ func (adapter compositionAdapter) Parse(ctx context.Context, req contracts.Parse
}, nil
}
type compositionChunker struct{}
func (chunker compositionChunker) Key() string {
return "generic-chunker"
}
func (chunker compositionChunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) {
if req.Source == nil {
return contracts.ChunkResult{}, errors.New("source document is required")
}
return contracts.ChunkResult{
Chunks: []contracts.SourceChunk{
{
ID: req.Source.ID + ":chunk:0",
SourceID: req.Source.ID,
Index: 0,
Units: append([]source.SourceUnit(nil), req.Source.Units...),
Metadata: map[string]any{"strategy": "whole-document"},
},
},
}, nil
}
type compositionExtractor struct{}
func (extractor compositionExtractor) Key() string {
@@ -105,6 +197,13 @@ func (extractor compositionExtractor) Extract(ctx context.Context, req contracts
if req.Source == nil {
return contracts.ExtractionResult{}, errors.New("source document is required")
}
units := req.Source.Units
if req.Chunk != nil {
units = req.Chunk.Units
}
if req.AmbientContext["synopsis"] == "" {
return contracts.ExtractionResult{}, errors.New("ambient synopsis is required")
}
return contracts.ExtractionResult{
Candidates: []artifacts.ArtifactCandidate{
@@ -117,8 +216,8 @@ func (extractor compositionExtractor) Extract(ctx context.Context, req contracts
SourceRefs: []source.SourceRef{
{
SourceID: req.Source.ID,
StartUnitID: req.Source.Units[0].ID,
EndUnitID: req.Source.Units[1].ID,
StartUnitID: units[0].ID,
EndUnitID: units[len(units)-1].ID,
},
},
},
@@ -126,6 +225,31 @@ func (extractor compositionExtractor) Extract(ctx context.Context, req contracts
}, nil
}
type compositionMerger struct{}
func (merger compositionMerger) Key() string {
return "generic-merger"
}
func (merger compositionMerger) 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 compositionNormalizer struct{}
func (normalizer compositionNormalizer) Key() string {
return "generic-normalizer"
}
func (normalizer compositionNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
return contracts.NormalizeResult{Candidates: req.Candidates}, nil
}
type compositionValidator struct{}
func (validator compositionValidator) Name() string {
@@ -148,3 +272,28 @@ func (validator compositionValidator) Validate(ctx context.Context, req contract
Decisions: decisions,
}, nil
}
type compositionOutputEncoder struct{}
func (encoder compositionOutputEncoder) Key() string {
return "generic-output"
}
func (encoder compositionOutputEncoder) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
payload := struct {
RunID string `json:"run_id"`
ApprovedCount int `json:"approved_count"`
}{
RunID: req.Manifest.RunID,
ApprovedCount: len(req.Approved),
}
encoded, err := json.Marshal(payload)
if err != nil {
return contracts.OutputResult{}, err
}
return contracts.OutputResult{
Bytes: encoded,
ContentType: "application/json",
}, nil
}

View File

@@ -46,8 +46,33 @@ type InputAdapter interface {
Parse(ctx context.Context, req ParseRequest) (*source.SourceDocument, error)
}
type SourceChunk struct {
ID string `json:"id"`
SourceID string `json:"source_id"`
Index int `json:"index"`
Units []source.SourceUnit `json:"units"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type ChunkRequest struct {
Source *source.SourceDocument `json:"-"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type ChunkResult struct {
Chunks []SourceChunk `json:"chunks"`
Warnings []Warning `json:"warnings,omitempty"`
}
type Chunker interface {
Key() string
Chunk(ctx context.Context, req ChunkRequest) (ChunkResult, error)
}
type ExtractionRequest struct {
Source *source.SourceDocument `json:"-"`
Chunk *SourceChunk `json:"chunk,omitempty"`
AmbientContext map[string]any `json:"ambient_context,omitempty"`
LLMClient StructuredLLMClient `json:"-"`
Metadata map[string]any `json:"metadata,omitempty"`
}
@@ -65,6 +90,45 @@ type Extractor interface {
Extract(ctx context.Context, req ExtractionRequest) (ExtractionResult, error)
}
type ChunkArtifacts struct {
Chunk SourceChunk `json:"chunk"`
Candidates []artifacts.ArtifactCandidate `json:"candidates"`
}
type MergeRequest struct {
Source *source.SourceDocument `json:"-"`
LaneID string `json:"lane_id"`
ChunkArtifacts []ChunkArtifacts `json:"chunk_artifacts"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type MergeResult struct {
Candidates []artifacts.ArtifactCandidate `json:"candidates"`
Warnings []Warning `json:"warnings,omitempty"`
}
type Merger interface {
Key() string
Merge(ctx context.Context, req MergeRequest) (MergeResult, error)
}
type NormalizeRequest struct {
Source *source.SourceDocument `json:"-"`
LaneID string `json:"lane_id"`
Candidates []artifacts.ArtifactCandidate `json:"candidates"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type NormalizeResult struct {
Candidates []artifacts.ArtifactCandidate `json:"candidates"`
Warnings []Warning `json:"warnings,omitempty"`
}
type Normalizer interface {
Key() string
Normalize(ctx context.Context, req NormalizeRequest) (NormalizeResult, error)
}
type ValidationRequest struct {
Source *source.SourceDocument `json:"-"`
Candidates []artifacts.ArtifactCandidate `json:"candidates"`
@@ -95,3 +159,22 @@ type Warning struct {
ReasonCode string `json:"reason_code"`
Message string `json:"message"`
}
type OutputRequest struct {
Manifest artifacts.RunManifest `json:"manifest"`
Approved []artifacts.Artifact `json:"approved,omitempty"`
Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"`
Warnings []Warning `json:"warnings,omitempty"`
Metadata map[string]any `json:"metadata,omitempty"`
}
type OutputResult struct {
Bytes []byte `json:"-"`
ContentType string `json:"content_type,omitempty"`
Warnings []Warning `json:"warnings,omitempty"`
}
type OutputEncoder interface {
Key() string
Encode(ctx context.Context, req OutputRequest) (OutputResult, error)
}

View File

@@ -10,9 +10,13 @@ import (
)
var _ InputAdapter = fakeAdapter{}
var _ Chunker = fakeChunker{}
var _ Extractor = fakeExtractor{}
var _ Merger = fakeMerger{}
var _ Normalizer = fakeNormalizer{}
var _ Validator = fakeValidator{}
var _ StructuredLLMClient = fakeLLMClient{}
var _ OutputEncoder = fakeOutputEncoder{}
func TestFakeExtractorReturnsCandidateAndValidator(t *testing.T) {
validator := fakeValidator{name: "generic-validator"}
@@ -74,6 +78,166 @@ func TestFakeExtractorReturnsCandidateAndValidator(t *testing.T) {
}
}
func TestFakeChunkerReturnsSourceChunks(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."},
},
}
chunker := fakeChunker{key: "generic-chunker"}
result, err := chunker.Chunk(context.Background(), ChunkRequest{Source: doc})
if err != nil {
t.Fatalf("Chunk() error = %v, want nil", err)
}
if chunker.Key() != "generic-chunker" {
t.Fatalf("Key() = %q, want generic-chunker", chunker.Key())
}
if len(result.Chunks) != 1 {
t.Fatalf("len(Chunks) = %d, want 1", len(result.Chunks))
}
chunk := result.Chunks[0]
if chunk.ID != "source-1:chunk:0" {
t.Fatalf("SourceChunk.ID = %q, want source-1:chunk:0", chunk.ID)
}
if chunk.SourceID != doc.ID {
t.Fatalf("SourceChunk.SourceID = %q, want %q", chunk.SourceID, doc.ID)
}
if chunk.Index != 0 {
t.Fatalf("SourceChunk.Index = %d, want 0", chunk.Index)
}
if len(chunk.Units) != 1 {
t.Fatalf("len(SourceChunk.Units) = %d, want 1", len(chunk.Units))
}
}
func TestFakeExtractorReceivesChunkAndAmbientContext(t *testing.T) {
extractor := fakeExtractor{
key: "generic-extractor",
artifactType: "generic-artifact",
schemaVersion: "v1",
}
doc := &source.SourceDocument{
ID: "source-1",
Kind: "document",
Format: "text/plain",
Digest: "sha256:abc123",
Units: []source.SourceUnit{
{ID: "u1", Kind: "section", Text: "First source text."},
{ID: "u2", Kind: "section", Text: "Second source text."},
},
}
chunk := SourceChunk{
ID: "source-1:chunk:1",
SourceID: doc.ID,
Index: 1,
Units: []source.SourceUnit{doc.Units[1]},
}
result, err := extractor.Extract(context.Background(), ExtractionRequest{
Source: doc,
Chunk: &chunk,
AmbientContext: map[string]any{"mode": "chunked"},
})
if err != nil {
t.Fatalf("Extract() error = %v, want nil", err)
}
if len(result.Candidates) != 1 {
t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates))
}
candidate := result.Candidates[0]
if string(candidate.Payload) != `{"value":"chunked"}` {
t.Fatalf("ArtifactCandidate.Payload = %s, want chunked payload", candidate.Payload)
}
if len(candidate.SourceRefs) != 1 {
t.Fatalf("len(SourceRefs) = %d, want 1", len(candidate.SourceRefs))
}
ref := candidate.SourceRefs[0]
if ref.StartUnitID != "u2" || ref.EndUnitID != "u2" {
t.Fatalf("SourceRef = %+v, want u2 range", ref)
}
}
func TestFakeMergeNormalizeAndOutputContracts(t *testing.T) {
candidate := artifacts.ArtifactCandidate{
Index: 0,
ExtractorKey: "generic-extractor",
ArtifactType: "generic-artifact",
SchemaVersion: "v1",
Payload: json.RawMessage(`{"value":"example"}`),
}
chunk := SourceChunk{
ID: "source-1:chunk:0",
SourceID: "source-1",
Index: 0,
Units: []source.SourceUnit{
{ID: "u1", Kind: "section", Text: "Source text."},
},
}
merger := fakeMerger{key: "generic-merger"}
normalizer := fakeNormalizer{key: "generic-normalizer"}
encoder := fakeOutputEncoder{key: "generic-output"}
merged, err := merger.Merge(context.Background(), MergeRequest{
LaneID: "generic-artifact",
ChunkArtifacts: []ChunkArtifacts{
{
Chunk: chunk,
Candidates: []artifacts.ArtifactCandidate{candidate},
},
},
})
if err != nil {
t.Fatalf("Merge() error = %v, want nil", err)
}
if merger.Key() != "generic-merger" {
t.Fatalf("Merger.Key() = %q, want generic-merger", merger.Key())
}
if len(merged.Candidates) != 1 {
t.Fatalf("len(merged.Candidates) = %d, want 1", len(merged.Candidates))
}
normalized, err := normalizer.Normalize(context.Background(), NormalizeRequest{
LaneID: "generic-artifact",
Candidates: merged.Candidates,
})
if err != nil {
t.Fatalf("Normalize() error = %v, want nil", err)
}
if normalizer.Key() != "generic-normalizer" {
t.Fatalf("Normalizer.Key() = %q, want generic-normalizer", normalizer.Key())
}
if len(normalized.Candidates) != 1 {
t.Fatalf("len(normalized.Candidates) = %d, want 1", len(normalized.Candidates))
}
encoded, err := encoder.Encode(context.Background(), OutputRequest{
Manifest: artifacts.RunManifest{RunID: "run-1"},
Approved: []artifacts.Artifact{
artifacts.ArtifactFromCandidate(normalized.Candidates[0]),
},
})
if err != nil {
t.Fatalf("Encode() error = %v, want nil", err)
}
if encoder.Key() != "generic-output" {
t.Fatalf("OutputEncoder.Key() = %q, want generic-output", encoder.Key())
}
if encoded.ContentType != "application/json" {
t.Fatalf("ContentType = %q, want application/json", encoded.ContentType)
}
if string(encoded.Bytes) != `{"run_id":"run-1","approved_count":1}` {
t.Fatalf("Bytes = %s, want encoded output", encoded.Bytes)
}
}
type fakeAdapter struct {
key string
doc *source.SourceDocument
@@ -87,6 +251,27 @@ func (adapter fakeAdapter) Parse(ctx context.Context, req ParseRequest) (*source
return adapter.doc, nil
}
type fakeChunker struct {
key string
}
func (chunker fakeChunker) Key() string {
return chunker.key
}
func (chunker fakeChunker) Chunk(ctx context.Context, req ChunkRequest) (ChunkResult, error) {
return ChunkResult{
Chunks: []SourceChunk{
{
ID: req.Source.ID + ":chunk:0",
SourceID: req.Source.ID,
Index: 0,
Units: append([]source.SourceUnit(nil), req.Source.Units...),
},
},
}, nil
}
type fakeExtractor struct {
key string
artifactType string
@@ -111,6 +296,15 @@ func (extractor fakeExtractor) Validators() []Validator {
}
func (extractor fakeExtractor) Extract(ctx context.Context, req ExtractionRequest) (ExtractionResult, error) {
units := req.Source.Units
if req.Chunk != nil {
units = req.Chunk.Units
}
payload := json.RawMessage(`{"value":"example"}`)
if req.AmbientContext["mode"] == "chunked" {
payload = json.RawMessage(`{"value":"chunked"}`)
}
return ExtractionResult{
Candidates: []artifacts.ArtifactCandidate{
{
@@ -118,12 +312,12 @@ func (extractor fakeExtractor) Extract(ctx context.Context, req ExtractionReques
ExtractorKey: extractor.key,
ArtifactType: extractor.artifactType,
SchemaVersion: extractor.schemaVersion,
Payload: json.RawMessage(`{"value":"example"}`),
Payload: payload,
SourceRefs: []source.SourceRef{
{
SourceID: req.Source.ID,
StartUnitID: req.Source.Units[0].ID,
EndUnitID: req.Source.Units[0].ID,
StartUnitID: units[0].ID,
EndUnitID: units[len(units)-1].ID,
},
},
},
@@ -131,6 +325,35 @@ func (extractor fakeExtractor) Extract(ctx context.Context, req ExtractionReques
}, nil
}
type fakeMerger struct {
key string
}
func (merger fakeMerger) Key() string {
return merger.key
}
func (merger fakeMerger) Merge(ctx context.Context, req MergeRequest) (MergeResult, error) {
var candidates []artifacts.ArtifactCandidate
for _, chunkArtifacts := range req.ChunkArtifacts {
candidates = append(candidates, chunkArtifacts.Candidates...)
}
return MergeResult{Candidates: candidates}, nil
}
type fakeNormalizer struct {
key string
}
func (normalizer fakeNormalizer) Key() string {
return normalizer.key
}
func (normalizer fakeNormalizer) Normalize(ctx context.Context, req NormalizeRequest) (NormalizeResult, error) {
return NormalizeResult{Candidates: req.Candidates}, nil
}
type fakeValidator struct {
name string
}
@@ -163,3 +386,18 @@ func (client fakeLLMClient) CompleteStructured(ctx context.Context, req Structur
Content: json.RawMessage(`{"value":"example"}`),
}, nil
}
type fakeOutputEncoder struct {
key string
}
func (encoder fakeOutputEncoder) Key() string {
return encoder.key
}
func (encoder fakeOutputEncoder) Encode(ctx context.Context, req OutputRequest) (OutputResult, error) {
return OutputResult{
Bytes: []byte(`{"run_id":"` + req.Manifest.RunID + `","approved_count":1}`),
ContentType: "application/json",
}, nil
}

View File

@@ -0,0 +1,102 @@
package pipeline
import (
"fmt"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type ChunkerConstructor func() (contracts.Chunker, error)
type ChunkerRegistry struct {
constructors map[string]ChunkerConstructor
specs map[string]ModuleSpec
}
func NewChunkerRegistry() *ChunkerRegistry {
return &ChunkerRegistry{
constructors: make(map[string]ChunkerConstructor),
specs: make(map[string]ModuleSpec),
}
}
func (r *ChunkerRegistry) Register(key string, constructor ChunkerConstructor) error {
return r.RegisterWithSpec(defaultModuleSpec(key, StageChunk), constructor)
}
func (r *ChunkerRegistry) RegisterWithSpec(spec ModuleSpec, constructor ChunkerConstructor) error {
if r == nil {
return fmt.Errorf("chunker registry must not be nil")
}
normalizedSpec := normalizeModuleSpec(spec)
if err := validateModuleSpec("chunker", StageChunk, normalizedSpec); err != nil {
return err
}
if constructor == nil {
return fmt.Errorf("chunker constructor for %q must not be nil", normalizedSpec.Key)
}
if _, ok := r.constructors[normalizedSpec.Key]; ok {
return fmt.Errorf("chunker %q is already registered", normalizedSpec.Key)
}
if r.constructors == nil {
r.constructors = make(map[string]ChunkerConstructor)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.constructors[normalizedSpec.Key] = constructor
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
func (r *ChunkerRegistry) Build(key string) (contracts.Chunker, error) {
if r == nil {
return nil, fmt.Errorf("chunker registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
if normalizedKey == "" {
return nil, fmt.Errorf("chunker key must not be empty")
}
constructor, ok := r.constructors[normalizedKey]
if !ok {
return nil, fmt.Errorf("chunker %q is not registered", normalizedKey)
}
chunker, err := constructor()
if err != nil {
return nil, fmt.Errorf("build chunker %q: %w", normalizedKey, err)
}
if chunker == nil {
return nil, fmt.Errorf("chunker %q constructor returned nil", normalizedKey)
}
if chunker.Key() != normalizedKey {
return nil, fmt.Errorf("chunker %q returned key %q", normalizedKey, chunker.Key())
}
return chunker, nil
}
func (r *ChunkerRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false
}
spec, ok := r.specs[strings.TrimSpace(key)]
if !ok {
return ModuleSpec{}, false
}
return cloneModuleSpec(spec), true
}
func (r *ChunkerRegistry) RegisteredKeys() []string {
if r == nil {
return nil
}
return sortedRegistryKeys(r.constructors)
}

View File

@@ -0,0 +1,383 @@
package pipeline
import (
"context"
"errors"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type registryBehaviorCase[M any] struct {
name string
key string
stage ModuleStage
wrongStage ModuleStage
newRegistry func() any
register func(any, string, func() (M, error)) error
registerWithSpec func(any, ModuleSpec, func() (M, error)) error
build func(any, string) (M, error)
spec func(any, string) (ModuleSpec, bool)
registeredKeys func(any) []string
nilRegister func(string, func() (M, error)) error
nilBuild func(string) (M, error)
nilSpec func(string) (ModuleSpec, bool)
nilRegisteredKey func() []string
constructor func(string) func() (M, error)
moduleKey func(M) string
}
func TestChunkerRegistryBehavior(t *testing.T) {
runRegistryBehaviorTests(t, registryBehaviorCase[contracts.Chunker]{
name: "ChunkerRegistry",
key: "generic-chunker",
stage: StageChunk,
wrongStage: StageExtract,
newRegistry: func() any {
return NewChunkerRegistry()
},
register: func(registry any, key string, constructor func() (contracts.Chunker, error)) error {
return registry.(*ChunkerRegistry).Register(key, constructor)
},
registerWithSpec: func(registry any, spec ModuleSpec, constructor func() (contracts.Chunker, error)) error {
return registry.(*ChunkerRegistry).RegisterWithSpec(spec, constructor)
},
build: func(registry any, key string) (contracts.Chunker, error) {
return registry.(*ChunkerRegistry).Build(key)
},
spec: func(registry any, key string) (ModuleSpec, bool) {
return registry.(*ChunkerRegistry).Spec(key)
},
registeredKeys: func(registry any) []string {
return registry.(*ChunkerRegistry).RegisteredKeys()
},
nilRegister: func(key string, constructor func() (contracts.Chunker, error)) error {
var registry *ChunkerRegistry
return registry.Register(key, constructor)
},
nilBuild: func(key string) (contracts.Chunker, error) {
var registry *ChunkerRegistry
return registry.Build(key)
},
nilSpec: func(key string) (ModuleSpec, bool) {
var registry *ChunkerRegistry
return registry.Spec(key)
},
nilRegisteredKey: func() []string {
var registry *ChunkerRegistry
return registry.RegisteredKeys()
},
constructor: func(key string) func() (contracts.Chunker, error) {
return func() (contracts.Chunker, error) {
return registryChunker{key: key}, nil
}
},
moduleKey: func(module contracts.Chunker) string {
return module.Key()
},
})
}
func runRegistryBehaviorTests[M any](t *testing.T, testCase registryBehaviorCase[M]) {
t.Helper()
t.Run(testCase.name+"/register and build", func(t *testing.T) {
registry := testCase.newRegistry()
if err := testCase.register(registry, testCase.key, testCase.constructor(testCase.key)); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
module, err := testCase.build(registry, testCase.key)
if err != nil {
t.Fatalf("Build() error = %v, want nil", err)
}
if got := testCase.moduleKey(module); got != testCase.key {
t.Fatalf("module key = %q, want %q", got, testCase.key)
}
})
t.Run(testCase.name+"/metadata registration and lookup", func(t *testing.T) {
registry := testCase.newRegistry()
spec := ModuleSpec{
Key: " " + testCase.key + " ",
Stage: testCase.stage,
Provides: []string{" beta ", "alpha", "", "beta"},
Requires: []string{" source ", "source", ""},
}
if err := testCase.registerWithSpec(registry, spec, testCase.constructor(testCase.key)); err != nil {
t.Fatalf("RegisterWithSpec() error = %v, want nil", err)
}
got, ok := testCase.spec(registry, " "+testCase.key+"\n")
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{
Key: testCase.key,
Stage: testCase.stage,
Provides: []string{"alpha", "beta"},
Requires: []string{"source"},
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
}
got.Provides[0] = "changed"
again, ok := testCase.spec(registry, testCase.key)
if !ok {
t.Fatal("Spec() after caller mutation ok = false, want true")
}
if !reflect.DeepEqual(again, want) {
t.Fatalf("Spec() after caller mutation = %#v, want %#v", again, want)
}
})
t.Run(testCase.name+"/default spec from register", func(t *testing.T) {
registry := testCase.newRegistry()
if err := testCase.register(registry, " "+testCase.key+" ", testCase.constructor(testCase.key)); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
spec, ok := testCase.spec(registry, testCase.key)
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{Key: testCase.key, Stage: testCase.stage}
if !reflect.DeepEqual(spec, want) {
t.Fatalf("Spec() = %#v, want %#v", spec, want)
}
})
t.Run(testCase.name+"/wrong stage rejection", func(t *testing.T) {
registry := testCase.newRegistry()
err := testCase.registerWithSpec(registry, ModuleSpec{Key: testCase.key, Stage: testCase.wrongStage}, testCase.constructor(testCase.key))
if err == nil {
t.Fatal("RegisterWithSpec() error = nil, want error")
}
if !strings.Contains(err.Error(), "stage") {
t.Fatalf("RegisterWithSpec() error = %q, want stage error", err.Error())
}
})
t.Run(testCase.name+"/key trimming", func(t *testing.T) {
registry := testCase.newRegistry()
if err := testCase.register(registry, " "+testCase.key+" ", testCase.constructor(testCase.key)); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
module, err := testCase.build(registry, "\t"+testCase.key+"\n")
if err != nil {
t.Fatalf("Build() error = %v, want nil", err)
}
if got := testCase.moduleKey(module); got != testCase.key {
t.Fatalf("module key = %q, want %q", got, testCase.key)
}
})
t.Run(testCase.name+"/empty key rejection", func(t *testing.T) {
registry := testCase.newRegistry()
err := testCase.register(registry, " \t", testCase.constructor(testCase.key))
if err == nil {
t.Fatal("Register() error = nil, want error")
}
if !strings.Contains(err.Error(), "key must not be empty") {
t.Fatalf("Register() error = %q, want empty key error", err.Error())
}
})
t.Run(testCase.name+"/duplicate key rejection", func(t *testing.T) {
registry := testCase.newRegistry()
if err := testCase.register(registry, testCase.key, testCase.constructor(testCase.key)); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
err := testCase.register(registry, " "+testCase.key+" ", testCase.constructor(testCase.key))
if err == nil {
t.Fatal("Register() error = nil, want error")
}
if !strings.Contains(err.Error(), "already registered") {
t.Fatalf("Register() error = %q, want duplicate key error", err.Error())
}
})
t.Run(testCase.name+"/nil constructor rejection", func(t *testing.T) {
registry := testCase.newRegistry()
err := testCase.register(registry, testCase.key, nil)
if err == nil {
t.Fatal("Register() error = nil, want error")
}
if !strings.Contains(err.Error(), "constructor") {
t.Fatalf("Register() error = %q, want constructor error", err.Error())
}
})
t.Run(testCase.name+"/unknown key build error", func(t *testing.T) {
registry := testCase.newRegistry()
_, err := testCase.build(registry, "missing")
if err == nil {
t.Fatal("Build() error = nil, want error")
}
if !strings.Contains(err.Error(), "not registered") {
t.Fatalf("Build() error = %q, want unknown key error", err.Error())
}
})
t.Run(testCase.name+"/constructor error wrapping", func(t *testing.T) {
registry := testCase.newRegistry()
constructorErr := errors.New("constructor failed")
if err := testCase.register(registry, testCase.key, func() (M, error) {
var zero M
return zero, constructorErr
}); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
_, err := testCase.build(registry, testCase.key)
if err == nil {
t.Fatal("Build() error = nil, want error")
}
if !errors.Is(err, constructorErr) {
t.Fatalf("Build() error = %v, want wrapped constructor error", err)
}
if !strings.Contains(err.Error(), testCase.key) {
t.Fatalf("Build() error = %q, want key context", err.Error())
}
})
t.Run(testCase.name+"/nil module rejection", func(t *testing.T) {
registry := testCase.newRegistry()
if err := testCase.register(registry, testCase.key, func() (M, error) {
var zero M
return zero, nil
}); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
_, err := testCase.build(registry, testCase.key)
if err == nil {
t.Fatal("Build() error = nil, want error")
}
if !strings.Contains(err.Error(), "returned nil") {
t.Fatalf("Build() error = %q, want nil module error", err.Error())
}
})
t.Run(testCase.name+"/key mismatch rejection", func(t *testing.T) {
registry := testCase.newRegistry()
if err := testCase.register(registry, testCase.key, testCase.constructor("other")); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
_, err := testCase.build(registry, testCase.key)
if err == nil {
t.Fatal("Build() error = nil, want error")
}
if !strings.Contains(err.Error(), "returned") {
t.Fatalf("Build() error = %q, want mismatch error", err.Error())
}
})
t.Run(testCase.name+"/sorted registered keys", func(t *testing.T) {
registry := testCase.newRegistry()
for _, key := range []string{"zeta", "alpha", "middle"} {
if err := testCase.register(registry, key, testCase.constructor(key)); err != nil {
t.Fatalf("Register(%q) error = %v, want nil", key, err)
}
}
keys := testCase.registeredKeys(registry)
want := []string{"alpha", "middle", "zeta"}
if !reflect.DeepEqual(keys, want) {
t.Fatalf("RegisteredKeys() = %#v, want %#v", keys, want)
}
keys[0] = "changed"
if got := testCase.registeredKeys(registry); !reflect.DeepEqual(got, want) {
t.Fatalf("RegisteredKeys() after caller mutation = %#v, want %#v", got, want)
}
})
t.Run(testCase.name+"/nil registry behavior", func(t *testing.T) {
if err := testCase.nilRegister(testCase.key, testCase.constructor(testCase.key)); err == nil {
t.Fatal("Register() error = nil, want error")
}
if _, err := testCase.nilBuild(testCase.key); err == nil {
t.Fatal("Build() error = nil, want error")
}
if _, ok := testCase.nilSpec(testCase.key); ok {
t.Fatal("Spec() ok = true, want false")
}
if keys := testCase.nilRegisteredKey(); keys != nil {
t.Fatalf("RegisteredKeys() = %#v, want nil", keys)
}
})
t.Run(testCase.name+"/unknown spec lookup", func(t *testing.T) {
registry := testCase.newRegistry()
if _, ok := testCase.spec(registry, "missing"); ok {
t.Fatal("Spec() ok = true, want false")
}
})
}
type registryChunker struct {
key string
}
func (chunker registryChunker) Key() string {
return chunker.key
}
func (chunker registryChunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) {
return contracts.ChunkResult{}, nil
}
type registryMerger struct {
key string
}
func (merger registryMerger) Key() string {
return merger.key
}
func (merger registryMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) {
return contracts.MergeResult{}, nil
}
type registryNormalizer struct {
key string
}
func (normalizer registryNormalizer) Key() string {
return normalizer.key
}
func (normalizer registryNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
return contracts.NormalizeResult{}, nil
}
type registryOutputEncoder struct {
key string
}
func (encoder registryOutputEncoder) Key() string {
return encoder.key
}
func (encoder registryOutputEncoder) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
return contracts.OutputResult{}, nil
}
type registryValidator struct {
name string
}
func (validator registryValidator) Name() string {
return validator.name
}
func (validator registryValidator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) {
return contracts.ValidationResult{}, nil
}

View File

@@ -2,7 +2,6 @@ package pipeline
import (
"fmt"
"sort"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
@@ -12,31 +11,44 @@ type ExtractorConstructor func() (contracts.Extractor, error)
type ExtractorRegistry struct {
constructors map[string]ExtractorConstructor
specs map[string]ModuleSpec
}
func NewExtractorRegistry() *ExtractorRegistry {
return &ExtractorRegistry{
constructors: make(map[string]ExtractorConstructor),
specs: make(map[string]ModuleSpec),
}
}
func (r *ExtractorRegistry) Register(key string, constructor ExtractorConstructor) error {
return r.RegisterWithSpec(defaultModuleSpec(key, StageExtract), constructor)
}
func (r *ExtractorRegistry) RegisterWithSpec(spec ModuleSpec, constructor ExtractorConstructor) error {
if r == nil {
return fmt.Errorf("extractor registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
if normalizedKey == "" {
return fmt.Errorf("extractor key must not be empty")
normalizedSpec := normalizeModuleSpec(spec)
if err := validateModuleSpec("extractor", StageExtract, normalizedSpec); err != nil {
return err
}
if constructor == nil {
return fmt.Errorf("extractor constructor for %q must not be nil", normalizedKey)
return fmt.Errorf("extractor constructor for %q must not be nil", normalizedSpec.Key)
}
if _, ok := r.constructors[normalizedKey]; ok {
return fmt.Errorf("extractor %q is already registered", normalizedKey)
if _, ok := r.constructors[normalizedSpec.Key]; ok {
return fmt.Errorf("extractor %q is already registered", normalizedSpec.Key)
}
r.constructors[normalizedKey] = constructor
if r.constructors == nil {
r.constructors = make(map[string]ExtractorConstructor)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.constructors[normalizedSpec.Key] = constructor
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
@@ -69,15 +81,22 @@ func (r *ExtractorRegistry) Build(key string) (contracts.Extractor, error) {
return extractor, nil
}
func (r *ExtractorRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false
}
spec, ok := r.specs[strings.TrimSpace(key)]
if !ok {
return ModuleSpec{}, false
}
return cloneModuleSpec(spec), true
}
func (r *ExtractorRegistry) RegisteredKeys() []string {
if r == nil {
return nil
}
keys := make([]string, 0, len(r.constructors))
for key := range r.constructors {
keys = append(keys, key)
}
sort.Strings(keys)
return keys
return sortedRegistryKeys(r.constructors)
}

View File

@@ -42,6 +42,81 @@ func TestExtractorRegistryRegisterAndBuildTrimKeys(t *testing.T) {
}
}
func TestExtractorRegistryRegisterWithSpecStoresMetadata(t *testing.T) {
registry := NewExtractorRegistry()
spec := ModuleSpec{
Key: " generic-extractor ",
Stage: StageExtract,
Provides: []string{" generic-artifact ", "source-citations", "generic-artifact", ""},
Requires: []string{" source-document ", "source-document", ""},
}
if err := registry.RegisterWithSpec(spec, fakeExtractorConstructor("generic-extractor")); err != nil {
t.Fatalf("RegisterWithSpec() error = %v, want nil", err)
}
got, ok := registry.Spec("\tgeneric-extractor\n")
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{
Key: "generic-extractor",
Stage: StageExtract,
Provides: []string{"generic-artifact", "source-citations"},
Requires: []string{"source-document"},
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
}
got.Provides[0] = "changed"
again, ok := registry.Spec("generic-extractor")
if !ok {
t.Fatal("Spec() after caller mutation ok = false, want true")
}
if !reflect.DeepEqual(again, want) {
t.Fatalf("Spec() after caller mutation = %#v, want %#v", again, want)
}
}
func TestExtractorRegistryRegisterStoresDefaultSpec(t *testing.T) {
registry := NewExtractorRegistry()
if err := registry.Register(" generic-extractor ", fakeExtractorConstructor("generic-extractor")); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
got, ok := registry.Spec("generic-extractor")
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{Key: "generic-extractor", Stage: StageExtract}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
}
}
func TestExtractorRegistryRegisterWithSpecRejectsWrongStage(t *testing.T) {
registry := NewExtractorRegistry()
err := registry.RegisterWithSpec(ModuleSpec{Key: "generic-extractor", Stage: StageInput}, fakeExtractorConstructor("generic-extractor"))
if err == nil {
t.Fatal("RegisterWithSpec() error = nil, want error")
}
if !strings.Contains(err.Error(), "stage") {
t.Fatalf("RegisterWithSpec() error = %q, want stage error", err.Error())
}
}
func TestExtractorRegistrySpecRejectsUnknownKey(t *testing.T) {
registry := NewExtractorRegistry()
if _, ok := registry.Spec("missing-extractor"); ok {
t.Fatal("Spec() ok = true, want false")
}
}
func TestExtractorRegistryRegisterRejectsEmptyKey(t *testing.T) {
registry := NewExtractorRegistry()
@@ -183,6 +258,9 @@ func TestExtractorRegistryNilRegistryBehavior(t *testing.T) {
if _, err := registry.Build("generic-extractor"); err == nil {
t.Fatal("Build() error = nil, want error")
}
if _, ok := registry.Spec("generic-extractor"); ok {
t.Fatal("Spec() ok = true, want false")
}
if keys := registry.RegisteredKeys(); keys != nil {
t.Fatalf("RegisteredKeys() = %#v, want nil", keys)
}

View File

@@ -0,0 +1,70 @@
package pipeline
import (
"context"
"encoding/json"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type AppendOrderMerger struct{}
func (m AppendOrderMerger) Key() string {
return DefaultMergeModule
}
func (m AppendOrderMerger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) {
var candidates []artifacts.ArtifactCandidate
for _, chunkArtifacts := range req.ChunkArtifacts {
candidates = append(candidates, copyArtifactCandidates(chunkArtifacts.Candidates)...)
}
return contracts.MergeResult{Candidates: candidates}, nil
}
type NoopNormalizer struct{}
func (n NoopNormalizer) Key() string {
return DefaultNormalizeModule
}
func (n NoopNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
return contracts.NormalizeResult{Candidates: copyArtifactCandidates(req.Candidates)}, nil
}
func copyArtifactCandidates(candidates []artifacts.ArtifactCandidate) []artifacts.ArtifactCandidate {
if len(candidates) == 0 {
return nil
}
copied := make([]artifacts.ArtifactCandidate, 0, len(candidates))
for _, candidate := range candidates {
copied = append(copied, copyArtifactCandidate(candidate))
}
return copied
}
func copyArtifactCandidate(candidate artifacts.ArtifactCandidate) artifacts.ArtifactCandidate {
return artifacts.ArtifactCandidate{
Index: candidate.Index,
ExtractorKey: candidate.ExtractorKey,
ArtifactType: candidate.ArtifactType,
SchemaVersion: candidate.SchemaVersion,
Payload: append(json.RawMessage(nil), candidate.Payload...),
SourceRefs: append([]source.SourceRef(nil), candidate.SourceRefs...),
Metadata: copyArtifactMetadata(candidate.Metadata),
}
}
func copyArtifactMetadata(metadata map[string]any) map[string]any {
if len(metadata) == 0 {
return nil
}
copied := make(map[string]any, len(metadata))
for key, value := range metadata {
copied[key] = value
}
return copied
}

View File

@@ -0,0 +1,221 @@
package pipeline
import (
"context"
"encoding/json"
"reflect"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestGenericMergeAndNormalizeKeys(t *testing.T) {
merger := AppendOrderMerger{}
normalizer := NoopNormalizer{}
if merger.Key() != DefaultMergeModule {
t.Fatalf("AppendOrderMerger.Key() = %q, want %q", merger.Key(), DefaultMergeModule)
}
if normalizer.Key() != DefaultNormalizeModule {
t.Fatalf("NoopNormalizer.Key() = %q, want %q", normalizer.Key(), DefaultNormalizeModule)
}
}
func TestAppendOrderMergerConcatenatesByChunkAndCandidateOrder(t *testing.T) {
merger := AppendOrderMerger{}
chunks := []contracts.ChunkArtifacts{
{
Chunk: sourceChunk(0),
Candidates: []artifacts.ArtifactCandidate{
candidate(2, "first-b"),
candidate(1, "first-a"),
},
},
{
Chunk: sourceChunk(1),
Candidates: []artifacts.ArtifactCandidate{
candidate(4, "second-b"),
candidate(3, "second-a"),
},
},
}
result, err := merger.Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: chunks})
if err != nil {
t.Fatalf("Merge() error = %v, want nil", err)
}
if len(result.Warnings) != 0 {
t.Fatalf("Warnings = %#v, want none", result.Warnings)
}
got := candidateNames(result.Candidates)
want := []string{"first-b", "first-a", "second-b", "second-a"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("candidate order = %#v, want %#v", got, want)
}
}
func TestAppendOrderMergerReturnsMutationSafeCandidates(t *testing.T) {
merger := AppendOrderMerger{}
input := []contracts.ChunkArtifacts{
{
Chunk: sourceChunk(0),
Candidates: []artifacts.ArtifactCandidate{
candidate(1, "original"),
},
},
}
result, err := merger.Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: input})
if err != nil {
t.Fatalf("Merge() error = %v, want nil", err)
}
if len(result.Candidates) != 1 {
t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates))
}
input[0].Candidates[0].Index = 99
input[0].Candidates[0].Payload[0] = '['
input[0].Candidates[0].SourceRefs[0].StartUnitID = "changed"
input[0].Candidates[0].Metadata["name"] = "changed"
got := result.Candidates[0]
if got.Index != 1 {
t.Fatalf("Index = %d, want 1", got.Index)
}
if string(got.Payload) != `{"name":"original"}` {
t.Fatalf("Payload = %s, want original payload", got.Payload)
}
if got.SourceRefs[0].StartUnitID != "u1" {
t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs)
}
if got.Metadata["name"] != "original" {
t.Fatalf("Metadata = %#v, want original metadata", got.Metadata)
}
}
func TestNoopNormalizerPreservesOrderAndValues(t *testing.T) {
normalizer := NoopNormalizer{}
input := []artifacts.ArtifactCandidate{
candidate(3, "third"),
candidate(1, "first"),
candidate(2, "second"),
}
result, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input})
if err != nil {
t.Fatalf("Normalize() error = %v, want nil", err)
}
if len(result.Warnings) != 0 {
t.Fatalf("Warnings = %#v, want none", result.Warnings)
}
got := candidateNames(result.Candidates)
want := []string{"third", "first", "second"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("candidate order = %#v, want %#v", got, want)
}
if !reflect.DeepEqual(result.Candidates[0].SourceRefs, input[0].SourceRefs) {
t.Fatalf("SourceRefs = %#v, want %#v", result.Candidates[0].SourceRefs, input[0].SourceRefs)
}
if !reflect.DeepEqual(result.Candidates[0].Metadata, input[0].Metadata) {
t.Fatalf("Metadata = %#v, want %#v", result.Candidates[0].Metadata, input[0].Metadata)
}
}
func TestNoopNormalizerReturnsMutationSafeCandidates(t *testing.T) {
normalizer := NoopNormalizer{}
input := []artifacts.ArtifactCandidate{candidate(1, "original")}
result, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{Candidates: input})
if err != nil {
t.Fatalf("Normalize() error = %v, want nil", err)
}
if len(result.Candidates) != 1 {
t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates))
}
input[0].Index = 99
input[0].Payload[0] = '['
input[0].SourceRefs[0].EndUnitID = "changed"
input[0].Metadata["name"] = "changed"
got := result.Candidates[0]
if got.Index != 1 {
t.Fatalf("Index = %d, want 1", got.Index)
}
if string(got.Payload) != `{"name":"original"}` {
t.Fatalf("Payload = %s, want original payload", got.Payload)
}
if got.SourceRefs[0].EndUnitID != "u1" {
t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs)
}
if got.Metadata["name"] != "original" {
t.Fatalf("Metadata = %#v, want original metadata", got.Metadata)
}
}
func TestGenericMergeAndNormalizeHandleEmptyInput(t *testing.T) {
merger := AppendOrderMerger{}
normalizer := NoopNormalizer{}
mergeResult, err := merger.Merge(context.Background(), contracts.MergeRequest{})
if err != nil {
t.Fatalf("Merge() error = %v, want nil", err)
}
if len(mergeResult.Candidates) != 0 {
t.Fatalf("len(mergeResult.Candidates) = %d, want 0", len(mergeResult.Candidates))
}
if len(mergeResult.Warnings) != 0 {
t.Fatalf("merge warnings = %#v, want none", mergeResult.Warnings)
}
normalizeResult, err := normalizer.Normalize(context.Background(), contracts.NormalizeRequest{})
if err != nil {
t.Fatalf("Normalize() error = %v, want nil", err)
}
if len(normalizeResult.Candidates) != 0 {
t.Fatalf("len(normalizeResult.Candidates) = %d, want 0", len(normalizeResult.Candidates))
}
if len(normalizeResult.Warnings) != 0 {
t.Fatalf("normalize warnings = %#v, want none", normalizeResult.Warnings)
}
}
func candidate(index int, name string) artifacts.ArtifactCandidate {
return artifacts.ArtifactCandidate{
Index: index,
ExtractorKey: "generic-extractor",
ArtifactType: "generic-artifact",
SchemaVersion: "v1",
Payload: json.RawMessage(`{"name":"` + name + `"}`),
SourceRefs: []source.SourceRef{
{SourceID: "source-1", StartUnitID: "u1", EndUnitID: "u1"},
},
Metadata: map[string]any{
"name": name,
},
}
}
func candidateNames(candidates []artifacts.ArtifactCandidate) []string {
names := make([]string, 0, len(candidates))
for _, candidate := range candidates {
names = append(names, candidate.Metadata["name"].(string))
}
return names
}
func sourceChunk(index int) contracts.SourceChunk {
return contracts.SourceChunk{
ID: "chunk",
SourceID: "source-1",
Index: index,
Units: []source.SourceUnit{
{ID: "u1", Kind: "unit", Text: "Source unit."},
},
}
}

View File

@@ -2,7 +2,6 @@ package pipeline
import (
"fmt"
"sort"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
@@ -12,31 +11,44 @@ type InputAdapterConstructor func() (contracts.InputAdapter, error)
type InputAdapterRegistry struct {
constructors map[string]InputAdapterConstructor
specs map[string]ModuleSpec
}
func NewInputAdapterRegistry() *InputAdapterRegistry {
return &InputAdapterRegistry{
constructors: make(map[string]InputAdapterConstructor),
specs: make(map[string]ModuleSpec),
}
}
func (r *InputAdapterRegistry) Register(key string, constructor InputAdapterConstructor) error {
return r.RegisterWithSpec(defaultModuleSpec(key, StageInput), constructor)
}
func (r *InputAdapterRegistry) RegisterWithSpec(spec ModuleSpec, constructor InputAdapterConstructor) error {
if r == nil {
return fmt.Errorf("input adapter registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
if normalizedKey == "" {
return fmt.Errorf("input adapter key must not be empty")
normalizedSpec := normalizeModuleSpec(spec)
if err := validateModuleSpec("input adapter", StageInput, normalizedSpec); err != nil {
return err
}
if constructor == nil {
return fmt.Errorf("input adapter constructor for %q must not be nil", normalizedKey)
return fmt.Errorf("input adapter constructor for %q must not be nil", normalizedSpec.Key)
}
if _, ok := r.constructors[normalizedKey]; ok {
return fmt.Errorf("input adapter %q is already registered", normalizedKey)
if _, ok := r.constructors[normalizedSpec.Key]; ok {
return fmt.Errorf("input adapter %q is already registered", normalizedSpec.Key)
}
r.constructors[normalizedKey] = constructor
if r.constructors == nil {
r.constructors = make(map[string]InputAdapterConstructor)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.constructors[normalizedSpec.Key] = constructor
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
@@ -69,15 +81,22 @@ func (r *InputAdapterRegistry) Build(key string) (contracts.InputAdapter, error)
return adapter, nil
}
func (r *InputAdapterRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false
}
spec, ok := r.specs[strings.TrimSpace(key)]
if !ok {
return ModuleSpec{}, false
}
return cloneModuleSpec(spec), true
}
func (r *InputAdapterRegistry) RegisteredKeys() []string {
if r == nil {
return nil
}
keys := make([]string, 0, len(r.constructors))
for key := range r.constructors {
keys = append(keys, key)
}
sort.Strings(keys)
return keys
return sortedRegistryKeys(r.constructors)
}

View File

@@ -43,6 +43,81 @@ func TestInputAdapterRegistryRegisterAndBuildTrimKeys(t *testing.T) {
}
}
func TestInputAdapterRegistryRegisterWithSpecStoresMetadata(t *testing.T) {
registry := NewInputAdapterRegistry()
spec := ModuleSpec{
Key: " generic-input ",
Stage: StageInput,
Provides: []string{" parsed-source ", "source-document", "parsed-source", ""},
Requires: []string{" raw-bytes ", "raw-bytes", ""},
}
if err := registry.RegisterWithSpec(spec, fakeInputAdapterConstructor("generic-input")); err != nil {
t.Fatalf("RegisterWithSpec() error = %v, want nil", err)
}
got, ok := registry.Spec("\tgeneric-input\n")
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{
Key: "generic-input",
Stage: StageInput,
Provides: []string{"parsed-source", "source-document"},
Requires: []string{"raw-bytes"},
}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
}
got.Provides[0] = "changed"
again, ok := registry.Spec("generic-input")
if !ok {
t.Fatal("Spec() after caller mutation ok = false, want true")
}
if !reflect.DeepEqual(again, want) {
t.Fatalf("Spec() after caller mutation = %#v, want %#v", again, want)
}
}
func TestInputAdapterRegistryRegisterStoresDefaultSpec(t *testing.T) {
registry := NewInputAdapterRegistry()
if err := registry.Register(" generic-input ", fakeInputAdapterConstructor("generic-input")); err != nil {
t.Fatalf("Register() error = %v, want nil", err)
}
got, ok := registry.Spec("generic-input")
if !ok {
t.Fatal("Spec() ok = false, want true")
}
want := ModuleSpec{Key: "generic-input", Stage: StageInput}
if !reflect.DeepEqual(got, want) {
t.Fatalf("Spec() = %#v, want %#v", got, want)
}
}
func TestInputAdapterRegistryRegisterWithSpecRejectsWrongStage(t *testing.T) {
registry := NewInputAdapterRegistry()
err := registry.RegisterWithSpec(ModuleSpec{Key: "generic-input", Stage: StageExtract}, fakeInputAdapterConstructor("generic-input"))
if err == nil {
t.Fatal("RegisterWithSpec() error = nil, want error")
}
if !strings.Contains(err.Error(), "stage") {
t.Fatalf("RegisterWithSpec() error = %q, want stage error", err.Error())
}
}
func TestInputAdapterRegistrySpecRejectsUnknownKey(t *testing.T) {
registry := NewInputAdapterRegistry()
if _, ok := registry.Spec("missing-input"); ok {
t.Fatal("Spec() ok = true, want false")
}
}
func TestInputAdapterRegistryRegisterRejectsEmptyKey(t *testing.T) {
registry := NewInputAdapterRegistry()
@@ -184,6 +259,9 @@ func TestInputAdapterRegistryNilRegistryBehavior(t *testing.T) {
if _, err := registry.Build("generic-input"); err == nil {
t.Fatal("Build() error = nil, want error")
}
if _, ok := registry.Spec("generic-input"); ok {
t.Fatal("Spec() ok = true, want false")
}
if keys := registry.RegisteredKeys(); keys != nil {
t.Fatalf("RegisteredKeys() = %#v, want nil", keys)
}

View File

@@ -0,0 +1,102 @@
package pipeline
import (
"fmt"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type MergerConstructor func() (contracts.Merger, error)
type MergerRegistry struct {
constructors map[string]MergerConstructor
specs map[string]ModuleSpec
}
func NewMergerRegistry() *MergerRegistry {
return &MergerRegistry{
constructors: make(map[string]MergerConstructor),
specs: make(map[string]ModuleSpec),
}
}
func (r *MergerRegistry) Register(key string, constructor MergerConstructor) error {
return r.RegisterWithSpec(defaultModuleSpec(key, StageMerge), constructor)
}
func (r *MergerRegistry) RegisterWithSpec(spec ModuleSpec, constructor MergerConstructor) error {
if r == nil {
return fmt.Errorf("merger registry must not be nil")
}
normalizedSpec := normalizeModuleSpec(spec)
if err := validateModuleSpec("merger", StageMerge, normalizedSpec); err != nil {
return err
}
if constructor == nil {
return fmt.Errorf("merger constructor for %q must not be nil", normalizedSpec.Key)
}
if _, ok := r.constructors[normalizedSpec.Key]; ok {
return fmt.Errorf("merger %q is already registered", normalizedSpec.Key)
}
if r.constructors == nil {
r.constructors = make(map[string]MergerConstructor)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.constructors[normalizedSpec.Key] = constructor
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
func (r *MergerRegistry) Build(key string) (contracts.Merger, error) {
if r == nil {
return nil, fmt.Errorf("merger registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
if normalizedKey == "" {
return nil, fmt.Errorf("merger key must not be empty")
}
constructor, ok := r.constructors[normalizedKey]
if !ok {
return nil, fmt.Errorf("merger %q is not registered", normalizedKey)
}
merger, err := constructor()
if err != nil {
return nil, fmt.Errorf("build merger %q: %w", normalizedKey, err)
}
if merger == nil {
return nil, fmt.Errorf("merger %q constructor returned nil", normalizedKey)
}
if merger.Key() != normalizedKey {
return nil, fmt.Errorf("merger %q returned key %q", normalizedKey, merger.Key())
}
return merger, nil
}
func (r *MergerRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false
}
spec, ok := r.specs[strings.TrimSpace(key)]
if !ok {
return ModuleSpec{}, false
}
return cloneModuleSpec(spec), true
}
func (r *MergerRegistry) RegisteredKeys() []string {
if r == nil {
return nil
}
return sortedRegistryKeys(r.constructors)
}

View File

@@ -0,0 +1,58 @@
package pipeline
import (
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestMergerRegistryBehavior(t *testing.T) {
runRegistryBehaviorTests(t, registryBehaviorCase[contracts.Merger]{
name: "MergerRegistry",
key: "generic-merger",
stage: StageMerge,
wrongStage: StageExtract,
newRegistry: func() any {
return NewMergerRegistry()
},
register: func(registry any, key string, constructor func() (contracts.Merger, error)) error {
return registry.(*MergerRegistry).Register(key, constructor)
},
registerWithSpec: func(registry any, spec ModuleSpec, constructor func() (contracts.Merger, error)) error {
return registry.(*MergerRegistry).RegisterWithSpec(spec, constructor)
},
build: func(registry any, key string) (contracts.Merger, error) {
return registry.(*MergerRegistry).Build(key)
},
spec: func(registry any, key string) (ModuleSpec, bool) {
return registry.(*MergerRegistry).Spec(key)
},
registeredKeys: func(registry any) []string {
return registry.(*MergerRegistry).RegisteredKeys()
},
nilRegister: func(key string, constructor func() (contracts.Merger, error)) error {
var registry *MergerRegistry
return registry.Register(key, constructor)
},
nilBuild: func(key string) (contracts.Merger, error) {
var registry *MergerRegistry
return registry.Build(key)
},
nilSpec: func(key string) (ModuleSpec, bool) {
var registry *MergerRegistry
return registry.Spec(key)
},
nilRegisteredKey: func() []string {
var registry *MergerRegistry
return registry.RegisteredKeys()
},
constructor: func(key string) func() (contracts.Merger, error) {
return func() (contracts.Merger, error) {
return registryMerger{key: key}, nil
}
},
moduleKey: func(module contracts.Merger) string {
return module.Key()
},
})
}

View File

@@ -0,0 +1,99 @@
package pipeline
import (
"fmt"
"sort"
"strings"
)
type ModuleStage string
const (
StageInput ModuleStage = "input"
StageChunk ModuleStage = "chunk"
StageExtract ModuleStage = "extract"
StageMerge ModuleStage = "merge"
StageNormalize ModuleStage = "normalize"
StageValidate ModuleStage = "validate"
StageOutput ModuleStage = "output"
)
type ModuleSpec struct {
Key string
Stage ModuleStage
Provides []string
Requires []string
}
func defaultModuleSpec(key string, stage ModuleStage) ModuleSpec {
return ModuleSpec{
Key: key,
Stage: stage,
}
}
func normalizeModuleSpec(spec ModuleSpec) ModuleSpec {
return ModuleSpec{
Key: strings.TrimSpace(spec.Key),
Stage: spec.Stage,
Provides: normalizeCapabilities(spec.Provides),
Requires: normalizeCapabilities(spec.Requires),
}
}
func normalizeCapabilities(values []string) []string {
if len(values) == 0 {
return nil
}
seen := make(map[string]struct{}, len(values))
for _, value := range values {
normalized := strings.TrimSpace(value)
if normalized == "" {
continue
}
seen[normalized] = struct{}{}
}
if len(seen) == 0 {
return nil
}
capabilities := make([]string, 0, len(seen))
for value := range seen {
capabilities = append(capabilities, value)
}
sort.Strings(capabilities)
return capabilities
}
func cloneModuleSpec(spec ModuleSpec) ModuleSpec {
return ModuleSpec{
Key: spec.Key,
Stage: spec.Stage,
Provides: append([]string(nil), spec.Provides...),
Requires: append([]string(nil), spec.Requires...),
}
}
func validateModuleSpec(kind string, expectedStage ModuleStage, spec ModuleSpec) error {
if spec.Key == "" {
return fmt.Errorf("%s key must not be empty", kind)
}
if spec.Stage != expectedStage {
return fmt.Errorf("%s %q must use %q stage, got %q", kind, spec.Key, expectedStage, spec.Stage)
}
return nil
}
func sortedRegistryKeys[C any](constructors map[string]C) []string {
if len(constructors) == 0 {
return nil
}
keys := make([]string, 0, len(constructors))
for key := range constructors {
keys = append(keys, key)
}
sort.Strings(keys)
return keys
}

View File

@@ -0,0 +1,102 @@
package pipeline
import (
"fmt"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type NormalizerConstructor func() (contracts.Normalizer, error)
type NormalizerRegistry struct {
constructors map[string]NormalizerConstructor
specs map[string]ModuleSpec
}
func NewNormalizerRegistry() *NormalizerRegistry {
return &NormalizerRegistry{
constructors: make(map[string]NormalizerConstructor),
specs: make(map[string]ModuleSpec),
}
}
func (r *NormalizerRegistry) Register(key string, constructor NormalizerConstructor) error {
return r.RegisterWithSpec(defaultModuleSpec(key, StageNormalize), constructor)
}
func (r *NormalizerRegistry) RegisterWithSpec(spec ModuleSpec, constructor NormalizerConstructor) error {
if r == nil {
return fmt.Errorf("normalizer registry must not be nil")
}
normalizedSpec := normalizeModuleSpec(spec)
if err := validateModuleSpec("normalizer", StageNormalize, normalizedSpec); err != nil {
return err
}
if constructor == nil {
return fmt.Errorf("normalizer constructor for %q must not be nil", normalizedSpec.Key)
}
if _, ok := r.constructors[normalizedSpec.Key]; ok {
return fmt.Errorf("normalizer %q is already registered", normalizedSpec.Key)
}
if r.constructors == nil {
r.constructors = make(map[string]NormalizerConstructor)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.constructors[normalizedSpec.Key] = constructor
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
func (r *NormalizerRegistry) Build(key string) (contracts.Normalizer, error) {
if r == nil {
return nil, fmt.Errorf("normalizer registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
if normalizedKey == "" {
return nil, fmt.Errorf("normalizer key must not be empty")
}
constructor, ok := r.constructors[normalizedKey]
if !ok {
return nil, fmt.Errorf("normalizer %q is not registered", normalizedKey)
}
normalizer, err := constructor()
if err != nil {
return nil, fmt.Errorf("build normalizer %q: %w", normalizedKey, err)
}
if normalizer == nil {
return nil, fmt.Errorf("normalizer %q constructor returned nil", normalizedKey)
}
if normalizer.Key() != normalizedKey {
return nil, fmt.Errorf("normalizer %q returned key %q", normalizedKey, normalizer.Key())
}
return normalizer, nil
}
func (r *NormalizerRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false
}
spec, ok := r.specs[strings.TrimSpace(key)]
if !ok {
return ModuleSpec{}, false
}
return cloneModuleSpec(spec), true
}
func (r *NormalizerRegistry) RegisteredKeys() []string {
if r == nil {
return nil
}
return sortedRegistryKeys(r.constructors)
}

View File

@@ -0,0 +1,58 @@
package pipeline
import (
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestNormalizerRegistryBehavior(t *testing.T) {
runRegistryBehaviorTests(t, registryBehaviorCase[contracts.Normalizer]{
name: "NormalizerRegistry",
key: "generic-normalizer",
stage: StageNormalize,
wrongStage: StageExtract,
newRegistry: func() any {
return NewNormalizerRegistry()
},
register: func(registry any, key string, constructor func() (contracts.Normalizer, error)) error {
return registry.(*NormalizerRegistry).Register(key, constructor)
},
registerWithSpec: func(registry any, spec ModuleSpec, constructor func() (contracts.Normalizer, error)) error {
return registry.(*NormalizerRegistry).RegisterWithSpec(spec, constructor)
},
build: func(registry any, key string) (contracts.Normalizer, error) {
return registry.(*NormalizerRegistry).Build(key)
},
spec: func(registry any, key string) (ModuleSpec, bool) {
return registry.(*NormalizerRegistry).Spec(key)
},
registeredKeys: func(registry any) []string {
return registry.(*NormalizerRegistry).RegisteredKeys()
},
nilRegister: func(key string, constructor func() (contracts.Normalizer, error)) error {
var registry *NormalizerRegistry
return registry.Register(key, constructor)
},
nilBuild: func(key string) (contracts.Normalizer, error) {
var registry *NormalizerRegistry
return registry.Build(key)
},
nilSpec: func(key string) (ModuleSpec, bool) {
var registry *NormalizerRegistry
return registry.Spec(key)
},
nilRegisteredKey: func() []string {
var registry *NormalizerRegistry
return registry.RegisteredKeys()
},
constructor: func(key string) func() (contracts.Normalizer, error) {
return func() (contracts.Normalizer, error) {
return registryNormalizer{key: key}, nil
}
},
moduleKey: func(module contracts.Normalizer) string {
return module.Key()
},
})
}

View File

@@ -0,0 +1,102 @@
package pipeline
import (
"fmt"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type OutputEncoderConstructor func() (contracts.OutputEncoder, error)
type OutputEncoderRegistry struct {
constructors map[string]OutputEncoderConstructor
specs map[string]ModuleSpec
}
func NewOutputEncoderRegistry() *OutputEncoderRegistry {
return &OutputEncoderRegistry{
constructors: make(map[string]OutputEncoderConstructor),
specs: make(map[string]ModuleSpec),
}
}
func (r *OutputEncoderRegistry) Register(key string, constructor OutputEncoderConstructor) error {
return r.RegisterWithSpec(defaultModuleSpec(key, StageOutput), constructor)
}
func (r *OutputEncoderRegistry) RegisterWithSpec(spec ModuleSpec, constructor OutputEncoderConstructor) error {
if r == nil {
return fmt.Errorf("output encoder registry must not be nil")
}
normalizedSpec := normalizeModuleSpec(spec)
if err := validateModuleSpec("output encoder", StageOutput, normalizedSpec); err != nil {
return err
}
if constructor == nil {
return fmt.Errorf("output encoder constructor for %q must not be nil", normalizedSpec.Key)
}
if _, ok := r.constructors[normalizedSpec.Key]; ok {
return fmt.Errorf("output encoder %q is already registered", normalizedSpec.Key)
}
if r.constructors == nil {
r.constructors = make(map[string]OutputEncoderConstructor)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.constructors[normalizedSpec.Key] = constructor
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
func (r *OutputEncoderRegistry) Build(key string) (contracts.OutputEncoder, error) {
if r == nil {
return nil, fmt.Errorf("output encoder registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
if normalizedKey == "" {
return nil, fmt.Errorf("output encoder key must not be empty")
}
constructor, ok := r.constructors[normalizedKey]
if !ok {
return nil, fmt.Errorf("output encoder %q is not registered", normalizedKey)
}
encoder, err := constructor()
if err != nil {
return nil, fmt.Errorf("build output encoder %q: %w", normalizedKey, err)
}
if encoder == nil {
return nil, fmt.Errorf("output encoder %q constructor returned nil", normalizedKey)
}
if encoder.Key() != normalizedKey {
return nil, fmt.Errorf("output encoder %q returned key %q", normalizedKey, encoder.Key())
}
return encoder, nil
}
func (r *OutputEncoderRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false
}
spec, ok := r.specs[strings.TrimSpace(key)]
if !ok {
return ModuleSpec{}, false
}
return cloneModuleSpec(spec), true
}
func (r *OutputEncoderRegistry) RegisteredKeys() []string {
if r == nil {
return nil
}
return sortedRegistryKeys(r.constructors)
}

View File

@@ -0,0 +1,58 @@
package pipeline
import (
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestOutputEncoderRegistryBehavior(t *testing.T) {
runRegistryBehaviorTests(t, registryBehaviorCase[contracts.OutputEncoder]{
name: "OutputEncoderRegistry",
key: "generic-output",
stage: StageOutput,
wrongStage: StageExtract,
newRegistry: func() any {
return NewOutputEncoderRegistry()
},
register: func(registry any, key string, constructor func() (contracts.OutputEncoder, error)) error {
return registry.(*OutputEncoderRegistry).Register(key, constructor)
},
registerWithSpec: func(registry any, spec ModuleSpec, constructor func() (contracts.OutputEncoder, error)) error {
return registry.(*OutputEncoderRegistry).RegisterWithSpec(spec, constructor)
},
build: func(registry any, key string) (contracts.OutputEncoder, error) {
return registry.(*OutputEncoderRegistry).Build(key)
},
spec: func(registry any, key string) (ModuleSpec, bool) {
return registry.(*OutputEncoderRegistry).Spec(key)
},
registeredKeys: func(registry any) []string {
return registry.(*OutputEncoderRegistry).RegisteredKeys()
},
nilRegister: func(key string, constructor func() (contracts.OutputEncoder, error)) error {
var registry *OutputEncoderRegistry
return registry.Register(key, constructor)
},
nilBuild: func(key string) (contracts.OutputEncoder, error) {
var registry *OutputEncoderRegistry
return registry.Build(key)
},
nilSpec: func(key string) (ModuleSpec, bool) {
var registry *OutputEncoderRegistry
return registry.Spec(key)
},
nilRegisteredKey: func() []string {
var registry *OutputEncoderRegistry
return registry.RegisteredKeys()
},
constructor: func(key string) func() (contracts.OutputEncoder, error) {
return func() (contracts.OutputEncoder, error) {
return registryOutputEncoder{key: key}, nil
}
},
moduleKey: func(module contracts.OutputEncoder) string {
return module.Key()
},
})
}

View File

@@ -0,0 +1,404 @@
package pipeline
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"sort"
"strings"
)
const (
DefaultChunkModule = "generic"
DefaultMergeModule = "appendorder"
DefaultNormalizeModule = "noop"
DefaultOutputModule = "json"
DefaultLLMProfile = "default"
)
type ModuleBinding struct {
Module string `json:"module"`
LLMProfile string `json:"llm_profile,omitempty"`
Options map[string]any `json:"options,omitempty"`
}
type ArtifactLaneProfile struct {
Extract ModuleBinding `json:"extract"`
Merge ModuleBinding `json:"merge,omitempty"`
Normalize ModuleBinding `json:"normalize,omitempty"`
Validators []ModuleBinding `json:"validators,omitempty"`
}
type PipelineProfile struct {
ID string `json:"id"`
Input ModuleBinding `json:"input"`
Chunk ModuleBinding `json:"chunk,omitempty"`
Artifacts map[string]ArtifactLaneProfile `json:"artifacts"`
Output ModuleBinding `json:"output,omitempty"`
}
type ResolveOptions struct {
Only []string
}
type ResolvedArtifactLane struct {
ID string
Extract ModuleBinding
Merge ModuleBinding
Normalize ModuleBinding
Validators []ModuleBinding
}
type ResolvedPipeline struct {
ID string
Digest string
Input ModuleBinding
Chunk ModuleBinding
ArtifactLanes []ResolvedArtifactLane
Output ModuleBinding
}
type ModuleCatalog struct {
Inputs *InputAdapterRegistry
Chunkers *ChunkerRegistry
Extractors *ExtractorRegistry
Mergers *MergerRegistry
Normalizers *NormalizerRegistry
Validators *ValidatorRegistry
Outputs *OutputEncoderRegistry
}
func Binding(module string) ModuleBinding {
return ModuleBinding{Module: strings.TrimSpace(module)}
}
func ResolvePipeline(profile PipelineProfile, options ResolveOptions, catalog ModuleCatalog) (ResolvedPipeline, error) {
pipelineID := strings.TrimSpace(profile.ID)
if pipelineID == "" {
return ResolvedPipeline{}, fmt.Errorf("pipeline id must not be empty")
}
if len(profile.Artifacts) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must declare at least one artifact lane", pipelineID)
}
input := resolveBinding(profile.Input, "")
if input.Module == "" {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q input module must not be empty", pipelineID)
}
inputModuleSpec, err := inputSpec(catalog, input.Module)
if err != nil {
return ResolvedPipeline{}, moduleLookupError(pipelineID, "", StageInput, input.Module, err)
}
capabilities := newCapabilitySet()
if missing, ok := capabilities.missing(inputModuleSpec.Requires); ok {
return ResolvedPipeline{}, capabilityError(pipelineID, "", StageInput, input.Module, missing)
}
capabilities.add(inputModuleSpec.Provides...)
chunk := resolveBinding(profile.Chunk, DefaultChunkModule)
chunkSpec, err := chunkerSpec(catalog, chunk.Module)
if err != nil {
return ResolvedPipeline{}, moduleLookupError(pipelineID, "", StageChunk, chunk.Module, err)
}
if missing, ok := capabilities.missing(chunkSpec.Requires); ok {
return ResolvedPipeline{}, capabilityError(pipelineID, "", StageChunk, chunk.Module, missing)
}
capabilities.add(chunkSpec.Provides...)
lanesByID, selectedLaneIDs, err := selectedArtifactLanes(pipelineID, profile.Artifacts, options)
if err != nil {
return ResolvedPipeline{}, err
}
if len(selectedLaneIDs) == 0 {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q must select at least one artifact lane", pipelineID)
}
resolved := ResolvedPipeline{
ID: pipelineID,
Input: input,
Chunk: chunk,
Output: resolveBinding(profile.Output, DefaultOutputModule),
}
outputCapabilities := capabilities.clone()
for _, laneID := range selectedLaneIDs {
laneProfile := lanesByID[laneID]
lane, laneCapabilities, err := resolveArtifactLane(pipelineID, laneID, laneProfile, capabilities, catalog)
if err != nil {
return ResolvedPipeline{}, err
}
resolved.ArtifactLanes = append(resolved.ArtifactLanes, lane)
outputCapabilities.addSet(laneCapabilities)
}
outputSpec, err := outputSpec(catalog, resolved.Output.Module)
if err != nil {
return ResolvedPipeline{}, moduleLookupError(pipelineID, "", StageOutput, resolved.Output.Module, err)
}
if missing, ok := outputCapabilities.missing(outputSpec.Requires); ok {
return ResolvedPipeline{}, capabilityError(pipelineID, "", StageOutput, resolved.Output.Module, missing)
}
digest, err := resolvedPipelineDigest(resolved)
if err != nil {
return ResolvedPipeline{}, fmt.Errorf("pipeline %q digest: %w", pipelineID, err)
}
resolved.Digest = digest
return resolved, nil
}
func resolveArtifactLane(pipelineID, laneID string, profile ArtifactLaneProfile, inherited capabilitySet, catalog ModuleCatalog) (ResolvedArtifactLane, capabilitySet, error) {
lane := ResolvedArtifactLane{
ID: laneID,
Extract: resolveBinding(profile.Extract, ""),
Merge: resolveBinding(profile.Merge, DefaultMergeModule),
Normalize: resolveBinding(profile.Normalize, DefaultNormalizeModule),
Validators: resolveBindings(profile.Validators, ""),
}
if lane.Extract.Module == "" {
return ResolvedArtifactLane{}, nil, fmt.Errorf("pipeline %q lane %q extract module must not be empty", pipelineID, laneID)
}
capabilities := inherited.clone()
extractSpec, err := extractorSpec(catalog, lane.Extract.Module)
if err != nil {
return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageExtract, lane.Extract.Module, err)
}
if missing, ok := capabilities.missing(extractSpec.Requires); ok {
return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageExtract, lane.Extract.Module, missing)
}
capabilities.add(extractSpec.Provides...)
mergeSpec, err := mergerSpec(catalog, lane.Merge.Module)
if err != nil {
return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageMerge, lane.Merge.Module, err)
}
if missing, ok := capabilities.missing(mergeSpec.Requires); ok {
return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageMerge, lane.Merge.Module, missing)
}
capabilities.add(mergeSpec.Provides...)
normalizeSpec, err := normalizerSpec(catalog, lane.Normalize.Module)
if err != nil {
return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, err)
}
if missing, ok := capabilities.missing(normalizeSpec.Requires); ok {
return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageNormalize, lane.Normalize.Module, missing)
}
capabilities.add(normalizeSpec.Provides...)
for _, validator := range lane.Validators {
validatorSpec, err := validatorSpec(catalog, validator.Module)
if err != nil {
return ResolvedArtifactLane{}, nil, moduleLookupError(pipelineID, laneID, StageValidate, validator.Module, err)
}
if missing, ok := capabilities.missing(validatorSpec.Requires); ok {
return ResolvedArtifactLane{}, nil, capabilityError(pipelineID, laneID, StageValidate, validator.Module, missing)
}
capabilities.add(validatorSpec.Provides...)
}
return lane, capabilities, nil
}
func resolveBinding(binding ModuleBinding, defaultModule string) ModuleBinding {
module := strings.TrimSpace(binding.Module)
if module == "" {
module = defaultModule
}
llmProfile := strings.TrimSpace(binding.LLMProfile)
if llmProfile == "" {
llmProfile = DefaultLLMProfile
}
return ModuleBinding{
Module: module,
LLMProfile: llmProfile,
Options: cloneOptions(binding.Options),
}
}
func resolveBindings(bindings []ModuleBinding, defaultModule string) []ModuleBinding {
if len(bindings) == 0 {
return nil
}
resolved := make([]ModuleBinding, 0, len(bindings))
for _, binding := range bindings {
resolvedBinding := resolveBinding(binding, defaultModule)
resolved = append(resolved, resolvedBinding)
}
return resolved
}
func cloneOptions(options map[string]any) map[string]any {
if len(options) == 0 {
return nil
}
copied := make(map[string]any, len(options))
for key, value := range options {
copied[key] = value
}
return copied
}
func selectedArtifactLanes(pipelineID string, artifacts map[string]ArtifactLaneProfile, options ResolveOptions) (map[string]ArtifactLaneProfile, []string, error) {
lanesByID := make(map[string]ArtifactLaneProfile, len(artifacts))
for rawLaneID, lane := range artifacts {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
return nil, nil, fmt.Errorf("pipeline %q artifact lane id must not be empty", pipelineID)
}
if _, ok := lanesByID[laneID]; ok {
return nil, nil, fmt.Errorf("pipeline %q artifact lane %q is duplicated after trimming", pipelineID, laneID)
}
lanesByID[laneID] = lane
}
if len(options.Only) == 0 {
keys := make([]string, 0, len(lanesByID))
for laneID := range lanesByID {
keys = append(keys, laneID)
}
sort.Strings(keys)
return lanesByID, keys, nil
}
selected := make(map[string]struct{}, len(options.Only))
for _, rawLaneID := range options.Only {
laneID := strings.TrimSpace(rawLaneID)
if laneID == "" {
return nil, nil, fmt.Errorf("pipeline %q selected artifact lane id must not be empty", pipelineID)
}
if _, ok := lanesByID[laneID]; !ok {
return nil, nil, fmt.Errorf("pipeline %q selected artifact lane %q is not declared", pipelineID, laneID)
}
selected[laneID] = struct{}{}
}
keys := make([]string, 0, len(selected))
for laneID := range selected {
keys = append(keys, laneID)
}
sort.Strings(keys)
return lanesByID, keys, nil
}
func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) {
withoutDigest := struct {
ID string
Input ModuleBinding
Chunk ModuleBinding
ArtifactLanes []ResolvedArtifactLane
Output ModuleBinding
}{
ID: resolved.ID,
Input: resolved.Input,
Chunk: resolved.Chunk,
ArtifactLanes: resolved.ArtifactLanes,
Output: resolved.Output,
}
encoded, err := json.Marshal(withoutDigest)
if err != nil {
return "", err
}
sum := sha256.Sum256(encoded)
return "sha256:" + hex.EncodeToString(sum[:]), nil
}
func inputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Inputs, key)
}
func chunkerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Chunkers, key)
}
func extractorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Extractors, key)
}
func mergerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Mergers, key)
}
func normalizerSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Normalizers, key)
}
func validatorSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Validators, key)
}
func outputSpec(catalog ModuleCatalog, key string) (ModuleSpec, error) {
return registrySpec(catalog.Outputs, key)
}
type specRegistry interface {
Spec(key string) (ModuleSpec, bool)
}
func registrySpec(registry specRegistry, key string) (ModuleSpec, error) {
if registry == nil {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
}
spec, ok := registry.Spec(key)
if !ok {
return ModuleSpec{}, fmt.Errorf("module %q is not registered", key)
}
return spec, nil
}
func moduleLookupError(pipelineID, laneID string, stage ModuleStage, module string, err error) error {
if laneID != "" {
return fmt.Errorf("pipeline %q lane %q %s module %q: %w", pipelineID, laneID, stage, module, err)
}
return fmt.Errorf("pipeline %q %s module %q: %w", pipelineID, stage, module, err)
}
func capabilityError(pipelineID, laneID string, stage ModuleStage, module, capability string) error {
if laneID != "" {
return fmt.Errorf("pipeline %q lane %q %s module %q requires missing capability %q", pipelineID, laneID, stage, module, capability)
}
return fmt.Errorf("pipeline %q %s module %q requires missing capability %q", pipelineID, stage, module, capability)
}
type capabilitySet map[string]struct{}
func newCapabilitySet() capabilitySet {
return make(capabilitySet)
}
func (set capabilitySet) clone() capabilitySet {
copied := make(capabilitySet, len(set))
for capability := range set {
copied[capability] = struct{}{}
}
return copied
}
func (set capabilitySet) add(values ...string) {
for _, value := range values {
set[value] = struct{}{}
}
}
func (set capabilitySet) addSet(other capabilitySet) {
for value := range other {
set[value] = struct{}{}
}
}
func (set capabilitySet) missing(required []string) (string, bool) {
for _, capability := range required {
if _, ok := set[capability]; !ok {
return capability, true
}
}
return "", false
}

View File

@@ -0,0 +1,630 @@
package pipeline
import (
"context"
"encoding/json"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestResolvePipelineWithExplicitModules(t *testing.T) {
catalog := newProfileCatalog(t)
registerProfileSpecs(t, catalog,
ModuleSpec{Key: "window", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}},
ModuleSpec{Key: "record-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}},
ModuleSpec{Key: "dedupe", Stage: StageMerge, Requires: []string{"candidate"}, Provides: []string{"merged"}},
ModuleSpec{Key: "canonical", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}},
ModuleSpec{Key: "schema-check", Stage: StageValidate, Requires: []string{"normalized"}, Provides: []string{"validated"}},
ModuleSpec{Key: "ndjson", Stage: StageOutput, Requires: []string{"validated"}, Provides: []string{"encoded"}},
)
resolved, err := ResolvePipeline(PipelineProfile{
ID: " campaign ",
Input: ModuleBinding{Module: " text ", LLMProfile: " fast "},
Chunk: ModuleBinding{Module: " window ", Options: map[string]any{
"size": 10,
}},
Artifacts: map[string]ArtifactLaneProfile{
" records ": {
Extract: ModuleBinding{Module: " record-extractor ", LLMProfile: " careful "},
Merge: Binding(" dedupe "),
Normalize: Binding(" canonical "),
Validators: []ModuleBinding{Binding(" schema-check ")},
},
},
Output: Binding(" ndjson "),
}, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if resolved.ID != "campaign" {
t.Fatalf("ID = %q, want campaign", resolved.ID)
}
if !reflect.DeepEqual(resolved.Input, ModuleBinding{Module: "text", LLMProfile: "fast"}) {
t.Fatalf("Input = %#v, want trimmed explicit input", resolved.Input)
}
if resolved.Chunk.Module != "window" || resolved.Chunk.LLMProfile != DefaultLLMProfile {
t.Fatalf("Chunk = %#v, want explicit module and default LLM profile", resolved.Chunk)
}
if resolved.Chunk.Options["size"] != 10 {
t.Fatalf("Chunk.Options = %#v, want size option", resolved.Chunk.Options)
}
if len(resolved.ArtifactLanes) != 1 {
t.Fatalf("len(ArtifactLanes) = %d, want 1", len(resolved.ArtifactLanes))
}
lane := resolved.ArtifactLanes[0]
if lane.ID != "records" {
t.Fatalf("lane.ID = %q, want records", lane.ID)
}
if !reflect.DeepEqual(lane.Extract, ModuleBinding{Module: "record-extractor", LLMProfile: "careful"}) {
t.Fatalf("lane.Extract = %#v, want explicit extractor", lane.Extract)
}
if lane.Merge.Module != "dedupe" || lane.Normalize.Module != "canonical" {
t.Fatalf("lane merge/normalize = %#v/%#v, want explicit modules", lane.Merge, lane.Normalize)
}
if len(lane.Validators) != 1 || lane.Validators[0].Module != "schema-check" {
t.Fatalf("lane.Validators = %#v, want schema-check", lane.Validators)
}
if resolved.Output.Module != "ndjson" {
t.Fatalf("Output.Module = %q, want ndjson", resolved.Output.Module)
}
if !strings.HasPrefix(resolved.Digest, "sha256:") {
t.Fatalf("Digest = %q, want sha256 digest", resolved.Digest)
}
}
func TestResolvePipelineAppliesDefaults(t *testing.T) {
resolved, err := ResolvePipeline(PipelineProfile{
ID: "defaulted",
Input: Binding("text"),
Artifacts: map[string]ArtifactLaneProfile{
"events": {Extract: Binding("event-extractor")},
},
}, ResolveOptions{}, newProfileCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if resolved.Input.LLMProfile != DefaultLLMProfile {
t.Fatalf("Input.LLMProfile = %q, want %q", resolved.Input.LLMProfile, DefaultLLMProfile)
}
if !reflect.DeepEqual(resolved.Chunk, ModuleBinding{Module: DefaultChunkModule, LLMProfile: DefaultLLMProfile}) {
t.Fatalf("Chunk = %#v, want default chunk binding", resolved.Chunk)
}
if !reflect.DeepEqual(resolved.Output, ModuleBinding{Module: DefaultOutputModule, LLMProfile: DefaultLLMProfile}) {
t.Fatalf("Output = %#v, want default output binding", resolved.Output)
}
lane := resolved.ArtifactLanes[0]
if !reflect.DeepEqual(lane.Merge, ModuleBinding{Module: DefaultMergeModule, LLMProfile: DefaultLLMProfile}) {
t.Fatalf("Merge = %#v, want default merge binding", lane.Merge)
}
if !reflect.DeepEqual(lane.Normalize, ModuleBinding{Module: DefaultNormalizeModule, LLMProfile: DefaultLLMProfile}) {
t.Fatalf("Normalize = %#v, want default normalize binding", lane.Normalize)
}
if lane.Extract.LLMProfile != DefaultLLMProfile {
t.Fatalf("Extract.LLMProfile = %q, want %q", lane.Extract.LLMProfile, DefaultLLMProfile)
}
}
func TestResolvePipelineSelectsOnlyRequestedLanes(t *testing.T) {
profile := multiLaneProfile()
resolved, err := ResolvePipeline(profile, ResolveOptions{Only: []string{" summaries ", "events", "summaries"}}, newProfileCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
got := laneIDs(resolved.ArtifactLanes)
want := []string{"events", "summaries"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("lane IDs = %#v, want %#v", got, want)
}
}
func TestResolvePipelineRejectsUnknownOnlyLane(t *testing.T) {
_, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{"missing"}}, newProfileCatalog(t))
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
assertErrorContains(t, err, "pipeline", "missing", "not declared")
}
func TestResolvePipelineRejectsEmptyOnlyLane(t *testing.T) {
_, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{Only: []string{" \t"}}, newProfileCatalog(t))
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
assertErrorContains(t, err, "pipeline", "artifact lane", "empty")
}
func TestResolvePipelineRejectsEmptyArtifactSet(t *testing.T) {
_, err := ResolvePipeline(PipelineProfile{
ID: "empty",
Input: Binding("text"),
Artifacts: map[string]ArtifactLaneProfile{},
}, ResolveOptions{}, newProfileCatalog(t))
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
assertErrorContains(t, err, "empty", "artifact lane")
}
func TestResolvePipelineRejectsEmptyPipelineID(t *testing.T) {
_, err := ResolvePipeline(PipelineProfile{
ID: " ",
Input: Binding("text"),
Artifacts: map[string]ArtifactLaneProfile{
"events": {Extract: Binding("event-extractor")},
},
}, ResolveOptions{}, newProfileCatalog(t))
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
assertErrorContains(t, err, "pipeline id", "empty")
}
func TestResolvePipelineRejectsMissingInput(t *testing.T) {
_, err := ResolvePipeline(PipelineProfile{
ID: "missing-input",
Artifacts: map[string]ArtifactLaneProfile{
"events": {Extract: Binding("event-extractor")},
},
}, ResolveOptions{}, newProfileCatalog(t))
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
assertErrorContains(t, err, "missing-input", "input", "empty")
}
func TestResolvePipelineRejectsUnknownModuleKeys(t *testing.T) {
tests := []struct {
name string
profile PipelineProfile
want []string
}{
{
name: "input",
profile: PipelineProfile{
ID: "unknown-input",
Input: Binding("missing-input"),
Artifacts: map[string]ArtifactLaneProfile{
"events": {Extract: Binding("event-extractor")},
},
},
want: []string{"unknown-input", "input", "missing-input"},
},
{
name: "chunk",
profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
profile.Chunk = Binding("missing-chunk")
return profile
}),
want: []string{"baseline", "chunk", "missing-chunk"},
},
{
name: "extract",
profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
lane := profile.Artifacts["events"]
lane.Extract = Binding("missing-extractor")
profile.Artifacts["events"] = lane
return profile
}),
want: []string{"baseline", "events", "extract", "missing-extractor"},
},
{
name: "merge",
profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
lane := profile.Artifacts["events"]
lane.Merge = Binding("missing-merge")
profile.Artifacts["events"] = lane
return profile
}),
want: []string{"baseline", "events", "merge", "missing-merge"},
},
{
name: "normalize",
profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
lane := profile.Artifacts["events"]
lane.Normalize = Binding("missing-normalize")
profile.Artifacts["events"] = lane
return profile
}),
want: []string{"baseline", "events", "normalize", "missing-normalize"},
},
{
name: "validate",
profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
lane := profile.Artifacts["events"]
lane.Validators = []ModuleBinding{Binding("missing-validator")}
profile.Artifacts["events"] = lane
return profile
}),
want: []string{"baseline", "events", "validate", "missing-validator"},
},
{
name: "output",
profile: withProfileChange(func(profile PipelineProfile) PipelineProfile {
profile.Output = Binding("missing-output")
return profile
}),
want: []string{"baseline", "output", "missing-output"},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
_, err := ResolvePipeline(test.profile, ResolveOptions{}, newProfileCatalog(t))
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
assertErrorContains(t, err, test.want...)
})
}
}
func TestResolvePipelineRejectsMissingCapabilities(t *testing.T) {
tests := []struct {
name string
spec ModuleSpec
want []string
}{
{
name: "input",
spec: ModuleSpec{Key: "text", Stage: StageInput, Requires: []string{"raw"}},
want: []string{"baseline", "input", "text", "raw"},
},
{
name: "chunk",
spec: ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"missing"}},
want: []string{"baseline", "chunk", "generic", "missing"},
},
{
name: "extract",
spec: ModuleSpec{Key: "event-extractor", Stage: StageExtract, Requires: []string{"missing"}},
want: []string{"baseline", "events", "extract", "event-extractor", "missing"},
},
{
name: "merge",
spec: ModuleSpec{Key: "appendorder", Stage: StageMerge, Requires: []string{"missing"}},
want: []string{"baseline", "events", "merge", "appendorder", "missing"},
},
{
name: "normalize",
spec: ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"missing"}},
want: []string{"baseline", "events", "normalize", "noop", "missing"},
},
{
name: "validate",
spec: ModuleSpec{Key: "grounded", Stage: StageValidate, Requires: []string{"missing"}},
want: []string{"baseline", "events", "validate", "grounded", "missing"},
},
{
name: "output",
spec: ModuleSpec{Key: "json", Stage: StageOutput, Requires: []string{"missing"}},
want: []string{"baseline", "output", "json", "missing"},
},
}
for _, test := range tests {
t.Run(test.name, func(t *testing.T) {
catalog := newProfileCatalogWithOverride(t, test.spec)
profile := baselineProfile()
lane := profile.Artifacts["events"]
lane.Validators = []ModuleBinding{Binding("grounded")}
profile.Artifacts["events"] = lane
_, err := ResolvePipeline(profile, ResolveOptions{}, catalog)
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
assertErrorContains(t, err, test.want...)
})
}
}
func TestResolvePipelineOrdersLanesDeterministically(t *testing.T) {
resolved, err := ResolvePipeline(multiLaneProfile(), ResolveOptions{}, newProfileCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
got := laneIDs(resolved.ArtifactLanes)
want := []string{"events", "notes", "summaries"}
if !reflect.DeepEqual(got, want) {
t.Fatalf("lane IDs = %#v, want %#v", got, want)
}
}
func TestResolvePipelineDigestIsDeterministicForEquivalentMaps(t *testing.T) {
left := PipelineProfile{
ID: "digest",
Input: Binding("text"),
Output: Binding("json"),
Chunk: ModuleBinding{Module: "generic", Options: map[string]any{"b": 2, "a": 1}},
Artifacts: map[string]ArtifactLaneProfile{
"events": {Extract: Binding("event-extractor")},
"notes": {Extract: Binding("note-extractor")},
},
}
right := PipelineProfile{
ID: "digest",
Input: Binding("text"),
Output: Binding("json"),
Chunk: ModuleBinding{Module: "generic", Options: map[string]any{"a": 1, "b": 2}},
Artifacts: map[string]ArtifactLaneProfile{
"notes": {Extract: Binding("note-extractor")},
"events": {Extract: Binding("event-extractor")},
},
}
leftResolved, err := ResolvePipeline(left, ResolveOptions{}, newProfileCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline(left) error = %v, want nil", err)
}
rightResolved, err := ResolvePipeline(right, ResolveOptions{}, newProfileCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline(right) error = %v, want nil", err)
}
if leftResolved.Digest != rightResolved.Digest {
t.Fatalf("digests differ for equivalent profiles: %q != %q", leftResolved.Digest, rightResolved.Digest)
}
}
func TestResolvePipelineDigestChangesWhenBindingChanges(t *testing.T) {
left := baselineProfile()
right := baselineProfile()
right.Chunk = Binding("window")
catalog := newProfileCatalog(t)
registerProfileSpecs(t, catalog, ModuleSpec{Key: "window", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}})
leftResolved, err := ResolvePipeline(left, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline(left) error = %v, want nil", err)
}
rightResolved, err := ResolvePipeline(right, ResolveOptions{}, catalog)
if err != nil {
t.Fatalf("ResolvePipeline(right) error = %v, want nil", err)
}
if leftResolved.Digest == rightResolved.Digest {
t.Fatalf("digest = %q for both profiles, want changed digest", leftResolved.Digest)
}
}
func TestBindingTrimsModuleAndLeavesResolutionFieldsEmpty(t *testing.T) {
binding := Binding(" module ")
if binding.Module != "module" {
t.Fatalf("Module = %q, want module", binding.Module)
}
if binding.LLMProfile != "" {
t.Fatalf("LLMProfile = %q, want empty", binding.LLMProfile)
}
if binding.Options != nil {
t.Fatalf("Options = %#v, want nil", binding.Options)
}
}
func TestResolvedPipelineDigestExcludesDigestField(t *testing.T) {
resolved, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, newProfileCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
changed := resolved
changed.Digest = "sha256:changed"
leftDigest, err := resolvedPipelineDigest(resolved)
if err != nil {
t.Fatalf("resolvedPipelineDigest(resolved) error = %v, want nil", err)
}
rightDigest, err := resolvedPipelineDigest(changed)
if err != nil {
t.Fatalf("resolvedPipelineDigest(changed) error = %v, want nil", err)
}
if leftDigest != rightDigest {
t.Fatalf("digest with changed digest field = %q, want %q", rightDigest, leftDigest)
}
}
func baselineProfile() PipelineProfile {
return PipelineProfile{
ID: "baseline",
Input: Binding("text"),
Artifacts: map[string]ArtifactLaneProfile{
"events": {Extract: Binding("event-extractor")},
},
}
}
func multiLaneProfile() PipelineProfile {
profile := baselineProfile()
profile.ID = "multi"
profile.Artifacts = map[string]ArtifactLaneProfile{
"summaries": {Extract: Binding("note-extractor")},
"events": {Extract: Binding("event-extractor")},
"notes": {Extract: Binding("note-extractor")},
}
return profile
}
func withProfileChange(change func(PipelineProfile) PipelineProfile) PipelineProfile {
return change(baselineProfile())
}
func laneIDs(lanes []ResolvedArtifactLane) []string {
ids := make([]string, 0, len(lanes))
for _, lane := range lanes {
ids = append(ids, lane.ID)
}
return ids
}
func assertErrorContains(t *testing.T, err error, values ...string) {
t.Helper()
message := err.Error()
for _, value := range values {
if !strings.Contains(message, value) {
t.Fatalf("error = %q, want substring %q", message, value)
}
}
}
func newProfileCatalog(t *testing.T) ModuleCatalog {
t.Helper()
catalog := emptyProfileCatalog()
registerProfileSpecs(t, catalog, defaultProfileSpecs()...)
return catalog
}
func newProfileCatalogWithOverride(t *testing.T, override ModuleSpec) ModuleCatalog {
t.Helper()
specs := defaultProfileSpecs()
for index, spec := range specs {
if spec.Stage == override.Stage && spec.Key == override.Key {
specs[index] = override
catalog := emptyProfileCatalog()
registerProfileSpecs(t, catalog, specs...)
return catalog
}
}
catalog := emptyProfileCatalog()
registerProfileSpecs(t, catalog, specs...)
registerProfileSpecs(t, catalog, override)
return catalog
}
func emptyProfileCatalog() ModuleCatalog {
return ModuleCatalog{
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),
Validators: NewValidatorRegistry(),
Outputs: NewOutputEncoderRegistry(),
}
}
func defaultProfileSpecs() []ModuleSpec {
return []ModuleSpec{
ModuleSpec{Key: "text", Stage: StageInput, Provides: []string{"source"}},
ModuleSpec{Key: "generic", Stage: StageChunk, Requires: []string{"source"}, Provides: []string{"chunk"}},
ModuleSpec{Key: "event-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}},
ModuleSpec{Key: "note-extractor", Stage: StageExtract, Requires: []string{"chunk"}, Provides: []string{"candidate"}},
ModuleSpec{Key: "appendorder", Stage: StageMerge, Requires: []string{"candidate"}, Provides: []string{"merged"}},
ModuleSpec{Key: "noop", Stage: StageNormalize, Requires: []string{"merged"}, Provides: []string{"normalized"}},
ModuleSpec{Key: "grounded", Stage: StageValidate, Requires: []string{"normalized"}, Provides: []string{"validated"}},
ModuleSpec{Key: "json", Stage: StageOutput, Requires: []string{"normalized"}, Provides: []string{"encoded"}},
}
}
func registerProfileSpecs(t *testing.T, catalog ModuleCatalog, specs ...ModuleSpec) {
t.Helper()
for _, spec := range specs {
switch spec.Stage {
case StageInput:
if err := catalog.Inputs.RegisterWithSpec(spec, profileInputConstructor(spec.Key)); err != nil {
t.Fatalf("register input spec %#v: %v", spec, err)
}
case StageChunk:
if err := catalog.Chunkers.RegisterWithSpec(spec, profileChunkerConstructor(spec.Key)); err != nil {
t.Fatalf("register chunk spec %#v: %v", spec, err)
}
case StageExtract:
if err := catalog.Extractors.RegisterWithSpec(spec, profileExtractorConstructor(spec.Key)); err != nil {
t.Fatalf("register extractor spec %#v: %v", spec, err)
}
case StageMerge:
if err := catalog.Mergers.RegisterWithSpec(spec, profileMergerConstructor(spec.Key)); err != nil {
t.Fatalf("register merger spec %#v: %v", spec, err)
}
case StageNormalize:
if err := catalog.Normalizers.RegisterWithSpec(spec, profileNormalizerConstructor(spec.Key)); err != nil {
t.Fatalf("register normalizer spec %#v: %v", spec, err)
}
case StageValidate:
if err := catalog.Validators.RegisterWithSpec(spec, profileValidatorConstructor(spec.Key)); err != nil {
t.Fatalf("register validator spec %#v: %v", spec, err)
}
case StageOutput:
if err := catalog.Outputs.RegisterWithSpec(spec, profileOutputConstructor(spec.Key)); err != nil {
t.Fatalf("register output spec %#v: %v", spec, err)
}
default:
t.Fatalf("unsupported spec stage %q", spec.Stage)
}
}
}
func profileInputConstructor(key string) InputAdapterConstructor {
return func() (contracts.InputAdapter, error) {
return profileInputAdapter{key: key}, nil
}
}
type profileInputAdapter struct {
key string
}
func (adapter profileInputAdapter) Key() string {
return adapter.key
}
func (adapter profileInputAdapter) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
return &source.SourceDocument{}, nil
}
func profileChunkerConstructor(key string) ChunkerConstructor {
return func() (contracts.Chunker, error) {
return registryChunker{key: key}, nil
}
}
func profileExtractorConstructor(key string) ExtractorConstructor {
return func() (contracts.Extractor, error) {
return registryFakeExtractor{key: key}, nil
}
}
func profileMergerConstructor(key string) MergerConstructor {
return func() (contracts.Merger, error) {
return registryMerger{key: key}, nil
}
}
func profileNormalizerConstructor(key string) NormalizerConstructor {
return func() (contracts.Normalizer, error) {
return registryNormalizer{key: key}, nil
}
}
func profileValidatorConstructor(key string) ValidatorConstructor {
return func() (contracts.Validator, error) {
return registryValidator{name: key}, nil
}
}
func profileOutputConstructor(key string) OutputEncoderConstructor {
return func() (contracts.OutputEncoder, error) {
return registryOutputEncoder{key: key}, nil
}
}
func TestResolvedPipelineCanMarshalToCanonicalJSON(t *testing.T) {
resolved, err := ResolvePipeline(baselineProfile(), ResolveOptions{}, newProfileCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
if _, err := json.Marshal(resolved); err != nil {
t.Fatalf("json.Marshal(resolved) error = %v, want nil", err)
}
}

View File

@@ -12,54 +12,128 @@ import (
validate "gitea.maximumdirect.net/eric/notarius/internal/framework/validate"
)
func TestRunnerUsesExtractorRegistry(t *testing.T) {
var builtKeys []string
var executedKeys []string
registry := NewExtractorRegistry()
func TestRunnerUsesRegistries(t *testing.T) {
var built []string
var executed []string
registries := integrationRegistries(t, &built, &executed)
registerIntegrationExtractor(t, registry, "second", &builtKeys, &executedKeys, []contracts.Validator{
integrationValidator{name: "reject-second", approve: false},
})
registerIntegrationExtractor(t, registry, "first", &builtKeys, &executedKeys, []contracts.Validator{
integrationValidator{name: "approve-first", approve: true},
})
output, err := New(registry).Run(context.Background(), RunInput{
Source: integrationSourceDocument(),
ExtractorKeys: []string{"second", "first"},
output, err := New(registries).Run(context.Background(), RunInput{
Pipeline: integrationPipeline(),
SourceID: "source-1",
RawInput: []byte("source text"),
})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
if !reflect.DeepEqual(builtKeys, []string{"second", "first"}) {
t.Fatalf("built keys = %#v, want configured order", builtKeys)
wantBuilt := []string{"input", "chunk", "extract-first", "merge", "normalize", "extract-second", "merge", "normalize", "output"}
if !reflect.DeepEqual(built, wantBuilt) {
t.Fatalf("built = %#v, want %#v", built, wantBuilt)
}
if !reflect.DeepEqual(executedKeys, []string{"second", "first"}) {
t.Fatalf("executed keys = %#v, want configured order", executedKeys)
if !reflect.DeepEqual(executed, []string{"extract-first:chunk-0", "extract-second:chunk-0"}) {
t.Fatalf("executed = %#v, want extractor chunk execution", executed)
}
if got := artifactKeys(output.Approved); !reflect.DeepEqual(got, []string{"first"}) {
t.Fatalf("approved keys = %#v, want [first]", got)
if got := artifactKeys(output.Approved); !reflect.DeepEqual(got, []string{"extract-first"}) {
t.Fatalf("approved keys = %#v, want [extract-first]", got)
}
if got := rejectedKeys(output.Rejected); !reflect.DeepEqual(got, []string{"second"}) {
t.Fatalf("rejected keys = %#v, want [second]", got)
if got := rejectedKeys(output.Rejected); !reflect.DeepEqual(got, []string{"extract-second"}) {
t.Fatalf("rejected keys = %#v, want [extract-second]", got)
}
}
func registerIntegrationExtractor(t *testing.T, registry *ExtractorRegistry, key string, builtKeys *[]string, executedKeys *[]string, validators []contracts.Validator) {
func integrationRegistries(t *testing.T, built, executed *[]string) Registries {
t.Helper()
registries := Registries{
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),
Outputs: NewOutputEncoderRegistry(),
}
if err := registries.Inputs.Register("input", func() (contracts.InputAdapter, error) {
*built = append(*built, "input")
return integrationInput{}, nil
}); err != nil {
t.Fatalf("register input: %v", err)
}
if err := registries.Chunkers.Register("chunk", func() (contracts.Chunker, error) {
*built = append(*built, "chunk")
return integrationChunker{}, nil
}); err != nil {
t.Fatalf("register chunker: %v", err)
}
registerIntegrationExtractor(t, registries.Extractors, "extract-first", built, executed, []contracts.Validator{
integrationValidator{name: "approve-first", approve: true},
})
registerIntegrationExtractor(t, registries.Extractors, "extract-second", built, executed, []contracts.Validator{
integrationValidator{name: "reject-second", approve: false},
})
if err := registries.Mergers.Register("merge", func() (contracts.Merger, error) {
*built = append(*built, "merge")
return integrationMerger{}, nil
}); err != nil {
t.Fatalf("register merger: %v", err)
}
if err := registries.Normalizers.Register("normalize", func() (contracts.Normalizer, error) {
*built = append(*built, "normalize")
return integrationNormalizer{}, nil
}); err != nil {
t.Fatalf("register normalizer: %v", err)
}
if err := registries.Outputs.Register("output", func() (contracts.OutputEncoder, error) {
*built = append(*built, "output")
return integrationOutput{}, nil
}); err != nil {
t.Fatalf("register output: %v", err)
}
return registries
}
func registerIntegrationExtractor(t *testing.T, registry *ExtractorRegistry, key string, built, executed *[]string, validators []contracts.Validator) {
t.Helper()
if err := registry.Register(key, func() (contracts.Extractor, error) {
*builtKeys = append(*builtKeys, key)
return integrationExtractor{key: key, executedKeys: executedKeys, validators: validators}, nil
*built = append(*built, key)
return integrationExtractor{key: key, executed: executed, validators: validators}, nil
}); err != nil {
t.Fatalf("Register(%q) error = %v, want nil", key, err)
}
}
type integrationInput struct{}
func (input integrationInput) Key() string {
return "input"
}
func (input integrationInput) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
return integrationSourceDocument(), nil
}
type integrationChunker struct{}
func (chunker integrationChunker) Key() string {
return "chunk"
}
func (chunker integrationChunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) {
return contracts.ChunkResult{
Chunks: []contracts.SourceChunk{
{
ID: "chunk-0",
SourceID: req.Source.ID,
Index: 0,
Units: req.Source.Units,
},
},
}, nil
}
type integrationExtractor struct {
key string
executedKeys *[]string
executed *[]string
validators []contracts.Validator
}
@@ -80,7 +154,7 @@ func (extractor integrationExtractor) Validators() []contracts.Validator {
}
func (extractor integrationExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) {
*extractor.executedKeys = append(*extractor.executedKeys, extractor.key)
*extractor.executed = append(*extractor.executed, extractor.key+":"+req.Chunk.ID)
return contracts.ExtractionResult{
Candidates: []artifacts.ArtifactCandidate{
{Payload: []byte(`{"value":true}`)},
@@ -88,6 +162,40 @@ func (extractor integrationExtractor) Extract(ctx context.Context, req contracts
}, nil
}
type integrationNormalizer struct{}
type integrationMerger struct{}
func (merger integrationMerger) Key() string {
return "merge"
}
func (merger integrationMerger) 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
}
func (normalizer integrationNormalizer) Key() string {
return "normalize"
}
func (normalizer integrationNormalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
return contracts.NormalizeResult{Candidates: req.Candidates}, nil
}
type integrationOutput struct{}
func (output integrationOutput) Key() string {
return "output"
}
func (output integrationOutput) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
return contracts.OutputResult{Bytes: []byte(`{}`), ContentType: "application/json"}, nil
}
type integrationValidator struct {
name string
approve bool
@@ -112,6 +220,30 @@ func (validator integrationValidator) Validate(ctx context.Context, req contract
}, nil
}
func integrationPipeline() ResolvedPipeline {
return ResolvedPipeline{
ID: "pipeline-1",
Digest: "sha256:pipeline",
Input: Binding("input"),
Chunk: Binding("chunk"),
ArtifactLanes: []ResolvedArtifactLane{
{
ID: "first",
Extract: Binding("extract-first"),
Merge: Binding("merge"),
Normalize: Binding("normalize"),
},
{
ID: "second",
Extract: Binding("extract-second"),
Merge: Binding("merge"),
Normalize: Binding("normalize"),
},
},
Output: Binding("output"),
}
}
func integrationSourceDocument() *source.SourceDocument {
return &source.SourceDocument{
ID: "source-1",

View File

@@ -10,29 +10,40 @@ import (
"gitea.maximumdirect.net/eric/notarius/internal/framework/validate"
)
type ExtractorFactory interface {
Build(key string) (contracts.Extractor, error)
type Registries struct {
Inputs *InputAdapterRegistry
Chunkers *ChunkerRegistry
Extractors *ExtractorRegistry
Mergers *MergerRegistry
Normalizers *NormalizerRegistry
Validators *ValidatorRegistry
Outputs *OutputEncoderRegistry
}
type Runner struct {
extractors ExtractorFactory
registries Registries
}
func New(extractors ExtractorFactory) *Runner {
return &Runner{extractors: extractors}
func New(registries Registries) *Runner {
return &Runner{registries: registries}
}
type RunInput struct {
Source *source.SourceDocument
ExtractorKeys []string
Pipeline ResolvedPipeline
SourceID string
Path string
RawInput []byte
LLMClient contracts.StructuredLLMClient
Metadata map[string]any
}
type RunOutput struct {
Manifest artifacts.RunManifest `json:"manifest"`
Approved []artifacts.Artifact `json:"approved,omitempty"`
Rejected []artifacts.RejectedArtifact `json:"rejected,omitempty"`
Warnings []contracts.Warning `json:"warnings,omitempty"`
EncodedOutput []byte `json:"-"`
ContentType string `json:"content_type,omitempty"`
}
func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
@@ -40,54 +51,283 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (RunOutput, error) {
if r == nil {
return output, fmt.Errorf("runner must not be nil")
}
if r.extractors == nil {
return output, fmt.Errorf("runner extractor factory must not be nil")
if err := validateRunInput(input); err != nil {
return output, err
}
if err := source.ValidateDocument(input.Source); err != nil {
return output, fmt.Errorf("validate source document: %w", err)
if err := r.validateRegistries(input.Pipeline); err != nil {
return output, err
}
if len(input.ExtractorKeys) == 0 {
return output, fmt.Errorf("extractor keys must not be empty")
output.Manifest = manifestFromPipeline(input.Pipeline)
adapter, err := r.registries.Inputs.Build(input.Pipeline.Input.Module)
if err != nil {
return failOutput(output), fmt.Errorf("build input adapter %q: %w", input.Pipeline.Input.Module, err)
}
doc, err := adapter.Parse(ctx, contracts.ParseRequest{
SourceID: input.SourceID,
Path: input.Path,
Raw: input.RawInput,
Metadata: input.Metadata,
})
if err != nil {
return failOutput(output), fmt.Errorf("parse input with adapter %q: %w", adapter.Key(), err)
}
if err := source.ValidateDocument(doc); err != nil {
return failOutput(output), fmt.Errorf("validate source document: %w", err)
}
output.Manifest.SourceDigests = []string{doc.Digest}
chunker, err := r.registries.Chunkers.Build(input.Pipeline.Chunk.Module)
if err != nil {
return failOutput(output), fmt.Errorf("build chunker %q: %w", input.Pipeline.Chunk.Module, err)
}
chunkResult, err := chunker.Chunk(ctx, contracts.ChunkRequest{
Source: doc,
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, chunkResult.Warnings...)
if err != nil {
return failOutput(output), fmt.Errorf("chunk source with chunker %q: %w", chunker.Key(), err)
}
if len(chunkResult.Chunks) == 0 {
return failOutput(output), fmt.Errorf("chunker %q returned no chunks", chunker.Key())
}
nextCandidateIndex := 0
for _, extractorKey := range input.ExtractorKeys {
extractor, err := r.extractors.Build(extractorKey)
if err != nil {
return output, fmt.Errorf("build extractor %q: %w", extractorKey, err)
for _, lane := range input.Pipeline.ArtifactLanes {
if err := r.runLane(ctx, input, doc, chunkResult.Chunks, lane, &output, &nextCandidateIndex); err != nil {
return failOutput(output), err
}
if extractor == nil {
return output, fmt.Errorf("build extractor %q: returned nil extractor", extractorKey)
}
if len(output.Rejected) > 0 {
output.Manifest.ValidationStatus = "rejected"
} else {
output.Manifest.ValidationStatus = "approved"
}
encoder, err := r.registries.Outputs.Build(input.Pipeline.Output.Module)
if err != nil {
return failOutput(output), fmt.Errorf("build output encoder %q: %w", input.Pipeline.Output.Module, err)
}
encoded, err := encoder.Encode(ctx, contracts.OutputRequest{
Manifest: output.Manifest,
Approved: output.Approved,
Rejected: output.Rejected,
Warnings: output.Warnings,
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, encoded.Warnings...)
if err != nil {
return failOutput(output), fmt.Errorf("encode output with encoder %q: %w", encoder.Key(), err)
}
output.EncodedOutput = encoded.Bytes
output.ContentType = encoded.ContentType
return output, nil
}
func (r *Runner) runLane(ctx context.Context, input RunInput, doc *source.SourceDocument, chunks []contracts.SourceChunk, lane ResolvedArtifactLane, output *RunOutput, nextCandidateIndex *int) error {
extractor, err := r.registries.Extractors.Build(lane.Extract.Module)
if err != nil {
return fmt.Errorf("build extractor %q for lane %q: %w", lane.Extract.Module, lane.ID, err)
}
merger, err := r.registries.Mergers.Build(lane.Merge.Module)
if err != nil {
return fmt.Errorf("build merger %q for lane %q: %w", lane.Merge.Module, lane.ID, err)
}
normalizer, err := r.registries.Normalizers.Build(lane.Normalize.Module)
if err != nil {
return fmt.Errorf("build normalizer %q for lane %q: %w", lane.Normalize.Module, lane.ID, err)
}
var validators []contracts.Validator
if len(lane.Validators) > 0 {
validators, err = r.buildConfiguredValidators(lane)
if err != nil {
return err
}
} else {
validators = extractor.Validators()
}
chunkArtifacts := make([]contracts.ChunkArtifacts, 0, len(chunks))
for index := range chunks {
chunk := chunks[index]
result, err := extractor.Extract(ctx, contracts.ExtractionRequest{
Source: input.Source,
Source: doc,
Chunk: &chunk,
LLMClient: input.LLMClient,
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, result.Warnings...)
if err != nil {
return output, fmt.Errorf("extract with extractor %q: %w", extractor.Key(), err)
return fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, extractor.Key(), err)
}
candidates, err := normalizeCandidates(extractor, result.Candidates, &nextCandidateIndex)
candidates, err := normalizeCandidates(extractor, result.Candidates, nextCandidateIndex)
if err != nil {
return output, err
return err
}
chunkArtifacts = append(chunkArtifacts, contracts.ChunkArtifacts{
Chunk: chunk,
Candidates: candidates,
})
}
approved, rejected, warnings, err := runValidators(ctx, extractor, input.Source, candidates, input.Metadata)
mergeResult, err := merger.Merge(ctx, contracts.MergeRequest{
Source: doc,
LaneID: lane.ID,
ChunkArtifacts: chunkArtifacts,
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, mergeResult.Warnings...)
if err != nil {
return fmt.Errorf("merge lane %q with merger %q: %w", lane.ID, merger.Key(), err)
}
normalizeResult, err := normalizer.Normalize(ctx, contracts.NormalizeRequest{
Source: doc,
LaneID: lane.ID,
Candidates: mergeResult.Candidates,
Metadata: input.Metadata,
})
output.Warnings = append(output.Warnings, normalizeResult.Warnings...)
if err != nil {
return fmt.Errorf("normalize lane %q with normalizer %q: %w", lane.ID, normalizer.Key(), err)
}
approved, rejected, warnings, err := runValidators(ctx, extractor.Key(), validators, doc, normalizeResult.Candidates, input.Metadata)
output.Warnings = append(output.Warnings, warnings...)
output.Rejected = append(output.Rejected, rejected...)
if err != nil {
return output, err
return err
}
for _, candidate := range approved {
output.Approved = append(output.Approved, artifacts.ArtifactFromCandidate(candidate))
}
return nil
}
func (r *Runner) buildConfiguredValidators(lane ResolvedArtifactLane) ([]contracts.Validator, error) {
validators := make([]contracts.Validator, 0, len(lane.Validators))
for _, binding := range lane.Validators {
validator, err := r.registries.Validators.Build(binding.Module)
if err != nil {
return nil, fmt.Errorf("build validator %q for lane %q: %w", binding.Module, lane.ID, err)
}
validators = append(validators, validator)
}
return validators, nil
}
func (r *Runner) validateRegistries(pipeline ResolvedPipeline) error {
if r.registries.Inputs == nil {
return fmt.Errorf("input registry must not be nil")
}
if r.registries.Chunkers == nil {
return fmt.Errorf("chunker registry must not be nil")
}
if r.registries.Extractors == nil {
return fmt.Errorf("extractor registry must not be nil")
}
if r.registries.Mergers == nil {
return fmt.Errorf("merger registry must not be nil")
}
if r.registries.Normalizers == nil {
return fmt.Errorf("normalizer registry must not be nil")
}
if r.registries.Outputs == nil {
return fmt.Errorf("output encoder registry must not be nil")
}
if pipelineUsesConfiguredValidators(pipeline) && r.registries.Validators == nil {
return fmt.Errorf("validator registry must not be nil")
}
return nil
}
func validateRunInput(input RunInput) error {
if input.Pipeline.ID == "" {
return fmt.Errorf("resolved pipeline id must not be empty")
}
if input.Pipeline.Digest == "" {
return fmt.Errorf("resolved pipeline digest must not be empty")
}
if input.Pipeline.Input.Module == "" {
return fmt.Errorf("resolved pipeline input module must not be empty")
}
if input.Pipeline.Chunk.Module == "" {
return fmt.Errorf("resolved pipeline chunk module must not be empty")
}
if input.Pipeline.Output.Module == "" {
return fmt.Errorf("resolved pipeline output module must not be empty")
}
if len(input.Pipeline.ArtifactLanes) == 0 {
return fmt.Errorf("resolved pipeline artifact lanes must not be empty")
}
for _, lane := range input.Pipeline.ArtifactLanes {
if lane.ID == "" {
return fmt.Errorf("resolved pipeline artifact lane id must not be empty")
}
if lane.Extract.Module == "" {
return fmt.Errorf("resolved pipeline lane %q extract module must not be empty", lane.ID)
}
if lane.Merge.Module == "" {
return fmt.Errorf("resolved pipeline lane %q merge module must not be empty", lane.ID)
}
if lane.Normalize.Module == "" {
return fmt.Errorf("resolved pipeline lane %q normalize module must not be empty", lane.ID)
}
for _, validator := range lane.Validators {
if validator.Module == "" {
return fmt.Errorf("resolved pipeline lane %q validator module must not be empty", lane.ID)
}
}
}
return nil
}
func manifestFromPipeline(pipeline ResolvedPipeline) artifacts.RunManifest {
manifest := artifacts.RunManifest{
PipelineID: pipeline.ID,
PipelineDigest: pipeline.Digest,
InputModule: pipeline.Input.Module,
Chunker: pipeline.Chunk.Module,
OutputEncoder: pipeline.Output.Module,
ArtifactLanes: make([]artifacts.ArtifactLaneManifest, 0, len(pipeline.ArtifactLanes)),
}
return output, nil
for _, lane := range pipeline.ArtifactLanes {
laneManifest := artifacts.ArtifactLaneManifest{
ID: lane.ID,
Extractor: lane.Extract.Module,
Merger: lane.Merge.Module,
Normalizer: lane.Normalize.Module,
}
for _, validator := range lane.Validators {
laneManifest.Validators = append(laneManifest.Validators, validator.Module)
}
manifest.ArtifactLanes = append(manifest.ArtifactLanes, laneManifest)
}
return manifest
}
func failOutput(output RunOutput) RunOutput {
if output.Manifest.PipelineID != "" {
output.Manifest.ValidationStatus = "failed"
}
return output
}
func pipelineUsesConfiguredValidators(pipeline ResolvedPipeline) bool {
for _, lane := range pipeline.ArtifactLanes {
if len(lane.Validators) > 0 {
return true
}
}
return false
}
func normalizeCandidates(extractor contracts.Extractor, candidates []artifacts.ArtifactCandidate, nextIndex *int) ([]artifacts.ArtifactCandidate, error) {
@@ -119,14 +359,14 @@ func normalizeCandidates(extractor contracts.Extractor, candidates []artifacts.A
return normalized, nil
}
func runValidators(ctx context.Context, extractor contracts.Extractor, doc *source.SourceDocument, candidates []artifacts.ArtifactCandidate, metadata map[string]any) ([]artifacts.ArtifactCandidate, []artifacts.RejectedArtifact, []contracts.Warning, error) {
func runValidators(ctx context.Context, extractorKey string, validators []contracts.Validator, doc *source.SourceDocument, candidates []artifacts.ArtifactCandidate, metadata map[string]any) ([]artifacts.ArtifactCandidate, []artifacts.RejectedArtifact, []contracts.Warning, error) {
eligible := candidates
var rejected []artifacts.RejectedArtifact
var warnings []contracts.Warning
for validatorIndex, validator := range extractor.Validators() {
for validatorIndex, validator := range validators {
if validator == nil {
return nil, rejected, warnings, fmt.Errorf("extractor %q validator[%d] must not be nil", extractor.Key(), validatorIndex)
return nil, rejected, warnings, fmt.Errorf("extractor %q validator[%d] must not be nil", extractorKey, validatorIndex)
}
result, err := validator.Validate(ctx, contracts.ValidationRequest{
Source: doc,
@@ -135,13 +375,13 @@ func runValidators(ctx context.Context, extractor contracts.Extractor, doc *sour
})
warnings = append(warnings, result.Warnings...)
if err != nil {
return nil, rejected, warnings, fmt.Errorf("validate extractor %q with validator %q: %w", extractor.Key(), validator.Name(), err)
return nil, rejected, warnings, fmt.Errorf("validate extractor %q with validator %q: %w", extractorKey, validator.Name(), err)
}
if result.ValidatorName != validator.Name() {
return nil, rejected, warnings, fmt.Errorf("validator %q returned result for %q", validator.Name(), result.ValidatorName)
}
if err := validate.EnforceDecisionCardinality(eligible, result.Decisions); err != nil {
return nil, rejected, warnings, fmt.Errorf("validate extractor %q with validator %q: %w", extractor.Key(), validator.Name(), err)
return nil, rejected, warnings, fmt.Errorf("validate extractor %q with validator %q: %w", extractorKey, validator.Name(), err)
}
decisions := make(map[int]contracts.ValidationDecision, len(result.Decisions))

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,17 @@
{
"id": "fixture-source",
"units": [
{
"id": "u1",
"text": "First event."
},
{
"id": "u2",
"text": "Second event."
},
{
"id": "u3",
"text": "Third event."
}
]
}

View File

@@ -0,0 +1,51 @@
{
"manifest": {
"pipeline_id": "walking-skeleton",
"pipeline_digest": "sha256:5df1e501a2307ef75bbfeb59d315b3710571d52e5466a9c7f8320248740e6fca",
"validation_status": "approved",
"artifact_lanes": [
{
"id": "events",
"extractor": "fake/extract",
"merger": "appendorder",
"normalizer": "noop"
}
]
},
"approved": [
{
"extractor_key": "fake/extract",
"artifact_type": "fake_event",
"schema_version": "v1",
"payload": {
"chunk_id": "fixture-source:chunk:0",
"llm_call": 1,
"text": "First event. Second event."
},
"source_refs": [
{
"source_id": "fixture-source",
"start_unit_id": "u1",
"end_unit_id": "u2"
}
]
},
{
"extractor_key": "fake/extract",
"artifact_type": "fake_event",
"schema_version": "v1",
"payload": {
"chunk_id": "fixture-source:chunk:1",
"llm_call": 2,
"text": "Third event."
},
"source_refs": [
{
"source_id": "fixture-source",
"start_unit_id": "u3",
"end_unit_id": "u3"
}
]
}
]
}

View File

@@ -0,0 +1,102 @@
package pipeline
import (
"fmt"
"strings"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type ValidatorConstructor func() (contracts.Validator, error)
type ValidatorRegistry struct {
constructors map[string]ValidatorConstructor
specs map[string]ModuleSpec
}
func NewValidatorRegistry() *ValidatorRegistry {
return &ValidatorRegistry{
constructors: make(map[string]ValidatorConstructor),
specs: make(map[string]ModuleSpec),
}
}
func (r *ValidatorRegistry) Register(key string, constructor ValidatorConstructor) error {
return r.RegisterWithSpec(defaultModuleSpec(key, StageValidate), constructor)
}
func (r *ValidatorRegistry) RegisterWithSpec(spec ModuleSpec, constructor ValidatorConstructor) error {
if r == nil {
return fmt.Errorf("validator registry must not be nil")
}
normalizedSpec := normalizeModuleSpec(spec)
if err := validateModuleSpec("validator", StageValidate, normalizedSpec); err != nil {
return err
}
if constructor == nil {
return fmt.Errorf("validator constructor for %q must not be nil", normalizedSpec.Key)
}
if _, ok := r.constructors[normalizedSpec.Key]; ok {
return fmt.Errorf("validator %q is already registered", normalizedSpec.Key)
}
if r.constructors == nil {
r.constructors = make(map[string]ValidatorConstructor)
}
if r.specs == nil {
r.specs = make(map[string]ModuleSpec)
}
r.constructors[normalizedSpec.Key] = constructor
r.specs[normalizedSpec.Key] = cloneModuleSpec(normalizedSpec)
return nil
}
func (r *ValidatorRegistry) Build(key string) (contracts.Validator, error) {
if r == nil {
return nil, fmt.Errorf("validator registry must not be nil")
}
normalizedKey := strings.TrimSpace(key)
if normalizedKey == "" {
return nil, fmt.Errorf("validator key must not be empty")
}
constructor, ok := r.constructors[normalizedKey]
if !ok {
return nil, fmt.Errorf("validator %q is not registered", normalizedKey)
}
validator, err := constructor()
if err != nil {
return nil, fmt.Errorf("build validator %q: %w", normalizedKey, err)
}
if validator == nil {
return nil, fmt.Errorf("validator %q constructor returned nil", normalizedKey)
}
if validator.Name() != normalizedKey {
return nil, fmt.Errorf("validator %q returned name %q", normalizedKey, validator.Name())
}
return validator, nil
}
func (r *ValidatorRegistry) Spec(key string) (ModuleSpec, bool) {
if r == nil {
return ModuleSpec{}, false
}
spec, ok := r.specs[strings.TrimSpace(key)]
if !ok {
return ModuleSpec{}, false
}
return cloneModuleSpec(spec), true
}
func (r *ValidatorRegistry) RegisteredKeys() []string {
if r == nil {
return nil
}
return sortedRegistryKeys(r.constructors)
}

View File

@@ -0,0 +1,58 @@
package pipeline
import (
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestValidatorRegistryBehavior(t *testing.T) {
runRegistryBehaviorTests(t, registryBehaviorCase[contracts.Validator]{
name: "ValidatorRegistry",
key: "generic-validator",
stage: StageValidate,
wrongStage: StageExtract,
newRegistry: func() any {
return NewValidatorRegistry()
},
register: func(registry any, key string, constructor func() (contracts.Validator, error)) error {
return registry.(*ValidatorRegistry).Register(key, constructor)
},
registerWithSpec: func(registry any, spec ModuleSpec, constructor func() (contracts.Validator, error)) error {
return registry.(*ValidatorRegistry).RegisterWithSpec(spec, constructor)
},
build: func(registry any, key string) (contracts.Validator, error) {
return registry.(*ValidatorRegistry).Build(key)
},
spec: func(registry any, key string) (ModuleSpec, bool) {
return registry.(*ValidatorRegistry).Spec(key)
},
registeredKeys: func(registry any) []string {
return registry.(*ValidatorRegistry).RegisteredKeys()
},
nilRegister: func(key string, constructor func() (contracts.Validator, error)) error {
var registry *ValidatorRegistry
return registry.Register(key, constructor)
},
nilBuild: func(key string) (contracts.Validator, error) {
var registry *ValidatorRegistry
return registry.Build(key)
},
nilSpec: func(key string) (ModuleSpec, bool) {
var registry *ValidatorRegistry
return registry.Spec(key)
},
nilRegisteredKey: func() []string {
var registry *ValidatorRegistry
return registry.RegisteredKeys()
},
constructor: func(key string) func() (contracts.Validator, error) {
return func() (contracts.Validator, error) {
return registryValidator{name: key}, nil
}
},
moduleKey: func(module contracts.Validator) string {
return module.Name()
},
})
}

View File

@@ -0,0 +1,379 @@
package pipeline
import (
"context"
"crypto/sha256"
"encoding/hex"
"encoding/json"
"fmt"
"os"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func TestWalkingSkeletonFixture(t *testing.T) {
inputBytes := readTestFixture(t, "testdata/walking_skeleton_input.json")
expectedBytes := readTestFixture(t, "testdata/walking_skeleton_output.json")
llmClient := &walkingSkeletonLLMClient{}
resolved, err := ResolvePipeline(walkingSkeletonProfile(), ResolveOptions{}, walkingSkeletonCatalog(t))
if err != nil {
t.Fatalf("ResolvePipeline() error = %v, want nil", err)
}
output, err := New(walkingSkeletonRegistries(t)).Run(context.Background(), RunInput{
Pipeline: resolved,
SourceID: "fixture-source",
Path: "walking_skeleton_input.json",
RawInput: inputBytes,
LLMClient: llmClient,
})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
if output.ContentType != "application/json" {
t.Fatalf("ContentType = %q, want application/json", output.ContentType)
}
assertStructuralJSONEqual(t, output.EncodedOutput, expectedBytes)
if llmClient.calls != 2 {
t.Fatalf("LLM calls = %d, want chunk count 2", llmClient.calls)
}
}
func TestWalkingSkeletonResolutionRejectsMissingCapability(t *testing.T) {
catalog := walkingSkeletonCatalog(t)
catalog.Extractors = NewExtractorRegistry()
if err := catalog.Extractors.RegisterWithSpec(ModuleSpec{
Key: "fake/extract",
Stage: StageExtract,
Requires: []string{"missing"},
Provides: []string{"fake_artifacts"},
}, func() (contracts.Extractor, error) {
return walkingSkeletonExtractor{}, nil
}); err != nil {
t.Fatalf("RegisterWithSpec() error = %v, want nil", err)
}
_, err := ResolvePipeline(walkingSkeletonProfile(), ResolveOptions{}, catalog)
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
if !strings.Contains(err.Error(), "missing") {
t.Fatalf("ResolvePipeline() error = %q, want missing capability", err.Error())
}
}
func TestWalkingSkeletonResolutionRejectsUnknownOnlyLane(t *testing.T) {
_, err := ResolvePipeline(walkingSkeletonProfile(), ResolveOptions{Only: []string{"missing"}}, walkingSkeletonCatalog(t))
if err == nil {
t.Fatal("ResolvePipeline() error = nil, want error")
}
if !strings.Contains(err.Error(), "missing") || !strings.Contains(err.Error(), "not declared") {
t.Fatalf("ResolvePipeline() error = %q, want unknown lane error", err.Error())
}
}
func walkingSkeletonProfile() PipelineProfile {
return PipelineProfile{
ID: "walking-skeleton",
Input: Binding("fake/input"),
Chunk: Binding("fake/chunk"),
Output: Binding("json"),
Artifacts: map[string]ArtifactLaneProfile{
"events": {Extract: Binding("fake/extract")},
},
}
}
func walkingSkeletonCatalog(t *testing.T) ModuleCatalog {
t.Helper()
catalog := ModuleCatalog{
Inputs: NewInputAdapterRegistry(),
Chunkers: NewChunkerRegistry(),
Extractors: NewExtractorRegistry(),
Mergers: NewMergerRegistry(),
Normalizers: NewNormalizerRegistry(),
Outputs: NewOutputEncoderRegistry(),
}
if err := catalog.Inputs.RegisterWithSpec(ModuleSpec{
Key: "fake/input",
Stage: StageInput,
Provides: []string{"plain_text"},
}, func() (contracts.InputAdapter, error) {
return walkingSkeletonInput{}, nil
}); err != nil {
t.Fatalf("register fake input: %v", err)
}
if err := catalog.Chunkers.RegisterWithSpec(ModuleSpec{
Key: "fake/chunk",
Stage: StageChunk,
Requires: []string{"plain_text"},
Provides: []string{"chunks"},
}, func() (contracts.Chunker, error) {
return walkingSkeletonChunker{}, nil
}); err != nil {
t.Fatalf("register fake chunker: %v", err)
}
if err := catalog.Extractors.RegisterWithSpec(ModuleSpec{
Key: "fake/extract",
Stage: StageExtract,
Requires: []string{"chunks"},
Provides: []string{"fake_artifacts"},
}, func() (contracts.Extractor, error) {
return walkingSkeletonExtractor{}, nil
}); err != nil {
t.Fatalf("register fake extractor: %v", err)
}
if err := catalog.Mergers.RegisterWithSpec(ModuleSpec{
Key: DefaultMergeModule,
Stage: StageMerge,
Requires: []string{"fake_artifacts"},
}, func() (contracts.Merger, error) {
return AppendOrderMerger{}, nil
}); err != nil {
t.Fatalf("register append-order merger: %v", err)
}
if err := catalog.Normalizers.RegisterWithSpec(ModuleSpec{
Key: DefaultNormalizeModule,
Stage: StageNormalize,
}, func() (contracts.Normalizer, error) {
return NoopNormalizer{}, nil
}); err != nil {
t.Fatalf("register no-op normalizer: %v", err)
}
if err := catalog.Outputs.RegisterWithSpec(ModuleSpec{
Key: "json",
Stage: StageOutput,
}, func() (contracts.OutputEncoder, error) {
return walkingSkeletonOutput{}, nil
}); err != nil {
t.Fatalf("register fake output: %v", err)
}
return catalog
}
func walkingSkeletonRegistries(t *testing.T) Registries {
t.Helper()
catalog := walkingSkeletonCatalog(t)
return Registries{
Inputs: catalog.Inputs,
Chunkers: catalog.Chunkers,
Extractors: catalog.Extractors,
Mergers: catalog.Mergers,
Normalizers: catalog.Normalizers,
Outputs: catalog.Outputs,
}
}
type walkingSkeletonInput struct{}
func (input walkingSkeletonInput) Key() string {
return "fake/input"
}
func (input walkingSkeletonInput) Parse(ctx context.Context, req contracts.ParseRequest) (*source.SourceDocument, error) {
var fixture struct {
ID string `json:"id"`
Units []struct {
ID string `json:"id"`
Text string `json:"text"`
} `json:"units"`
}
if err := json.Unmarshal(req.Raw, &fixture); err != nil {
return nil, err
}
units := make([]source.SourceUnit, 0, len(fixture.Units))
for _, unit := range fixture.Units {
units = append(units, source.SourceUnit{
ID: unit.ID,
Kind: "unit",
Text: unit.Text,
})
}
return &source.SourceDocument{
ID: fixture.ID,
Kind: "fixture",
Format: "application/json",
Digest: rawDigest(req.Raw),
Units: units,
}, nil
}
type walkingSkeletonChunker struct{}
func (chunker walkingSkeletonChunker) Key() string {
return "fake/chunk"
}
func (chunker walkingSkeletonChunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) {
if len(req.Source.Units) < 3 {
return contracts.ChunkResult{}, fmt.Errorf("fixture source must contain at least three units")
}
return contracts.ChunkResult{
Chunks: []contracts.SourceChunk{
{
ID: req.Source.ID + ":chunk:0",
SourceID: req.Source.ID,
Index: 0,
Units: append([]source.SourceUnit(nil), req.Source.Units[:2]...),
},
{
ID: req.Source.ID + ":chunk:1",
SourceID: req.Source.ID,
Index: 1,
Units: append([]source.SourceUnit(nil), req.Source.Units[2:]...),
},
},
}, nil
}
type walkingSkeletonExtractor struct{}
func (extractor walkingSkeletonExtractor) Key() string {
return "fake/extract"
}
func (extractor walkingSkeletonExtractor) ArtifactType() string {
return "fake_event"
}
func (extractor walkingSkeletonExtractor) SchemaVersion() string {
return "v1"
}
func (extractor walkingSkeletonExtractor) Validators() []contracts.Validator {
return nil
}
func (extractor walkingSkeletonExtractor) Extract(ctx context.Context, req contracts.ExtractionRequest) (contracts.ExtractionResult, error) {
var response struct {
Call int `json:"call"`
}
if _, err := req.LLMClient.CompleteStructured(ctx, contracts.StructuredCompletionRequest{
StageName: "fake/extract",
ResponseSchemaName: "fake_event",
}, &response); err != nil {
return contracts.ExtractionResult{}, err
}
payload, err := json.Marshal(map[string]any{
"chunk_id": req.Chunk.ID,
"llm_call": response.Call,
"text": chunkText(req.Chunk.Units),
})
if err != nil {
return contracts.ExtractionResult{}, err
}
return contracts.ExtractionResult{
Candidates: []artifacts.ArtifactCandidate{
{
Payload: payload,
SourceRefs: []source.SourceRef{
{
SourceID: req.Source.ID,
StartUnitID: req.Chunk.Units[0].ID,
EndUnitID: req.Chunk.Units[len(req.Chunk.Units)-1].ID,
},
},
},
},
}, nil
}
type walkingSkeletonLLMClient struct {
calls int
}
func (client *walkingSkeletonLLMClient) CompleteStructured(ctx context.Context, req contracts.StructuredCompletionRequest, out any) (contracts.StructuredCompletionResponse, error) {
client.calls++
if response, ok := out.(*struct {
Call int `json:"call"`
}); ok {
response.Call = client.calls
}
content, err := json.Marshal(map[string]any{"call": client.calls})
if err != nil {
return contracts.StructuredCompletionResponse{}, err
}
return contracts.StructuredCompletionResponse{
Content: content,
}, nil
}
type walkingSkeletonOutput struct{}
func (output walkingSkeletonOutput) Key() string {
return "json"
}
func (output walkingSkeletonOutput) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
encoded, err := json.Marshal(struct {
Manifest artifacts.RunManifest `json:"manifest"`
Approved []artifacts.Artifact `json:"approved"`
}{
Manifest: artifacts.RunManifest{
PipelineID: req.Manifest.PipelineID,
PipelineDigest: req.Manifest.PipelineDigest,
ArtifactLanes: req.Manifest.ArtifactLanes,
ValidationStatus: req.Manifest.ValidationStatus,
},
Approved: req.Approved,
})
if err != nil {
return contracts.OutputResult{}, err
}
return contracts.OutputResult{
Bytes: encoded,
ContentType: "application/json",
}, nil
}
func readTestFixture(t *testing.T, path string) []byte {
t.Helper()
bytes, err := os.ReadFile(path)
if err != nil {
t.Fatalf("read fixture %q: %v", path, err)
}
return bytes
}
func assertStructuralJSONEqual(t *testing.T, gotBytes, wantBytes []byte) {
t.Helper()
var got any
if err := json.Unmarshal(gotBytes, &got); err != nil {
t.Fatalf("unmarshal actual JSON: %v\n%s", err, gotBytes)
}
var want any
if err := json.Unmarshal(wantBytes, &want); err != nil {
t.Fatalf("unmarshal expected JSON: %v\n%s", err, wantBytes)
}
if !reflect.DeepEqual(got, want) {
gotFormatted, _ := json.MarshalIndent(got, "", " ")
wantFormatted, _ := json.MarshalIndent(want, "", " ")
t.Fatalf("actual JSON:\n%s\nwant:\n%s", gotFormatted, wantFormatted)
}
}
func chunkText(units []source.SourceUnit) string {
parts := make([]string, 0, len(units))
for _, unit := range units {
parts = append(parts, unit.Text)
}
return strings.Join(parts, " ")
}
func rawDigest(raw []byte) string {
sum := sha256.Sum256(raw)
return "sha256:" + hex.EncodeToString(sum[:])
}