Implement raw module output contracts

2026-07-07 18:58:23 +00:00
parent 9e3f8809b3
commit c05ecb58d8
33 changed files with 1166 additions and 1780 deletions
--- a/internal/modules/merge/appendorder/merger.go
+++ b/internal/modules/merge/appendorder/merger.go
@@ -5,8 +5,6 @@ import (
 	"encoding/json"
 	"fmt"

-	"gitea.maximumdirect.net/eric/notarius/internal/core/artifacts"
-	"gitea.maximumdirect.net/eric/notarius/internal/core/source"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
 )
@@ -36,11 +34,34 @@ func (m *Merger) Merge(ctx context.Context, req contracts.MergeRequest) (contrac
 		return contracts.MergeResult{}, mergerErrorf("context error before merge: %w", err)
 	}

-	var candidates []artifacts.ArtifactCandidate
-	for _, chunkArtifacts := range req.ChunkArtifacts {
-		candidates = append(candidates, cloneCandidates(chunkArtifacts.Candidates)...)
+	if len(req.ExtractOutputs) == 1 {
+		payload := cloneRawPayload(req.ExtractOutputs[0].Payload)
+		return contracts.MergeResult{
+			Output: contracts.MergeOutput{
+				LaneID:    req.LaneID,
+				MergerKey: Key,
+				SourceID:  req.ExtractOutputs[0].SourceID,
+				Schema:    req.ExtractOutputs[0].Schema,
+				Payload:   payload,
+			},
+		}, nil
 	}
-	return contracts.MergeResult{Candidates: candidates}, nil
+
+	content, err := orderedContent(req.ExtractOutputs)
+	if err != nil {
+		return contracts.MergeResult{}, err
+	}
+	return contracts.MergeResult{
+		Output: contracts.MergeOutput{
+			LaneID:    req.LaneID,
+			MergerKey: Key,
+			SourceID:  sourceID(req.ExtractOutputs),
+			Payload: contracts.RawPayload{
+				Content:   content,
+				MediaType: "application/json",
+			},
+		},
+	}, nil
 }

 func ModuleSpec() pipeline.ModuleSpec {
@@ -57,27 +78,45 @@ func Register(registry *pipeline.MergerRegistry) error {
 	})
 }

-func cloneCandidates(candidates []artifacts.ArtifactCandidate) []artifacts.ArtifactCandidate {
-	if len(candidates) == 0 {
-		return nil
+func orderedContent(outputs []contracts.ExtractOutput) ([]byte, error) {
+	items := make([]map[string]any, 0, len(outputs))
+	for _, output := range outputs {
+		item := map[string]any{
+			"chunk_id":    output.ChunkID,
+			"chunk_index": output.ChunkIndex,
+			"media_type":  output.Payload.MediaType,
+		}
+		if json.Valid(output.Payload.Content) {
+			item["content"] = json.RawMessage(append([]byte(nil), output.Payload.Content...))
+		} else {
+			item["content"] = string(output.Payload.Content)
+		}
+		items = append(items, item)
 	}
-
-	out := make([]artifacts.ArtifactCandidate, 0, len(candidates))
-	for _, candidate := range candidates {
-		out = append(out, cloneCandidate(candidate))
+	content, err := json.Marshal(struct {
+		Outputs []map[string]any `json:"outputs"`
+	}{Outputs: items})
+	if err != nil {
+		return nil, mergerErrorf("encode merged output: %w", err)
 	}
-	return out
+	return content, nil
 }

-func cloneCandidate(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:      cloneMetadata(candidate.Metadata),
+func sourceID(outputs []contracts.ExtractOutput) string {
+	for _, output := range outputs {
+		if output.SourceID != "" {
+			return output.SourceID
+		}
+	}
+	return ""
+}
+
+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...),
 	}
 }

--- a/internal/modules/merge/appendorder/merger_test.go
+++ b/internal/modules/merge/appendorder/merger_test.go
@@ -6,8 +6,6 @@ import (
 	"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"
 	"gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline"
 )
@@ -35,113 +33,102 @@ func TestModuleSpecAndRegister(t *testing.T) {
 	}
 }

-func TestMergePreservesChunkAndCandidateOrder(t *testing.T) {
+func TestMergePassesThroughSingleExtractOutput(t *testing.T) {
+	input := extractOutput("chunk-0", 0, `{"name":"original"}`)
+
 	result, err := New().Merge(context.Background(), contracts.MergeRequest{
-		ChunkArtifacts: []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")},
-			},
-		},
+		LaneID:         "events",
+		ExtractOutputs: []contracts.ExtractOutput{input},
 	})
 	if err != nil {
 		t.Fatalf("Merge() error = %v, want nil", err)
 	}

-	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)
+	if result.Output.LaneID != "events" || result.Output.MergerKey != Key {
+		t.Fatalf("output provenance = %#v, want lane and merger", result.Output)
 	}
-	if len(result.Warnings) != 0 {
-		t.Fatalf("Warnings = %#v, want none", result.Warnings)
+	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 TestMergeDefensivelyCopiesCandidates(t *testing.T) {
-	input := []contracts.ChunkArtifacts{
-		{
-			Chunk:      sourceChunk(0),
-			Candidates: []artifacts.ArtifactCandidate{candidate(1, "original")},
-		},
-	}
+func TestMergeDefensivelyCopiesRawPayload(t *testing.T) {
+	input := extractOutput("chunk-0", 0, `{"name":"original"}`)

-	result, err := New().Merge(context.Background(), contracts.MergeRequest{ChunkArtifacts: input})
+	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 len(result.Candidates) != 1 {
-		t.Fatalf("len(Candidates) = %d, want 1", len(result.Candidates))
+
+	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 TestMergeWrapsMultipleOutputsInChunkOrder(t *testing.T) {
+	result, err := New().Merge(context.Background(), contracts.MergeRequest{
+		LaneID: "events",
+		ExtractOutputs: []contracts.ExtractOutput{
+			extractOutput("chunk-0", 0, `{"name":"first"}`),
+			extractOutput("chunk-1", 1, `{"name":"second"}`),
+		},
+	})
+	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)
 	}

-	input[0].Candidates[0].Index = 99
-	input[0].Candidates[0].Payload[0] = '['
-	input[0].Candidates[0].SourceRefs[0].StartUnitID = 99
-	input[0].Candidates[0].Metadata["name"] = "changed"
-
-	got := result.Candidates[0]
-	if got.Index != 1 {
-		t.Fatalf("Index = %d, want 1", got.Index)
+	var decoded struct {
+		Outputs []struct {
+			ChunkID    string          `json:"chunk_id"`
+			ChunkIndex int             `json:"chunk_index"`
+			Content    json.RawMessage `json:"content"`
+		} `json:"outputs"`
 	}
-	if string(got.Payload) != `{"name":"original"}` {
-		t.Fatalf("Payload = %s, want original payload", got.Payload)
+	if err := json.Unmarshal(result.Output.Payload.Content, &decoded); err != nil {
+		t.Fatalf("Unmarshal() error = %v, want nil", err)
 	}
-	if got.SourceRefs[0].StartUnitID != 1 {
-		t.Fatalf("SourceRefs = %#v, want original source ref", got.SourceRefs)
-	}
-	if got.Metadata["name"] != "original" {
-		t.Fatalf("Metadata = %#v, want original metadata", got.Metadata)
+	if len(decoded.Outputs) != 2 || decoded.Outputs[0].ChunkID != "chunk-0" || decoded.Outputs[1].ChunkID != "chunk-1" {
+		t.Fatalf("outputs = %#v, want chunk order", decoded.Outputs)
 	}
 }

 func TestMergeHandlesEmptyInput(t *testing.T) {
-	result, err := New().Merge(context.Background(), contracts.MergeRequest{})
+	result, err := New().Merge(context.Background(), contracts.MergeRequest{LaneID: "events"})
 	if err != nil {
 		t.Fatalf("Merge() error = %v, want nil", err)
 	}
-	if len(result.Candidates) != 0 {
-		t.Fatalf("len(Candidates) = %d, want 0", len(result.Candidates))
-	}
-	if len(result.Warnings) != 0 {
-		t.Fatalf("Warnings = %#v, want none", result.Warnings)
+	if string(result.Output.Payload.Content) != `{"outputs":[]}` {
+		t.Fatalf("content = %s, want empty outputs", result.Output.Payload.Content)
 	}
 }

-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: 1, EndUnitID: 1},
-		},
-		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: 1, Kind: "unit", Text: "Source unit."},
+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},
 		},
 	}
 }