Organize generic and Seriatim modules by domain

internal/modules/generic/chunk/units/chunker.go

@@ -0,0 +1,265 @@
+package units
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"math"
+	"strconv"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "generic"
+
+const (
+	defaultMaxUnits     = 50
+	defaultOverlapUnits = 0
+)
+
+var _ contracts.Chunker = (*Chunker)(nil)
+
+type Chunker struct{}
+
+func New() *Chunker {
+	return &Chunker{}
+}
+
+func (c *Chunker) Key() string {
+	return Key
+}
+
+func (c *Chunker) ReferenceSlots() []contracts.ReferenceSlot {
+	return nil
+}
+
+func (c *Chunker) Chunk(ctx context.Context, req contracts.ChunkRequest) (contracts.ChunkResult, error) {
+	if c == nil {
+		return contracts.ChunkResult{}, chunkerErrorf("chunker must not be nil")
+	}
+	if ctx == nil {
+		return contracts.ChunkResult{}, chunkerErrorf("context must not be nil")
+	}
+	if err := ctx.Err(); err != nil {
+		return contracts.ChunkResult{}, chunkerErrorf("context error before chunking: %w", err)
+	}
+	if req.Source == nil {
+		return contracts.ChunkResult{}, chunkerErrorf("source must not be nil")
+	}
+	if len(req.Source.Units) == 0 {
+		return contracts.ChunkResult{}, chunkerErrorf("source units must not be empty")
+	}
+	if err := source.ValidateDocument(req.Source); err != nil {
+		return contracts.ChunkResult{}, chunkerErrorf("validate source document: %w", err)
+	}
+
+	opts, err := chunkOptionsFrom(req.Options)
+	if err != nil {
+		return contracts.ChunkResult{}, err
+	}
+
+	step := opts.maxUnits - opts.overlapUnits
+	chunks := make([]contracts.SourceChunk, 0, (len(req.Source.Units)+step-1)/step)
+	for start := 0; start < len(req.Source.Units); start += step {
+		end := start + opts.maxUnits
+		if end > len(req.Source.Units) {
+			end = len(req.Source.Units)
+		}
+		units := cloneUnits(req.Source.Units[start:end])
+		content, err := chunkContent(units)
+		if err != nil {
+			return contracts.ChunkResult{}, err
+		}
+		chunks = append(chunks, contracts.SourceChunk{
+			ID:          fmt.Sprintf("chunk-%06d", len(chunks)+1),
+			SourceID:    req.Source.ID,
+			Index:       len(chunks),
+			StartUnitID: units[0].ID,
+			EndUnitID:   units[len(units)-1].ID,
+			Content:     content,
+			MediaType:   "application/json",
+			Units:       units,
+			Metadata: map[string]any{
+				"start_unit_id": units[0].ID,
+				"end_unit_id":   units[len(units)-1].ID,
+				"unit_count":    len(units),
+			},
+		})
+		if end == len(req.Source.Units) {
+			break
+		}
+	}
+
+	return contracts.ChunkResult{Chunks: chunks}, nil
+}
+
+func chunkContent(units []source.SourceUnit) ([]byte, error) {
+	content, err := json.Marshal(struct {
+		Units []source.SourceUnit `json:"units"`
+	}{
+		Units: units,
+	})
+	if err != nil {
+		return nil, chunkerErrorf("encode chunk content: %w", err)
+	}
+	return content, nil
+}
+
+func ModuleSpec() pipeline.ModuleSpec {
+	return pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageChunk,
+		Provides: []string{"chunks"},
+	}
+}
+
+func Register(registry *pipeline.ChunkerRegistry) error {
+	return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.Chunker, error) {
+		return New(), nil
+	})
+}
+
+type chunkOptions struct {
+	maxUnits     int
+	overlapUnits int
+}
+
+func chunkOptionsFrom(options map[string]any) (chunkOptions, error) {
+	opts := chunkOptions{
+		maxUnits:     defaultMaxUnits,
+		overlapUnits: defaultOverlapUnits,
+	}
+	var err error
+	if value, ok := options["max_units"]; ok {
+		opts.maxUnits, err = positiveIntOption("max_units", value)
+		if err != nil {
+			return chunkOptions{}, err
+		}
+	}
+	if value, ok := options["overlap_units"]; ok {
+		opts.overlapUnits, err = nonNegativeIntOption("overlap_units", value)
+		if err != nil {
+			return chunkOptions{}, err
+		}
+	}
+	if opts.overlapUnits >= opts.maxUnits {
+		return chunkOptions{}, chunkerErrorf("overlap_units must be less than max_units")
+	}
+	return opts, nil
+}
+
+func positiveIntOption(name string, value any) (int, error) {
+	got, err := intOption(name, value)
+	if err != nil {
+		return 0, err
+	}
+	if got <= 0 {
+		return 0, chunkerErrorf("%s must be positive", name)
+	}
+	return got, nil
+}
+
+func nonNegativeIntOption(name string, value any) (int, error) {
+	got, err := intOption(name, value)
+	if err != nil {
+		return 0, err
+	}
+	if got < 0 {
+		return 0, chunkerErrorf("%s must be non-negative", name)
+	}
+	return got, nil
+}
+
+func intOption(name string, value any) (int, error) {
+	switch typed := value.(type) {
+	case int:
+		return typed, nil
+	case int8:
+		return int(typed), nil
+	case int16:
+		return int(typed), nil
+	case int32:
+		return int(typed), nil
+	case int64:
+		if typed > maxInt() || typed < minInt() {
+			return 0, chunkerErrorf("%s is outside supported integer range", name)
+		}
+		return int(typed), nil
+	case uint:
+		if uint64(typed) > uint64(maxInt()) {
+			return 0, chunkerErrorf("%s is outside supported integer range", name)
+		}
+		return int(typed), nil
+	case uint8:
+		return int(typed), nil
+	case uint16:
+		return int(typed), nil
+	case uint32:
+		if uint64(typed) > uint64(maxInt()) {
+			return 0, chunkerErrorf("%s is outside supported integer range", name)
+		}
+		return int(typed), nil
+	case uint64:
+		if typed > uint64(maxInt()) {
+			return 0, chunkerErrorf("%s is outside supported integer range", name)
+		}
+		return int(typed), nil
+	case float64:
+		if typed != math.Trunc(typed) {
+			return 0, chunkerErrorf("%s must be an integer", name)
+		}
+		if typed > float64(maxInt()) || typed < float64(minInt()) {
+			return 0, chunkerErrorf("%s is outside supported integer range", name)
+		}
+		return int(typed), nil
+	case json.Number:
+		parsed, err := typed.Int64()
+		if err != nil {
+			return 0, chunkerErrorf("%s must be an integer", name)
+		}
+		if parsed > maxInt() || parsed < minInt() {
+			return 0, chunkerErrorf("%s is outside supported integer range", name)
+		}
+		return int(parsed), nil
+	default:
+		return 0, chunkerErrorf("%s must be an integer", name)
+	}
+}
+
+func maxInt() int64 {
+	return int64(1<<(strconv.IntSize-1) - 1)
+}
+
+func minInt() int64 {
+	return -maxInt() - 1
+}
+
+func cloneUnits(units []source.SourceUnit) []source.SourceUnit {
+	out := make([]source.SourceUnit, 0, len(units))
+	for _, unit := range units {
+		out = append(out, source.SourceUnit{
+			ID:       unit.ID,
+			Kind:     unit.Kind,
+			Text:     unit.Text,
+			Metadata: cloneMetadata(unit.Metadata),
+		})
+	}
+	return out
+}
+
+func cloneMetadata(metadata map[string]any) map[string]any {
+	if len(metadata) == 0 {
+		return nil
+	}
+	out := make(map[string]any, len(metadata))
+	for key, value := range metadata {
+		out[key] = value
+	}
+	return out
+}
+
+func chunkerErrorf(format string, args ...any) error {
+	return fmt.Errorf("generic chunker: "+format, args...)
+}

internal/modules/generic/chunk/units/chunker_test.go

@@ -0,0 +1,221 @@
+package units
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"reflect"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestModuleSpecAndRegister(t *testing.T) {
+	want := pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageChunk,
+		Provides: []string{"chunks"},
+	}
+	if got := ModuleSpec(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ModuleSpec() = %#v, want %#v", got, want)
+	}
+
+	registry := pipeline.NewChunkerRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	spec, ok := registry.Spec(Key)
+	if !ok {
+		t.Fatalf("Spec(%q) ok = false, want true", Key)
+	}
+	if !reflect.DeepEqual(spec, want) {
+		t.Fatalf("registered spec = %#v, want %#v", spec, want)
+	}
+	chunker, err := registry.Build(Key)
+	if err != nil {
+		t.Fatalf("Build(%q) error = %v, want nil", Key, err)
+	}
+	if chunker.Key() != Key {
+		t.Fatalf("Key() = %q, want %q", chunker.Key(), Key)
+	}
+	if slots := chunker.ReferenceSlots(); len(slots) != 0 {
+		t.Fatalf("ReferenceSlots() = %#v, want none", slots)
+	}
+}
+
+func TestChunkUsesDefaultsForSingleChunk(t *testing.T) {
+	result, err := New().Chunk(context.Background(), contracts.ChunkRequest{Source: testSource(3), Options: nil})
+	if err != nil {
+		t.Fatalf("Chunk() error = %v, want nil", err)
+	}
+
+	if got := chunkIDs(result.Chunks); !reflect.DeepEqual(got, []string{"chunk-000001"}) {
+		t.Fatalf("chunk IDs = %#v, want one stable ID", got)
+	}
+	chunk := result.Chunks[0]
+	if chunk.Index != 0 || chunk.SourceID != "source-1" {
+		t.Fatalf("chunk = %#v, want source and index fields", chunk)
+	}
+	if got := unitIDs(chunk.Units); !reflect.DeepEqual(got, []int{1, 2, 3}) {
+		t.Fatalf("unit IDs = %#v, want all units", got)
+	}
+	if chunk.StartUnitID != 1 || chunk.EndUnitID != 3 {
+		t.Fatalf("chunk boundaries = %d-%d, want 1-3", chunk.StartUnitID, chunk.EndUnitID)
+	}
+	if chunk.MediaType != "application/json" || len(chunk.Content) == 0 {
+		t.Fatalf("chunk payload = media type %q length %d, want JSON content", chunk.MediaType, len(chunk.Content))
+	}
+	if chunk.Metadata["start_unit_id"] != 1 || chunk.Metadata["end_unit_id"] != 3 || chunk.Metadata["unit_count"] != 3 {
+		t.Fatalf("metadata = %#v, want chunk bounds", chunk.Metadata)
+	}
+}
+
+func TestChunkExactBoundaries(t *testing.T) {
+	result, err := New().Chunk(context.Background(), contracts.ChunkRequest{
+		Source:  testSource(6),
+		Options: map[string]any{"max_units": 2},
+	})
+	if err != nil {
+		t.Fatalf("Chunk() error = %v, want nil", err)
+	}
+
+	if got := chunkIDs(result.Chunks); !reflect.DeepEqual(got, []string{"chunk-000001", "chunk-000002", "chunk-000003"}) {
+		t.Fatalf("chunk IDs = %#v, want stable IDs", got)
+	}
+	gotUnits := [][]int{unitIDs(result.Chunks[0].Units), unitIDs(result.Chunks[1].Units), unitIDs(result.Chunks[2].Units)}
+	wantUnits := [][]int{{1, 2}, {3, 4}, {5, 6}}
+	if !reflect.DeepEqual(gotUnits, wantUnits) {
+		t.Fatalf("chunk units = %#v, want %#v", gotUnits, wantUnits)
+	}
+}
+
+func TestChunkOverlap(t *testing.T) {
+	result, err := New().Chunk(context.Background(), contracts.ChunkRequest{
+		Source:  testSource(7),
+		Options: map[string]any{"max_units": 3, "overlap_units": 1},
+	})
+	if err != nil {
+		t.Fatalf("Chunk() error = %v, want nil", err)
+	}
+
+	gotUnits := make([][]int, 0, len(result.Chunks))
+	for _, chunk := range result.Chunks {
+		gotUnits = append(gotUnits, unitIDs(chunk.Units))
+	}
+	wantUnits := [][]int{{1, 2, 3}, {3, 4, 5}, {5, 6, 7}}
+	if !reflect.DeepEqual(gotUnits, wantUnits) {
+		t.Fatalf("chunk units = %#v, want %#v", gotUnits, wantUnits)
+	}
+}
+
+func TestChunkRejectsInvalidOptions(t *testing.T) {
+	tests := []struct {
+		name    string
+		options map[string]any
+		want    string
+	}{
+		{name: "max wrong type", options: map[string]any{"max_units": "2"}, want: "max_units"},
+		{name: "max fractional", options: map[string]any{"max_units": 1.5}, want: "integer"},
+		{name: "max zero", options: map[string]any{"max_units": 0}, want: "positive"},
+		{name: "overlap negative", options: map[string]any{"overlap_units": -1}, want: "non-negative"},
+		{name: "overlap too large", options: map[string]any{"max_units": 2, "overlap_units": 2}, want: "less than"},
+		{name: "json number", options: map[string]any{"max_units": json.Number("bad")}, want: "integer"},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			_, err := New().Chunk(context.Background(), contracts.ChunkRequest{
+				Source:  testSource(3),
+				Options: test.options,
+			})
+			if err == nil {
+				t.Fatal("Chunk() error = nil, want error")
+			}
+			if !strings.Contains(err.Error(), "generic chunker") || !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("Chunk() error = %q, want module context and %q", err.Error(), test.want)
+			}
+		})
+	}
+}
+
+func TestChunkRejectsEmptySource(t *testing.T) {
+	doc := testSource(1)
+	doc.Units = nil
+
+	_, err := New().Chunk(context.Background(), contracts.ChunkRequest{Source: doc})
+	if err == nil {
+		t.Fatal("Chunk() error = nil, want empty source error")
+	}
+	if !strings.Contains(err.Error(), "generic chunker") || !strings.Contains(err.Error(), "units") {
+		t.Fatalf("Chunk() error = %q, want empty source context", err.Error())
+	}
+}
+
+func TestChunkDefensivelyCopiesUnits(t *testing.T) {
+	doc := testSource(2)
+
+	result, err := New().Chunk(context.Background(), contracts.ChunkRequest{
+		Source:  doc,
+		Options: map[string]any{"max_units": 1},
+	})
+	if err != nil {
+		t.Fatalf("Chunk() error = %v, want nil", err)
+	}
+	if len(result.Chunks) != 2 {
+		t.Fatalf("len(Chunks) = %d, want 2", len(result.Chunks))
+	}
+
+	doc.Units[0].ID = 99
+	doc.Units[0].Metadata["speaker"] = "changed"
+
+	if result.Chunks[0].Units[0].ID != 1 {
+		t.Fatalf("chunk unit ID changed after source mutation: %#v", result.Chunks[0].Units[0])
+	}
+	if result.Chunks[0].Units[0].Metadata["speaker"] != "speaker-001" {
+		t.Fatalf("chunk unit metadata changed after source mutation: %#v", result.Chunks[0].Units[0].Metadata)
+	}
+}
+
+func testSource(count int) *source.SourceDocument {
+	units := make([]source.SourceUnit, 0, count)
+	for i := 1; i <= count; i++ {
+		units = append(units, source.SourceUnit{
+			ID:   i,
+			Kind: "unit",
+			Text: "Text for " + zeroPad3(i),
+			Metadata: map[string]any{
+				"speaker": "speaker-" + zeroPad3(i),
+			},
+		})
+	}
+	return &source.SourceDocument{
+		ID:     "source-1",
+		Kind:   "document",
+		Format: "text/plain",
+		Digest: "sha256:source",
+		Units:  units,
+	}
+}
+
+func zeroPad3(value int) string {
+	return fmt.Sprintf("%03d", value)
+}
+
+func chunkIDs(chunks []contracts.SourceChunk) []string {
+	ids := make([]string, 0, len(chunks))
+	for _, chunk := range chunks {
+		ids = append(ids, chunk.ID)
+	}
+	return ids
+}
+
+func unitIDs(units []source.SourceUnit) []int {
+	ids := make([]int, 0, len(units))
+	for _, unit := range units {
+		ids = append(ids, unit.ID)
+	}
+	return ids
+}

internal/modules/generic/merge/appendorder/merger.go

@@ -0,0 +1,236 @@
+package appendorder
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"mime"
+	"sort"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "appendorder"
+
+var _ contracts.Merger = (*Merger)(nil)
+
+type Merger struct{}
+
+func New() *Merger {
+	return &Merger{}
+}
+
+func (m *Merger) Key() string {
+	return Key
+}
+
+func (m *Merger) Merge(ctx context.Context, req contracts.MergeRequest) (contracts.MergeResult, error) {
+	if m == nil {
+		return contracts.MergeResult{}, mergerErrorf("merger must not be nil")
+	}
+	if ctx == nil {
+		return contracts.MergeResult{}, mergerErrorf("context must not be nil")
+	}
+	if err := ctx.Err(); err != nil {
+		return contracts.MergeResult{}, mergerErrorf("context error before merge: %w", err)
+	}
+
+	outputs, err := orderedOutputs(req.ExtractOutputs)
+	if err != nil {
+		return contracts.MergeResult{}, err
+	}
+	if len(outputs) == 1 {
+		payload := cloneRawPayload(outputs[0].Payload)
+		return contracts.MergeResult{
+			Output: contracts.MergeOutput{
+				LaneID:    req.LaneID,
+				MergerKey: Key,
+				SourceID:  outputs[0].SourceID,
+				Schema:    outputs[0].Schema,
+				Payload:   payload,
+			},
+		}, nil
+	}
+
+	content, err := mergedContent(outputs)
+	if err != nil {
+		return contracts.MergeResult{}, err
+	}
+	return contracts.MergeResult{
+		Output: contracts.MergeOutput{
+			LaneID:    req.LaneID,
+			MergerKey: Key,
+			SourceID:  sourceID(outputs),
+			Schema:    commonSchema(outputs),
+			Payload: contracts.RawPayload{
+				Content:   content,
+				MediaType: "application/json",
+			},
+		},
+	}, nil
+}
+
+func ModuleSpec() pipeline.ModuleSpec {
+	return pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageMerge,
+		Provides: []string{"merged"},
+	}
+}
+
+func Register(registry *pipeline.MergerRegistry) error {
+	return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.Merger, error) {
+		return New(), nil
+	})
+}
+
+func orderedOutputs(outputs []contracts.ExtractOutput) ([]contracts.ExtractOutput, error) {
+	ordered := make([]contracts.ExtractOutput, 0, len(outputs))
+	for _, output := range outputs {
+		if !isJSONMediaType(output.Payload.MediaType) {
+			return nil, mergerErrorf("extract output for chunk %q has unsupported media type %q", output.ChunkID, output.Payload.MediaType)
+		}
+		if !json.Valid(output.Payload.Content) {
+			return nil, mergerErrorf("extract output for chunk %q contains invalid JSON", output.ChunkID)
+		}
+		ordered = append(ordered, cloneExtractOutput(output))
+	}
+	sort.SliceStable(ordered, func(i, j int) bool {
+		return ordered[i].ChunkIndex < ordered[j].ChunkIndex
+	})
+	return ordered, nil
+}
+
+func mergedContent(outputs []contracts.ExtractOutput) ([]byte, error) {
+	values := make([]any, 0, len(outputs))
+	objects := make([]map[string]any, 0, len(outputs))
+	for _, output := range outputs {
+		var value any
+		if err := json.Unmarshal(output.Payload.Content, &value); err != nil {
+			return nil, mergerErrorf("decode extract output for chunk %q: %w", output.ChunkID, err)
+		}
+		values = append(values, value)
+		object, ok := value.(map[string]any)
+		if !ok {
+			continue
+		}
+		objects = append(objects, object)
+	}
+
+	if len(objects) == len(outputs) {
+		if field, ok := commonArrayField(objects); ok {
+			merged := make([]any, 0)
+			for _, object := range objects {
+				items := object[field].([]any)
+				merged = append(merged, items...)
+			}
+			return marshalMerged(map[string]any{field: merged})
+		}
+	}
+	return marshalMerged(values)
+}
+
+func commonArrayField(objects []map[string]any) (string, bool) {
+	if len(objects) == 0 {
+		return "", false
+	}
+	candidates := map[string]struct{}{}
+	for key, value := range objects[0] {
+		if _, ok := value.([]any); ok {
+			candidates[key] = struct{}{}
+		}
+	}
+	for _, object := range objects[1:] {
+		for key := range candidates {
+			if _, ok := object[key].([]any); !ok {
+				delete(candidates, key)
+			}
+		}
+	}
+	if len(candidates) != 1 {
+		return "", false
+	}
+	for key := range candidates {
+		return key, true
+	}
+	return "", false
+}
+
+func marshalMerged(value any) ([]byte, error) {
+	content, err := json.Marshal(value)
+	if err != nil {
+		return nil, mergerErrorf("encode merged output: %w", err)
+	}
+	return content, nil
+}
+
+func isJSONMediaType(mediaType string) bool {
+	base, _, err := mime.ParseMediaType(strings.TrimSpace(mediaType))
+	if err != nil {
+		base = strings.TrimSpace(mediaType)
+	}
+	return strings.EqualFold(base, "application/json")
+}
+
+func sourceID(outputs []contracts.ExtractOutput) string {
+	for _, output := range outputs {
+		if output.SourceID != "" {
+			return output.SourceID
+		}
+	}
+	return ""
+}
+
+func commonSchema(outputs []contracts.ExtractOutput) contracts.ResponseSchema {
+	if len(outputs) == 0 {
+		return contracts.ResponseSchema{}
+	}
+	schema := outputs[0].Schema
+	for _, output := range outputs[1:] {
+		if !sameResponseSchema(output.Schema, schema) {
+			return contracts.ResponseSchema{}
+		}
+	}
+	return schema
+}
+
+func sameResponseSchema(left contracts.ResponseSchema, right contracts.ResponseSchema) bool {
+	return left.ID == right.ID && left.Name == right.Name && left.Version == right.Version && string(left.JSONSchema) == string(right.JSONSchema)
+}
+
+func cloneExtractOutput(output contracts.ExtractOutput) contracts.ExtractOutput {
+	output.Schema = cloneResponseSchema(output.Schema)
+	output.Payload = cloneRawPayload(output.Payload)
+	return output
+}
+
+func cloneResponseSchema(schema contracts.ResponseSchema) contracts.ResponseSchema {
+	schema.JSONSchema = append([]byte(nil), schema.JSONSchema...)
+	return schema
+}
+
+func cloneRawPayload(payload contracts.RawPayload) contracts.RawPayload {
+	return contracts.RawPayload{
+		Content:   append([]byte(nil), payload.Content...),
+		MediaType: payload.MediaType,
+		Metadata:  cloneMetadata(payload.Metadata),
+		Warnings:  append([]contracts.Warning(nil), payload.Warnings...),
+	}
+}
+
+func cloneMetadata(metadata map[string]any) map[string]any {
+	if len(metadata) == 0 {
+		return nil
+	}
+	out := make(map[string]any, len(metadata))
+	for key, value := range metadata {
+		out[key] = value
+	}
+	return out
+}
+
+func mergerErrorf(format string, args ...any) error {
+	return fmt.Errorf("appendorder merger: "+format, args...)
+}

internal/modules/generic/merge/appendorder/merger_test.go

@@ -0,0 +1,191 @@
+package appendorder
+
+import (
+	"context"
+	"encoding/json"
+	"reflect"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestModuleSpecAndRegister(t *testing.T) {
+	want := pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageMerge,
+		Provides: []string{"merged"},
+	}
+	if got := ModuleSpec(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ModuleSpec() = %#v, want %#v", got, want)
+	}
+
+	registry := pipeline.NewMergerRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	spec, ok := registry.Spec(Key)
+	if !ok {
+		t.Fatalf("Spec(%q) ok = false, want true", Key)
+	}
+	if !reflect.DeepEqual(spec, want) {
+		t.Fatalf("registered spec = %#v, want %#v", spec, want)
+	}
+}
+
+func TestMergePassesThroughSingleExtractOutput(t *testing.T) {
+	input := extractOutput("chunk-0", 0, `{"name":"original"}`)
+
+	result, err := New().Merge(context.Background(), contracts.MergeRequest{
+		LaneID:         "events",
+		ExtractOutputs: []contracts.ExtractOutput{input},
+	})
+	if err != nil {
+		t.Fatalf("Merge() error = %v, want nil", err)
+	}
+
+	if result.Output.LaneID != "events" || result.Output.MergerKey != Key {
+		t.Fatalf("output provenance = %#v, want lane and merger", result.Output)
+	}
+	if string(result.Output.Payload.Content) != `{"name":"original"}` {
+		t.Fatalf("content = %s, want original content", result.Output.Payload.Content)
+	}
+	if result.Output.Payload.Metadata["name"] != "chunk-0" {
+		t.Fatalf("metadata = %#v, want original metadata", result.Output.Payload.Metadata)
+	}
+}
+
+func TestMergeDefensivelyCopiesRawPayload(t *testing.T) {
+	input := extractOutput("chunk-0", 0, `{"name":"original"}`)
+
+	result, err := New().Merge(context.Background(), contracts.MergeRequest{
+		LaneID:         "events",
+		ExtractOutputs: []contracts.ExtractOutput{input},
+	})
+	if err != nil {
+		t.Fatalf("Merge() error = %v, want nil", err)
+	}
+
+	input.Payload.Content[0] = '['
+	input.Payload.Metadata["name"] = "changed"
+
+	if string(result.Output.Payload.Content) != `{"name":"original"}` {
+		t.Fatalf("content changed after input mutation: %s", result.Output.Payload.Content)
+	}
+	if result.Output.Payload.Metadata["name"] != "chunk-0" {
+		t.Fatalf("metadata changed after input mutation: %#v", result.Output.Payload.Metadata)
+	}
+}
+
+func TestMergeConcatenatesCommonTopLevelArrayFieldInChunkOrder(t *testing.T) {
+	result, err := New().Merge(context.Background(), contracts.MergeRequest{
+		LaneID: "events",
+		ExtractOutputs: []contracts.ExtractOutput{
+			extractOutput("chunk-1", 1, `{"events":[{"name":"second"}]}`),
+			extractOutput("chunk-0", 0, `{"events":[{"name":"first"}]}`),
+		},
+	})
+	if err != nil {
+		t.Fatalf("Merge() error = %v, want nil", err)
+	}
+	if result.Output.Payload.MediaType != "application/json" {
+		t.Fatalf("MediaType = %q, want application/json", result.Output.Payload.MediaType)
+	}
+
+	var decoded struct {
+		Events []struct {
+			Name string `json:"name"`
+		} `json:"events"`
+	}
+	if err := json.Unmarshal(result.Output.Payload.Content, &decoded); err != nil {
+		t.Fatalf("Unmarshal() error = %v, want nil", err)
+	}
+	if len(decoded.Events) != 2 || decoded.Events[0].Name != "first" || decoded.Events[1].Name != "second" {
+		t.Fatalf("events = %#v, want concatenated chunk order", decoded.Events)
+	}
+	if result.Output.Schema.ID != "schema-id" {
+		t.Fatalf("schema = %#v, want common extract schema", result.Output.Schema)
+	}
+}
+
+func TestMergeFallsBackToOrderedJSONValueArrayWhenShapesDiffer(t *testing.T) {
+	result, err := New().Merge(context.Background(), contracts.MergeRequest{
+		LaneID: "events",
+		ExtractOutputs: []contracts.ExtractOutput{
+			extractOutput("chunk-1", 1, `{"notes":["second"]}`),
+			extractOutput("chunk-0", 0, `{"events":[{"name":"first"}]}`),
+		},
+	})
+	if err != nil {
+		t.Fatalf("Merge() error = %v, want nil", err)
+	}
+
+	var decoded []map[string]any
+	if err := json.Unmarshal(result.Output.Payload.Content, &decoded); err != nil {
+		t.Fatalf("Unmarshal() error = %v, want nil", err)
+	}
+	if len(decoded) != 2 {
+		t.Fatalf("len(decoded) = %d, want 2", len(decoded))
+	}
+	if _, ok := decoded[0]["events"]; !ok {
+		t.Fatalf("decoded[0] = %#v, want first chunk value", decoded[0])
+	}
+	if _, ok := decoded[1]["notes"]; !ok {
+		t.Fatalf("decoded[1] = %#v, want second chunk value", decoded[1])
+	}
+}
+
+func TestMergeRejectsInvalidJSONAndNonJSONMediaTypes(t *testing.T) {
+	tests := []struct {
+		name   string
+		output contracts.ExtractOutput
+		want   string
+	}{
+		{
+			name:   "invalid JSON",
+			output: extractOutput("chunk-0", 0, `{"events":[`),
+			want:   "invalid JSON",
+		},
+		{
+			name: "non JSON media type",
+			output: func() contracts.ExtractOutput {
+				output := extractOutput("chunk-0", 0, `{"events":[]}`)
+				output.Payload.MediaType = "text/plain"
+				return output
+			}(),
+			want: "unsupported media type",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			_, err := New().Merge(context.Background(), contracts.MergeRequest{
+				LaneID:         "events",
+				ExtractOutputs: []contracts.ExtractOutput{test.output},
+			})
+			if err == nil {
+				t.Fatal("Merge() error = nil, want error")
+			}
+			if !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("Merge() error = %q, want %q", err.Error(), test.want)
+			}
+		})
+	}
+}
+
+func extractOutput(chunkID string, chunkIndex int, content string) contracts.ExtractOutput {
+	return contracts.ExtractOutput{
+		LaneID:       "events",
+		ExtractorKey: "extract",
+		SourceID:     "source-1",
+		ChunkID:      chunkID,
+		ChunkIndex:   chunkIndex,
+		Schema:       contracts.ResponseSchema{ID: "schema-id", Name: "schema-name", Version: "v1"},
+		Payload: contracts.RawPayload{
+			Content:   []byte(content),
+			MediaType: "application/json",
+			Metadata:  map[string]any{"name": chunkID},
+		},
+	}
+}

internal/modules/generic/normalize/noop/normalizer.go

@@ -0,0 +1,87 @@
+package noop
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "noop"
+
+var _ contracts.Normalizer = (*Normalizer)(nil)
+
+type Normalizer struct{}
+
+func New() *Normalizer {
+	return &Normalizer{}
+}
+
+func (n *Normalizer) Key() string {
+	return Key
+}
+
+func (n *Normalizer) ReferenceSlots() []contracts.ReferenceSlot {
+	return nil
+}
+
+func (n *Normalizer) Normalize(ctx context.Context, req contracts.NormalizeRequest) (contracts.NormalizeResult, error) {
+	if n == nil {
+		return contracts.NormalizeResult{}, normalizerErrorf("normalizer must not be nil")
+	}
+	if ctx == nil {
+		return contracts.NormalizeResult{}, normalizerErrorf("context must not be nil")
+	}
+	if err := ctx.Err(); err != nil {
+		return contracts.NormalizeResult{}, normalizerErrorf("context error before normalize: %w", err)
+	}
+	return contracts.NormalizeResult{
+		Output: contracts.NormalizeOutput{
+			LaneID:        req.LaneID,
+			NormalizerKey: Key,
+			SourceID:      req.MergeOutput.SourceID,
+			Schema:        req.MergeOutput.Schema,
+			Payload:       cloneRawPayload(req.MergeOutput.Payload),
+		},
+	}, nil
+}
+
+func ModuleSpec() pipeline.ModuleSpec {
+	return pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageNormalize,
+		Requires: []string{"merged"},
+		Provides: []string{"normalized"},
+	}
+}
+
+func Register(registry *pipeline.NormalizerRegistry) error {
+	return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.Normalizer, error) {
+		return New(), nil
+	})
+}
+
+func cloneRawPayload(payload contracts.RawPayload) contracts.RawPayload {
+	return contracts.RawPayload{
+		Content:   append([]byte(nil), payload.Content...),
+		MediaType: payload.MediaType,
+		Metadata:  cloneMetadata(payload.Metadata),
+		Warnings:  append([]contracts.Warning(nil), payload.Warnings...),
+	}
+}
+
+func cloneMetadata(metadata map[string]any) map[string]any {
+	if len(metadata) == 0 {
+		return nil
+	}
+	out := make(map[string]any, len(metadata))
+	for key, value := range metadata {
+		out[key] = value
+	}
+	return out
+}
+
+func normalizerErrorf(format string, args ...any) error {
+	return fmt.Errorf("noop normalizer: "+format, args...)
+}

internal/modules/generic/normalize/noop/normalizer_test.go

@@ -0,0 +1,99 @@
+package noop
+
+import (
+	"context"
+	"reflect"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestModuleSpecAndRegister(t *testing.T) {
+	want := pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageNormalize,
+		Requires: []string{"merged"},
+		Provides: []string{"normalized"},
+	}
+	if got := ModuleSpec(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ModuleSpec() = %#v, want %#v", got, want)
+	}
+
+	registry := pipeline.NewNormalizerRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	spec, ok := registry.Spec(Key)
+	if !ok {
+		t.Fatalf("Spec(%q) ok = false, want true", Key)
+	}
+	if !reflect.DeepEqual(spec, want) {
+		t.Fatalf("registered spec = %#v, want %#v", spec, want)
+	}
+	normalizer, err := registry.Build(Key)
+	if err != nil {
+		t.Fatalf("Build(%q) error = %v, want nil", Key, err)
+	}
+	if slots := normalizer.ReferenceSlots(); len(slots) != 0 {
+		t.Fatalf("ReferenceSlots() = %#v, want none", slots)
+	}
+}
+
+func TestNormalizePassesThroughMergeOutput(t *testing.T) {
+	input := mergeOutput(`{"name":"original"}`)
+
+	result, err := New().Normalize(context.Background(), contracts.NormalizeRequest{
+		LaneID:      "events",
+		MergeOutput: input,
+	})
+	if err != nil {
+		t.Fatalf("Normalize() error = %v, want nil", err)
+	}
+
+	if result.Output.LaneID != "events" || result.Output.NormalizerKey != Key {
+		t.Fatalf("output provenance = %#v, want lane and normalizer", result.Output)
+	}
+	if string(result.Output.Payload.Content) != `{"name":"original"}` {
+		t.Fatalf("content = %s, want original content", result.Output.Payload.Content)
+	}
+	if result.Output.Payload.Metadata["name"] != "original" {
+		t.Fatalf("metadata = %#v, want original metadata", result.Output.Payload.Metadata)
+	}
+}
+
+func TestNormalizeDefensivelyCopiesRawPayload(t *testing.T) {
+	input := mergeOutput(`{"name":"original"}`)
+
+	result, err := New().Normalize(context.Background(), contracts.NormalizeRequest{
+		LaneID:      "events",
+		MergeOutput: input,
+	})
+	if err != nil {
+		t.Fatalf("Normalize() error = %v, want nil", err)
+	}
+
+	input.Payload.Content[0] = '['
+	input.Payload.Metadata["name"] = "changed"
+
+	if string(result.Output.Payload.Content) != `{"name":"original"}` {
+		t.Fatalf("content changed after input mutation: %s", result.Output.Payload.Content)
+	}
+	if result.Output.Payload.Metadata["name"] != "original" {
+		t.Fatalf("metadata changed after input mutation: %#v", result.Output.Payload.Metadata)
+	}
+}
+
+func mergeOutput(content string) contracts.MergeOutput {
+	return contracts.MergeOutput{
+		LaneID:    "events",
+		MergerKey: "merge",
+		SourceID:  "source-1",
+		Schema:    contracts.ResponseSchema{ID: "schema-id", Name: "schema-name", Version: "v1"},
+		Payload: contracts.RawPayload{
+			Content:   []byte(content),
+			MediaType: "application/json",
+			Metadata:  map[string]any{"name": "original"},
+		},
+	}
+}

internal/modules/generic/output/json/encoder.go

@@ -0,0 +1,272 @@
+package json
+
+import (
+	"context"
+	stdjson "encoding/json"
+	"fmt"
+	"mime"
+	"regexp"
+	"sort"
+	"strings"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "json"
+
+const contentTypeJSON = "application/json"
+
+var safeOutputFileChar = regexp.MustCompile(`[^A-Za-z0-9._-]`)
+
+var _ contracts.OutputEncoder = (*Encoder)(nil)
+
+type Encoder struct{}
+
+func New() *Encoder {
+	return &Encoder{}
+}
+
+func (e *Encoder) Key() string {
+	return Key
+}
+
+func (e *Encoder) Encode(ctx context.Context, req contracts.OutputRequest) (contracts.OutputResult, error) {
+	if e == nil {
+		return contracts.OutputResult{}, encoderErrorf("encoder must not be nil")
+	}
+	if ctx == nil {
+		return contracts.OutputResult{}, encoderErrorf("context must not be nil")
+	}
+	if err := ctx.Err(); err != nil {
+		return contracts.OutputResult{}, encoderErrorf("context error before encoding: %w", err)
+	}
+
+	files, err := logicalFiles(req)
+	if err != nil {
+		return contracts.OutputResult{}, err
+	}
+	return contracts.OutputResult{Files: files}, nil
+}
+
+func ModuleSpec() pipeline.ModuleSpec {
+	return pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageOutput,
+		Requires: []string{"normalized"},
+		Provides: []string{"encoded"},
+	}
+}
+
+func Register(registry *pipeline.OutputEncoderRegistry) error {
+	return registry.RegisterWithSpec(ModuleSpec(), func() (contracts.OutputEncoder, error) {
+		return New(), nil
+	})
+}
+
+type indexFile struct {
+	ManifestFile string            `json:"manifest_file"`
+	OutputFiles  []outputFileIndex `json:"output_files"`
+	RejectedFile string            `json:"rejected_file"`
+	WarningsFile string            `json:"warnings_file"`
+}
+
+type outputFileIndex struct {
+	LaneID     string `json:"lane_id"`
+	MediaType  string `json:"media_type,omitempty"`
+	File       string `json:"file"`
+	ModuleKey  string `json:"module_key,omitempty"`
+	SchemaID   string `json:"schema_id,omitempty"`
+	SchemaName string `json:"schema_name,omitempty"`
+	SchemaVer  string `json:"schema_version,omitempty"`
+}
+
+type rejectedFile struct {
+	Rejected []contracts.RejectedOutput `json:"rejected"`
+}
+
+type warningsFile struct {
+	Warnings []contracts.Warning `json:"warnings"`
+}
+
+func logicalFiles(req contracts.OutputRequest) ([]contracts.OutputFile, error) {
+	outputs := cloneNormalizeOutputs(req.NormalizeOutputs)
+	sort.SliceStable(outputs, func(i, j int) bool {
+		return outputs[i].LaneID < outputs[j].LaneID
+	})
+
+	outputIndexes := make([]outputFileIndex, 0, len(outputs))
+	files := make([]contracts.OutputFile, 0, len(outputs)+4)
+	manifestFile, err := jsonFile("manifest.json", req.Manifest)
+	if err != nil {
+		return nil, err
+	}
+	files = append(files, manifestFile)
+
+	usedOutputFiles := make(map[string]string, len(outputs))
+	for _, output := range outputs {
+		name, err := outputFileName(output.LaneID)
+		if err != nil {
+			return nil, err
+		}
+		if existingLane, ok := usedOutputFiles[name]; ok {
+			return nil, encoderErrorf("lanes %q and %q produce duplicate output file %q", existingLane, output.LaneID, name)
+		}
+		usedOutputFiles[name] = output.LaneID
+		outputIndexes = append(outputIndexes, outputFileIndex{
+			LaneID:     output.LaneID,
+			MediaType:  output.Payload.MediaType,
+			File:       name,
+			ModuleKey:  output.NormalizerKey,
+			SchemaID:   output.Schema.ID,
+			SchemaName: output.Schema.Name,
+			SchemaVer:  output.Schema.Version,
+		})
+		file, err := rawOutputFile(name, output.Payload)
+		if err != nil {
+			return nil, err
+		}
+		files = append(files, file)
+	}
+
+	index := indexFile{
+		ManifestFile: "manifest.json",
+		OutputFiles:  outputIndexes,
+		RejectedFile: "rejected.json",
+		WarningsFile: "warnings.json",
+	}
+	indexOutput, err := jsonFile("index.json", index)
+	if err != nil {
+		return nil, err
+	}
+	rejectedOutput, err := jsonFile("rejected.json", rejectedFile{Rejected: cloneRejected(req.Rejected)})
+	if err != nil {
+		return nil, err
+	}
+	warningsOutput, err := jsonFile("warnings.json", warningsFile{Warnings: cloneWarnings(req.Warnings)})
+	if err != nil {
+		return nil, err
+	}
+	files = append(files, indexOutput, rejectedOutput, warningsOutput)
+	sort.Slice(files, func(i, j int) bool {
+		return files[i].Name < files[j].Name
+	})
+	return files, nil
+}
+
+func rawOutputFile(name string, payload contracts.RawPayload) (contracts.OutputFile, error) {
+	content := append([]byte(nil), payload.Content...)
+	if len(content) == 0 {
+		content = []byte("null")
+	}
+	mediaType := strings.TrimSpace(payload.MediaType)
+	if mediaType == "" {
+		mediaType = "application/octet-stream"
+	}
+	if !isJSONMediaType(mediaType) {
+		return contracts.OutputFile{}, encoderErrorf("normalized output %q has unsupported media type %q", name, mediaType)
+	}
+	var decoded any
+	if err := stdjson.Unmarshal(content, &decoded); err != nil {
+		return contracts.OutputFile{}, encoderErrorf("normalized output %q contains invalid JSON: %w", name, err)
+	}
+	pretty, err := marshalPretty(decoded)
+	if err != nil {
+		return contracts.OutputFile{}, err
+	}
+	return contracts.OutputFile{
+		Name:        name,
+		ContentType: mediaType,
+		Bytes:       pretty,
+	}, nil
+}
+
+func isJSONMediaType(mediaType string) bool {
+	base, _, err := mime.ParseMediaType(strings.TrimSpace(mediaType))
+	if err != nil {
+		base = strings.TrimSpace(mediaType)
+	}
+	return strings.EqualFold(base, contentTypeJSON)
+}
+
+func jsonFile(name string, value any) (contracts.OutputFile, error) {
+	data, err := marshalPretty(value)
+	if err != nil {
+		return contracts.OutputFile{}, encoderErrorf("encode %s: %w", name, err)
+	}
+	return contracts.OutputFile{
+		Name:        name,
+		ContentType: contentTypeJSON,
+		Bytes:       data,
+	}, nil
+}
+
+func marshalPretty(value any) ([]byte, error) {
+	data, err := stdjson.MarshalIndent(value, "", "  ")
+	if err != nil {
+		return nil, err
+	}
+	return append(data, '\n'), nil
+}
+
+func outputFileName(laneID string) (string, error) {
+	sanitized := safeOutputFileChar.ReplaceAllString(strings.TrimSpace(laneID), "_")
+	for strings.Contains(sanitized, "..") {
+		sanitized = strings.ReplaceAll(sanitized, "..", "__")
+	}
+	sanitized = strings.Trim(sanitized, "._")
+	if sanitized == "" {
+		return "", encoderErrorf("lane id %q cannot produce a safe file name", laneID)
+	}
+	return "lanes/" + sanitized + ".json", nil
+}
+
+func cloneNormalizeOutputs(outputs []contracts.NormalizeOutput) []contracts.NormalizeOutput {
+	if len(outputs) == 0 {
+		return nil
+	}
+	out := make([]contracts.NormalizeOutput, 0, len(outputs))
+	for _, output := range outputs {
+		output.Payload = cloneRawPayload(output.Payload)
+		out = append(out, output)
+	}
+	return out
+}
+
+func cloneRawPayload(payload contracts.RawPayload) contracts.RawPayload {
+	return contracts.RawPayload{
+		Content:   append([]byte(nil), payload.Content...),
+		MediaType: payload.MediaType,
+		Metadata:  cloneMetadata(payload.Metadata),
+		Warnings:  append([]contracts.Warning(nil), payload.Warnings...),
+	}
+}
+
+func cloneRejected(rejected []contracts.RejectedOutput) []contracts.RejectedOutput {
+	if len(rejected) == 0 {
+		return []contracts.RejectedOutput{}
+	}
+	return append([]contracts.RejectedOutput(nil), rejected...)
+}
+
+func cloneWarnings(warnings []contracts.Warning) []contracts.Warning {
+	if len(warnings) == 0 {
+		return []contracts.Warning{}
+	}
+	return append([]contracts.Warning(nil), warnings...)
+}
+
+func cloneMetadata(metadata map[string]any) map[string]any {
+	if len(metadata) == 0 {
+		return nil
+	}
+	out := make(map[string]any, len(metadata))
+	for key, value := range metadata {
+		out[key] = value
+	}
+	return out
+}
+
+func encoderErrorf(format string, args ...any) error {
+	return fmt.Errorf("json output encoder: "+format, args...)
+}

internal/modules/generic/output/json/encoder_test.go

@@ -0,0 +1,425 @@
+package json
+
+import (
+	"context"
+	stdjson "encoding/json"
+	"reflect"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestModuleSpecAndRegister(t *testing.T) {
+	want := pipeline.ModuleSpec{
+		Key:      Key,
+		Stage:    pipeline.StageOutput,
+		Requires: []string{"normalized"},
+		Provides: []string{"encoded"},
+	}
+	if got := ModuleSpec(); !reflect.DeepEqual(got, want) {
+		t.Fatalf("ModuleSpec() = %#v, want %#v", got, want)
+	}
+
+	registry := pipeline.NewOutputEncoderRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	spec, ok := registry.Spec(Key)
+	if !ok {
+		t.Fatalf("Spec(%q) ok = false, want true", Key)
+	}
+	if !reflect.DeepEqual(spec, want) {
+		t.Fatalf("registered spec = %#v, want %#v", spec, want)
+	}
+}
+
+func TestEncodeReturnsLogicalFilesForNormalizedOutputs(t *testing.T) {
+	req := contracts.OutputRequest{
+		Manifest: artifacts.RunManifest{RunID: "run-1", PipelineID: "pipeline-1"},
+		NormalizeOutputs: []contracts.NormalizeOutput{
+			normalizeOutput("spells", `{"spell_casts":[{"spell":"Cure Wounds"}]}`),
+			normalizeOutput("notes/items", `{"items":[{"name":"Torch"}]}`),
+		},
+		Rejected: []contracts.RejectedOutput{
+			{
+				Stage:         "extract",
+				LaneID:        "spells",
+				ModuleKey:     "dnd/spells",
+				ChunkID:       "chunk-1",
+				ChunkIndex:    1,
+				ReasonCode:    "invalid",
+				Message:       "not accepted",
+				AttemptCount:  1,
+				ValidatorName: "validator",
+			},
+		},
+		Warnings: []contracts.Warning{{ReasonCode: "warning", Message: "check source"}},
+	}
+
+	result, err := New().Encode(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+
+	wantNames := []string{
+		"index.json",
+		"lanes/notes_items.json",
+		"lanes/spells.json",
+		"manifest.json",
+		"rejected.json",
+		"warnings.json",
+	}
+	if got := outputFileNames(result.Files); !reflect.DeepEqual(got, wantNames) {
+		t.Fatalf("file names = %#v, want %#v", got, wantNames)
+	}
+
+	for _, file := range result.Files {
+		if !strings.HasSuffix(string(file.Bytes), "\n") {
+			t.Fatalf("%s does not end with newline: %q", file.Name, string(file.Bytes))
+		}
+		if file.ContentType == contentTypeJSON && !stdjson.Valid(file.Bytes) {
+			t.Fatalf("%s has invalid JSON: %s", file.Name, file.Bytes)
+		}
+	}
+
+	spells := decodeObject(t, fileBytes(t, result.Files, "lanes/spells.json"))
+	spellCasts := spells["spell_casts"].([]any)
+	if spellCasts[0].(map[string]any)["spell"] != "Cure Wounds" {
+		t.Fatalf("spells output = %#v, want raw normalized content", spells)
+	}
+
+	index := decodeObject(t, fileBytes(t, result.Files, "index.json"))
+	outputFiles := index["output_files"].([]any)
+	if len(outputFiles) != 2 {
+		t.Fatalf("len(index output_files) = %d, want 2", len(outputFiles))
+	}
+	firstIndex := outputFiles[0].(map[string]any)
+	secondIndex := outputFiles[1].(map[string]any)
+	if firstIndex["lane_id"] != "notes/items" || secondIndex["lane_id"] != "spells" {
+		t.Fatalf("output_files = %#v, want sorted by lane id", outputFiles)
+	}
+
+	rejected := decodeObject(t, fileBytes(t, result.Files, "rejected.json"))
+	if got := rejected["rejected"].([]any); len(got) != 1 {
+		t.Fatalf("rejected = %#v, want one rejected output", got)
+	}
+}
+
+func TestEncodeIncludesRejectedAndWarningsWhenEmpty(t *testing.T) {
+	result, err := New().Encode(context.Background(), contracts.OutputRequest{
+		Manifest: artifacts.RunManifest{RunID: "run-1"},
+	})
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+
+	rejected := decodeObject(t, fileBytes(t, result.Files, "rejected.json"))
+	if got := rejected["rejected"].([]any); len(got) != 0 {
+		t.Fatalf("rejected = %#v, want empty array", got)
+	}
+	warnings := decodeObject(t, fileBytes(t, result.Files, "warnings.json"))
+	if got := warnings["warnings"].([]any); len(got) != 0 {
+		t.Fatalf("warnings = %#v, want empty array", got)
+	}
+}
+
+func TestEncodePrettyPrintsJSON(t *testing.T) {
+	result, err := New().Encode(context.Background(), contracts.OutputRequest{
+		Manifest: artifacts.RunManifest{RunID: "run-1"},
+	})
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+
+	manifest := string(fileBytes(t, result.Files, "manifest.json"))
+	if !strings.Contains(manifest, "\n  \"run_id\": \"run-1\"\n") {
+		t.Fatalf("manifest JSON = %q, want two-space indentation", manifest)
+	}
+}
+
+func TestEncodeIncludesManifestReferences(t *testing.T) {
+	result, err := New().Encode(context.Background(), contracts.OutputRequest{
+		Manifest: artifacts.RunManifest{
+			RunID: "run-1",
+			References: []artifacts.ReferenceProvenance{
+				{
+					Stage:         "extract",
+					LaneID:        "events",
+					SlotName:      "roster",
+					OriginType:    "file",
+					OriginURI:     "file:///tmp/roster.txt",
+					Digest:        "sha256:reference",
+					MediaType:     "text/plain; charset=utf-8",
+					SizeBytes:     12,
+					BindingSource: "config",
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+
+	manifest := decodeObject(t, fileBytes(t, result.Files, "manifest.json"))
+	references := manifest["references"].([]any)
+	if len(references) != 1 {
+		t.Fatalf("references = %#v, want one entry", references)
+	}
+	reference := references[0].(map[string]any)
+	if reference["lane_id"] != "events" || reference["slot_name"] != "roster" || reference["digest"] != "sha256:reference" {
+		t.Fatalf("reference manifest = %#v, want lane slot digest", reference)
+	}
+	if _, ok := reference["content"]; ok {
+		t.Fatalf("reference manifest = %#v, want no content field", reference)
+	}
+}
+
+func TestEncodeIncludesManifestRawOutputProvenance(t *testing.T) {
+	result, err := New().Encode(context.Background(), contracts.OutputRequest{
+		Manifest: artifacts.RunManifest{
+			RunID: "run-1",
+			NormalizedOutputs: []artifacts.NormalizedOutputManifest{
+				{
+					LaneID:    "spells",
+					ModuleKey: "noop",
+					SourceID:  "source-1",
+					MediaType: contentTypeJSON,
+					Schema: artifacts.OutputSchemaProvenance{
+						ID:      "schema-id",
+						Name:    "schema-name",
+						Version: "v1",
+					},
+				},
+			},
+			RejectedOutputs: []artifacts.RejectedOutputManifest{
+				{
+					Stage:        "extract",
+					LaneID:       "spells",
+					ModuleKey:    "dnd/spells",
+					ChunkID:      "chunk-0",
+					ReasonCode:   "raw_output_rejected",
+					AttemptCount: 2,
+				},
+			},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+
+	manifest := decodeObject(t, fileBytes(t, result.Files, "manifest.json"))
+	normalized := manifest["normalized_outputs"].([]any)
+	if len(normalized) != 1 {
+		t.Fatalf("normalized_outputs = %#v, want one entry", normalized)
+	}
+	normalizedEntry := normalized[0].(map[string]any)
+	if normalizedEntry["lane_id"] != "spells" || normalizedEntry["media_type"] != contentTypeJSON {
+		t.Fatalf("normalized output manifest = %#v, want lane and media type", normalizedEntry)
+	}
+	rejected := manifest["rejected_outputs"].([]any)
+	if len(rejected) != 1 {
+		t.Fatalf("rejected_outputs = %#v, want one entry", rejected)
+	}
+	rejectedEntry := rejected[0].(map[string]any)
+	if rejectedEntry["attempt_count"] != float64(2) || rejectedEntry["chunk_id"] != "chunk-0" {
+		t.Fatalf("rejected output manifest = %#v, want attempt count and chunk", rejectedEntry)
+	}
+}
+
+func TestEncodeRejectsLaneIDWithoutSafeFileName(t *testing.T) {
+	_, err := New().Encode(context.Background(), contracts.OutputRequest{
+		NormalizeOutputs: []contracts.NormalizeOutput{normalizeOutput("///", `{"value":true}`)},
+	})
+	if err == nil {
+		t.Fatal("Encode() error = nil, want unsafe lane id error")
+	}
+	if !strings.Contains(err.Error(), "json output encoder") || !strings.Contains(err.Error(), "safe file name") {
+		t.Fatalf("Encode() error = %q, want safe file name context", err.Error())
+	}
+}
+
+func TestEncodeSanitizesParentPathSequences(t *testing.T) {
+	result, err := New().Encode(context.Background(), contracts.OutputRequest{
+		NormalizeOutputs: []contracts.NormalizeOutput{normalizeOutput("dnd..spell.", `{"value":true}`)},
+	})
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+
+	if got := outputFileNames(result.Files); !containsString(got, "lanes/dnd__spell.json") {
+		t.Fatalf("file names = %#v, want sanitized output filename", got)
+	}
+}
+
+func TestEncodeRejectsInvalidJSONAndUnsupportedMediaTypes(t *testing.T) {
+	tests := []struct {
+		name   string
+		output contracts.NormalizeOutput
+		want   string
+	}{
+		{
+			name:   "invalid JSON",
+			output: normalizeOutput("spells", `{"spell_casts":[`),
+			want:   "invalid JSON",
+		},
+		{
+			name: "unsupported media type",
+			output: func() contracts.NormalizeOutput {
+				output := normalizeOutput("spells", `{"spell_casts":[]}`)
+				output.Payload.MediaType = "text/plain"
+				return output
+			}(),
+			want: "unsupported media type",
+		},
+	}
+
+	for _, test := range tests {
+		t.Run(test.name, func(t *testing.T) {
+			_, err := New().Encode(context.Background(), contracts.OutputRequest{
+				NormalizeOutputs: []contracts.NormalizeOutput{test.output},
+			})
+			if err == nil {
+				t.Fatal("Encode() error = nil, want error")
+			}
+			if !strings.Contains(err.Error(), test.want) {
+				t.Fatalf("Encode() error = %q, want %q", err.Error(), test.want)
+			}
+		})
+	}
+}
+
+func TestEncodeRejectsSanitizedFilenameCollisions(t *testing.T) {
+	_, err := New().Encode(context.Background(), contracts.OutputRequest{
+		NormalizeOutputs: []contracts.NormalizeOutput{
+			normalizeOutput("a/b", `{"value":"slash"}`),
+			normalizeOutput("a?b", `{"value":"question"}`),
+		},
+	})
+	if err == nil {
+		t.Fatal("Encode() error = nil, want duplicate file error")
+	}
+	if !strings.Contains(err.Error(), "duplicate output file") {
+		t.Fatalf("Encode() error = %q, want duplicate file context", err.Error())
+	}
+}
+
+func TestEncodeDoesNotMutateInputs(t *testing.T) {
+	req := contracts.OutputRequest{
+		Manifest: artifacts.RunManifest{RunID: "run-1"},
+		NormalizeOutputs: []contracts.NormalizeOutput{
+			normalizeOutput("spells", `{"name":"original"}`),
+		},
+		Rejected: []contracts.RejectedOutput{
+			{Stage: "extract", LaneID: "spells", Message: "not accepted"},
+		},
+		Warnings: []contracts.Warning{{ReasonCode: "warning", Message: "message"}},
+	}
+	before := mustMarshal(t, req)
+
+	result, err := New().Encode(context.Background(), req)
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+	after := mustMarshal(t, req)
+	if before != after {
+		t.Fatalf("request mutated:\nbefore: %s\nafter:  %s", before, after)
+	}
+
+	req.NormalizeOutputs[0].Payload.Content[0] = '['
+	req.NormalizeOutputs[0].Payload.Metadata["name"] = "changed"
+	req.Rejected[0].Message = "changed"
+	req.Warnings[0].Message = "changed"
+
+	if !stdjson.Valid(fileBytes(t, result.Files, "lanes/spells.json")) {
+		t.Fatal("output changed after request mutation")
+	}
+	warnings := decodeObject(t, fileBytes(t, result.Files, "warnings.json"))
+	gotWarnings := warnings["warnings"].([]any)
+	if gotWarnings[0].(map[string]any)["message"] != "message" {
+		t.Fatalf("warnings output changed after request mutation: %#v", gotWarnings)
+	}
+}
+
+func TestOutputFilesDoNotContainWarnings(t *testing.T) {
+	result, err := New().Encode(context.Background(), contracts.OutputRequest{
+		NormalizeOutputs: []contracts.NormalizeOutput{normalizeOutput("spells", `{"spell":"Shield"}`)},
+		Warnings: []contracts.Warning{
+			{ReasonCode: "pipeline-warning", Message: "warning"},
+		},
+	})
+	if err != nil {
+		t.Fatalf("Encode() error = %v, want nil", err)
+	}
+
+	outputFile := decodeObject(t, fileBytes(t, result.Files, "lanes/spells.json"))
+	if _, ok := outputFile["warnings"]; ok {
+		t.Fatalf("output file contains warnings: %#v", outputFile)
+	}
+}
+
+func normalizeOutput(laneID string, content string) contracts.NormalizeOutput {
+	return contracts.NormalizeOutput{
+		LaneID:        laneID,
+		NormalizerKey: "noop",
+		SourceID:      "source-1",
+		Schema: contracts.ResponseSchema{
+			ID:      "schema-id",
+			Name:    "schema-name",
+			Version: "v1",
+		},
+		Payload: contracts.RawPayload{
+			Content:   []byte(content),
+			MediaType: contentTypeJSON,
+			Metadata:  map[string]any{"name": laneID},
+		},
+	}
+}
+
+func outputFileNames(files []contracts.OutputFile) []string {
+	names := make([]string, 0, len(files))
+	for _, file := range files {
+		names = append(names, file.Name)
+	}
+	return names
+}
+
+func containsString(values []string, want string) bool {
+	for _, value := range values {
+		if value == want {
+			return true
+		}
+	}
+	return false
+}
+
+func fileBytes(t *testing.T, files []contracts.OutputFile, name string) []byte {
+	t.Helper()
+	for _, file := range files {
+		if file.Name == name {
+			return file.Bytes
+		}
+	}
+	t.Fatalf("file %q not found in %#v", name, outputFileNames(files))
+	return nil
+}
+
+func decodeObject(t *testing.T, data []byte) map[string]any {
+	t.Helper()
+	var got map[string]any
+	if err := stdjson.Unmarshal(data, &got); err != nil {
+		t.Fatalf("Unmarshal() error = %v, want nil\n%s", err, data)
+	}
+	return got
+}
+
+func mustMarshal(t *testing.T, value any) string {
+	t.Helper()
+	data, err := stdjson.Marshal(value)
+	if err != nil {
+		t.Fatalf("Marshal() error = %v, want nil", err)
+	}
+	return string(data)
+}

internal/modules/generic/register/register.go

@@ -6,14 +6,14 @@ import (
 
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/llm"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/chunk/generic"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/merge/appendorder"
-	"gitea.maximumdirect.net/eric/notarius/internal/modules/normalize/noop"
-	jsonoutput "gitea.maximumdirect.net/eric/notarius/internal/modules/output/json"
-	alwaysaccept "gitea.maximumdirect.net/eric/notarius/internal/validators/generic/always_accept"
-	alwaysreject "gitea.maximumdirect.net/eric/notarius/internal/validators/generic/always_reject"
-	validjson "gitea.maximumdirect.net/eric/notarius/internal/validators/generic/valid_json"
-	validjsonschema "gitea.maximumdirect.net/eric/notarius/internal/validators/generic/valid_json_schema"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/generic/chunk/units"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/generic/merge/appendorder"
+	"gitea.maximumdirect.net/eric/notarius/internal/modules/generic/normalize/noop"
+	jsonoutput "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/output/json"
+	alwaysaccept "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/validate/always_accept"
+	alwaysreject "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/validate/always_reject"
+	validjson "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/validate/valid_json"
+	validjsonschema "gitea.maximumdirect.net/eric/notarius/internal/modules/generic/validate/valid_json_schema"
 )
 
 // Register adds all production domain-neutral modules and validators.
@@ -26,7 +26,7 @@ func Register(registries pipeline.Registries, assets *llm.AssetRegistry) error {
 		name     string
 		register func() error
 	}{
-		{name: "generic chunker", register: func() error { return generic.Register(registries.Chunkers) }},
+		{name: "generic chunker", register: func() error { return units.Register(registries.Chunkers) }},
 		{name: "appendorder merger", register: func() error { return appendorder.Register(registries.Mergers) }},
 		{name: "noop normalizer", register: func() error { return noop.Register(registries.Normalizers) }},
 		{name: "always accept validator", register: func() error { return alwaysaccept.Register(registries.Validators) }},

internal/modules/generic/validate/always_accept/validator.go

@@ -0,0 +1,43 @@
+package alwaysaccept
+
+import (
+	"context"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "generic/always_accept"
+
+var _ contracts.Validator = (*Validator)(nil)
+
+type Validator struct{}
+
+func New() *Validator {
+	return &Validator{}
+}
+
+func (v *Validator) Name() string {
+	return Key
+}
+
+func (v *Validator) ExecutionClass() contracts.ExecutionClass {
+	return contracts.ExecutionClassDeterministic
+}
+
+func (v *Validator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) {
+	return contracts.ValidationResult{Approved: true}, nil
+}
+
+func Spec() pipeline.ValidatorSpec {
+	return pipeline.ValidatorSpec{
+		Key:            Key,
+		ExecutionClass: contracts.ExecutionClassDeterministic,
+	}
+}
+
+func Register(registry *pipeline.ValidatorRegistry) error {
+	return registry.RegisterWithSpec(Spec(), func() (contracts.Validator, error) {
+		return New(), nil
+	})
+}

internal/modules/generic/validate/always_accept/validator_test.go

@@ -0,0 +1,40 @@
+package alwaysaccept
+
+import (
+	"context"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestValidatorApproves(t *testing.T) {
+	result, err := New().Validate(context.Background(), contracts.ValidationRequest{})
+	if err != nil {
+		t.Fatalf("Validate() error = %v, want nil", err)
+	}
+	if !result.Approved {
+		t.Fatalf("Approved = false, want true")
+	}
+	if result.ReasonCode != "" || result.Message != "" {
+		t.Fatalf("result = %#v, want approval without rejection details", result)
+	}
+}
+
+func TestSpecAndRegister(t *testing.T) {
+	if Spec().Key != Key || Spec().ExecutionClass != contracts.ExecutionClassDeterministic {
+		t.Fatalf("Spec() = %#v, want key and deterministic execution", Spec())
+	}
+
+	registry := pipeline.NewValidatorRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	validator, err := registry.Build(Key)
+	if err != nil {
+		t.Fatalf("Build(%q) error = %v, want nil", Key, err)
+	}
+	if validator.Name() != Key {
+		t.Fatalf("Name() = %q, want %q", validator.Name(), Key)
+	}
+}

internal/modules/generic/validate/always_reject/validator.go

@@ -0,0 +1,48 @@
+package alwaysreject
+
+import (
+	"context"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "generic/always_reject"
+const ReasonCode = "always_reject"
+
+var _ contracts.Validator = (*Validator)(nil)
+
+type Validator struct{}
+
+func New() *Validator {
+	return &Validator{}
+}
+
+func (v *Validator) Name() string {
+	return Key
+}
+
+func (v *Validator) ExecutionClass() contracts.ExecutionClass {
+	return contracts.ExecutionClassDeterministic
+}
+
+func (v *Validator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) {
+	return contracts.ValidationResult{
+		Approved:   false,
+		ReasonCode: ReasonCode,
+		Message:    "output rejected by always-reject validator",
+	}, nil
+}
+
+func Spec() pipeline.ValidatorSpec {
+	return pipeline.ValidatorSpec{
+		Key:            Key,
+		ExecutionClass: contracts.ExecutionClassDeterministic,
+	}
+}
+
+func Register(registry *pipeline.ValidatorRegistry) error {
+	return registry.RegisterWithSpec(Spec(), func() (contracts.Validator, error) {
+		return New(), nil
+	})
+}

internal/modules/generic/validate/always_reject/validator_test.go

@@ -0,0 +1,43 @@
+package alwaysreject
+
+import (
+	"context"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestValidatorRejects(t *testing.T) {
+	result, err := New().Validate(context.Background(), contracts.ValidationRequest{})
+	if err != nil {
+		t.Fatalf("Validate() error = %v, want nil", err)
+	}
+	if result.Approved {
+		t.Fatalf("Approved = true, want false")
+	}
+	if result.ReasonCode != ReasonCode {
+		t.Fatalf("ReasonCode = %q, want %q", result.ReasonCode, ReasonCode)
+	}
+	if result.Message == "" {
+		t.Fatal("Message = empty, want rejection message")
+	}
+}
+
+func TestSpecAndRegister(t *testing.T) {
+	if Spec().Key != Key || Spec().ExecutionClass != contracts.ExecutionClassDeterministic {
+		t.Fatalf("Spec() = %#v, want key and deterministic execution", Spec())
+	}
+
+	registry := pipeline.NewValidatorRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	validator, err := registry.Build(Key)
+	if err != nil {
+		t.Fatalf("Build(%q) error = %v, want nil", Key, err)
+	}
+	if validator.Name() != Key {
+		t.Fatalf("Name() = %q, want %q", validator.Name(), Key)
+	}
+}

internal/modules/generic/validate/valid_json/validator.go

@@ -0,0 +1,52 @@
+package validjson
+
+import (
+	"context"
+	"encoding/json"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "generic/valid_json"
+const ReasonCodeInvalidJSON = "invalid_json"
+
+var _ contracts.Validator = (*Validator)(nil)
+
+type Validator struct{}
+
+func New() *Validator {
+	return &Validator{}
+}
+
+func (v *Validator) Name() string {
+	return Key
+}
+
+func (v *Validator) ExecutionClass() contracts.ExecutionClass {
+	return contracts.ExecutionClassDeterministic
+}
+
+func (v *Validator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) {
+	if !json.Valid(req.Payload.Content) {
+		return contracts.ValidationResult{
+			Approved:   false,
+			ReasonCode: ReasonCodeInvalidJSON,
+			Message:    "payload is not valid JSON",
+		}, nil
+	}
+	return contracts.ValidationResult{Approved: true}, nil
+}
+
+func Spec() pipeline.ValidatorSpec {
+	return pipeline.ValidatorSpec{
+		Key:            Key,
+		ExecutionClass: contracts.ExecutionClassDeterministic,
+	}
+}
+
+func Register(registry *pipeline.ValidatorRegistry) error {
+	return registry.RegisterWithSpec(Spec(), func() (contracts.Validator, error) {
+		return New(), nil
+	})
+}

internal/modules/generic/validate/valid_json/validator_test.go

@@ -0,0 +1,66 @@
+package validjson
+
+import (
+	"context"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestValidatorAcceptsValidJSON(t *testing.T) {
+	tests := []string{
+		`{"value":true}`,
+		`[1,2,3]`,
+		`"value"`,
+	}
+	for _, payload := range tests {
+		result, err := New().Validate(context.Background(), requestWithPayload(payload))
+		if err != nil {
+			t.Fatalf("Validate(%s) error = %v, want nil", payload, err)
+		}
+		if !result.Approved {
+			t.Fatalf("Validate(%s) = %#v, want approved", payload, result)
+		}
+	}
+}
+
+func TestValidatorRejectsInvalidJSON(t *testing.T) {
+	result, err := New().Validate(context.Background(), requestWithPayload(`{"value":`))
+	if err != nil {
+		t.Fatalf("Validate() error = %v, want nil", err)
+	}
+	if result.Approved {
+		t.Fatalf("Approved = true, want false")
+	}
+	if result.ReasonCode != ReasonCodeInvalidJSON {
+		t.Fatalf("ReasonCode = %q, want %q", result.ReasonCode, ReasonCodeInvalidJSON)
+	}
+}
+
+func TestSpecAndRegister(t *testing.T) {
+	if Spec().Key != Key || Spec().ExecutionClass != contracts.ExecutionClassDeterministic {
+		t.Fatalf("Spec() = %#v, want key and deterministic execution", Spec())
+	}
+
+	registry := pipeline.NewValidatorRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	validator, err := registry.Build(Key)
+	if err != nil {
+		t.Fatalf("Build(%q) error = %v, want nil", Key, err)
+	}
+	if validator.Name() != Key {
+		t.Fatalf("Name() = %q, want %q", validator.Name(), Key)
+	}
+}
+
+func requestWithPayload(payload string) contracts.ValidationRequest {
+	return contracts.ValidationRequest{
+		Payload: contracts.RawPayload{
+			Content:   []byte(payload),
+			MediaType: "application/json",
+		},
+	}
+}

internal/modules/generic/validate/valid_json_schema/validator.go

@@ -0,0 +1,81 @@
+package validjsonschema
+
+import (
+	"bytes"
+	"context"
+	"fmt"
+
+	"github.com/santhosh-tekuri/jsonschema/v6"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+const Key = "generic/valid_json_schema"
+const ReasonCodeInvalidJSON = "invalid_json"
+const ReasonCodeSchemaInvalid = "json_schema_invalid"
+
+var _ contracts.Validator = (*Validator)(nil)
+
+type Validator struct{}
+
+func New() *Validator {
+	return &Validator{}
+}
+
+func (v *Validator) Name() string {
+	return Key
+}
+
+func (v *Validator) ExecutionClass() contracts.ExecutionClass {
+	return contracts.ExecutionClassDeterministic
+}
+
+func (v *Validator) Validate(ctx context.Context, req contracts.ValidationRequest) (contracts.ValidationResult, error) {
+	if len(req.Schema.JSONSchema) == 0 {
+		return contracts.ValidationResult{}, fmt.Errorf("response schema content is not available")
+	}
+
+	instance, err := jsonschema.UnmarshalJSON(bytes.NewReader(req.Payload.Content))
+	if err != nil {
+		return contracts.ValidationResult{
+			Approved:   false,
+			ReasonCode: ReasonCodeInvalidJSON,
+			Message:    "payload is not valid JSON",
+		}, nil
+	}
+
+	schemaDocument, err := jsonschema.UnmarshalJSON(bytes.NewReader(req.Schema.JSONSchema))
+	if err != nil {
+		return contracts.ValidationResult{}, fmt.Errorf("parse response schema: %w", err)
+	}
+	compiler := jsonschema.NewCompiler()
+	if err := compiler.AddResource("schema.json", schemaDocument); err != nil {
+		return contracts.ValidationResult{}, fmt.Errorf("load response schema: %w", err)
+	}
+	schema, err := compiler.Compile("schema.json")
+	if err != nil {
+		return contracts.ValidationResult{}, fmt.Errorf("compile response schema: %w", err)
+	}
+	if err := schema.Validate(instance); err != nil {
+		return contracts.ValidationResult{
+			Approved:   false,
+			ReasonCode: ReasonCodeSchemaInvalid,
+			Message:    "payload does not conform to response schema",
+		}, nil
+	}
+	return contracts.ValidationResult{Approved: true}, nil
+}
+
+func Spec() pipeline.ValidatorSpec {
+	return pipeline.ValidatorSpec{
+		Key:            Key,
+		ExecutionClass: contracts.ExecutionClassDeterministic,
+	}
+}
+
+func Register(registry *pipeline.ValidatorRegistry) error {
+	return registry.RegisterWithSpec(Spec(), func() (contracts.Validator, error) {
+		return New(), nil
+	})
+}

internal/modules/generic/validate/valid_json_schema/validator_test.go

@@ -0,0 +1,111 @@
+package validjsonschema
+
+import (
+	"context"
+	"strings"
+	"testing"
+
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
+	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
+)
+
+func TestValidatorAcceptsSchemaConformantJSON(t *testing.T) {
+	result, err := New().Validate(context.Background(), requestWithSchema(`{"name":"Aria"}`, objectSchema()))
+	if err != nil {
+		t.Fatalf("Validate() error = %v, want nil", err)
+	}
+	if !result.Approved {
+		t.Fatalf("Validate() = %#v, want approved", result)
+	}
+}
+
+func TestValidatorRejectsInvalidPayloadJSON(t *testing.T) {
+	result, err := New().Validate(context.Background(), requestWithSchema(`{"name":`, objectSchema()))
+	if err != nil {
+		t.Fatalf("Validate() error = %v, want nil", err)
+	}
+	if result.Approved {
+		t.Fatalf("Approved = true, want false")
+	}
+	if result.ReasonCode != ReasonCodeInvalidJSON {
+		t.Fatalf("ReasonCode = %q, want %q", result.ReasonCode, ReasonCodeInvalidJSON)
+	}
+}
+
+func TestValidatorRejectsSchemaNonConformance(t *testing.T) {
+	result, err := New().Validate(context.Background(), requestWithSchema(`{"name":3}`, objectSchema()))
+	if err != nil {
+		t.Fatalf("Validate() error = %v, want nil", err)
+	}
+	if result.Approved {
+		t.Fatalf("Approved = true, want false")
+	}
+	if result.ReasonCode != ReasonCodeSchemaInvalid {
+		t.Fatalf("ReasonCode = %q, want %q", result.ReasonCode, ReasonCodeSchemaInvalid)
+	}
+}
+
+func TestValidatorErrorsWhenSchemaContentMissing(t *testing.T) {
+	_, err := New().Validate(context.Background(), requestWithSchema(`{"name":"Aria"}`, nil))
+	if err == nil {
+		t.Fatal("Validate() error = nil, want missing schema content error")
+	}
+	if !strings.Contains(err.Error(), "schema content") {
+		t.Fatalf("Validate() error = %q, want schema content context", err.Error())
+	}
+}
+
+func TestValidatorErrorsWhenSchemaContentIsMalformed(t *testing.T) {
+	_, err := New().Validate(context.Background(), requestWithSchema(`{"name":"Aria"}`, []byte(`{"type":`)))
+	if err == nil {
+		t.Fatal("Validate() error = nil, want malformed schema error")
+	}
+	if !strings.Contains(err.Error(), "parse response schema") {
+		t.Fatalf("Validate() error = %q, want parse schema context", err.Error())
+	}
+}
+
+func TestSpecAndRegister(t *testing.T) {
+	if Spec().Key != Key || Spec().ExecutionClass != contracts.ExecutionClassDeterministic {
+		t.Fatalf("Spec() = %#v, want key and deterministic execution", Spec())
+	}
+
+	registry := pipeline.NewValidatorRegistry()
+	if err := Register(registry); err != nil {
+		t.Fatalf("Register() error = %v, want nil", err)
+	}
+	validator, err := registry.Build(Key)
+	if err != nil {
+		t.Fatalf("Build(%q) error = %v, want nil", Key, err)
+	}
+	if validator.Name() != Key {
+		t.Fatalf("Name() = %q, want %q", validator.Name(), Key)
+	}
+}
+
+func requestWithSchema(payload string, schema []byte) contracts.ValidationRequest {
+	return contracts.ValidationRequest{
+		Schema: contracts.ResponseSchema{
+			ID:         "test.schema",
+			Name:       "test_schema",
+			Version:    "v1",
+			JSONSchema: append([]byte(nil), schema...),
+		},
+		Payload: contracts.RawPayload{
+			Content:   []byte(payload),
+			MediaType: "application/json",
+		},
+	}
+}
+
+func objectSchema() []byte {
+	return []byte(`{
+		"$schema": "https://json-schema.org/draft/2020-12/schema",
+		"type": "object",
+		"required": ["name"],
+		"properties": {
+			"name": {"type": "string"}
+		},
+		"additionalProperties": false
+	}`)
+}