From aa14faa3cb71e16bdcd4e997a23c1ac105c39d8b Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 7 Jul 2026 19:23:35 +0000 Subject: [PATCH] Migrate production modules to raw outputs --- docs/integrations/dnd-spell-artifacts.md | 28 ++-- docs/internal/modules.md | 7 +- internal/cli/run_test.go | 10 +- .../modules/extract/dnd/spells/extractor.go | 19 +-- .../extract/dnd/spells/extractor_test.go | 27 ++-- .../modules/extract/dnd/spells/runner_test.go | 19 +-- internal/modules/merge/appendorder/merger.go | 130 +++++++++++++++--- .../modules/merge/appendorder/merger_test.go | 85 ++++++++++-- 8 files changed, 247 insertions(+), 78 deletions(-) diff --git a/docs/integrations/dnd-spell-artifacts.md b/docs/integrations/dnd-spell-artifacts.md index baa4700..3fdeb0c 100644 --- a/docs/integrations/dnd-spell-artifacts.md +++ b/docs/integrations/dnd-spell-artifacts.md @@ -15,9 +15,9 @@ This document is the durable raw output contract for the implemented The extractor requires source chunks and transcript source capability. It returns the structured LLM response as raw JSON. The default `appendorder` -merger passes a single chunk output through and wraps multiple chunk outputs in -an ordered `outputs` array. The default `noop` normalizer passes the merge -output through unchanged. +merger passes a single chunk output through and concatenates multiple +`spell_casts` arrays in chunk order. The default `noop` normalizer passes the +merge output through unchanged. ## Output Shape @@ -45,18 +45,24 @@ For a single chunk, `outputs/spells.json` has this shape: `spell_casts` must be present. It may be empty when no spell casts are found. -For multiple chunks with the default merger, the lane output has this shape: +For multiple chunks with the default merger, the lane output keeps the same +top-level shape and concatenates `spell_casts` in chunk order: ```json { - "outputs": [ + "spell_casts": [ { - "chunk_id": "chunk-000001", - "chunk_index": 0, - "media_type": "application/json", - "content": { - "spell_casts": [] - } + "caster": "Aria", + "spell": "Cure Wounds", + "effect": "heals an injured ally", + "narrative_description": "Aria raises her holy symbol and casts Cure Wounds.", + "source_refs": [ + { + "source_id": "session-alpha", + "start_unit_id": 1, + "end_unit_id": 1 + } + ] } ] } diff --git a/docs/internal/modules.md b/docs/internal/modules.md index c95eb34..671c3ca 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -191,8 +191,11 @@ only; spell-cast artifacts must still be grounded in the source transcript. Package: `internal/modules/merge/appendorder` The `appendorder` merger preserves chunk order for raw extract outputs. A -single extract output is passed through as the merge output. Multiple extract -outputs are wrapped in one JSON payload under `outputs`. +single JSON extract output is passed through as the merge output. Multiple JSON +object outputs with one common top-level array field are merged by concatenating +that array field in chunk order. Other valid JSON shapes are merged as a JSON +array of decoded values in chunk order. Non-JSON media types and invalid JSON +are rejected. Provides: diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 51b3ffb..5fbfe39 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -2254,11 +2254,11 @@ func TestExampleFixtureFailureCoverage(t *testing.T) { wantStderr: "completion unavailable", }, { - name: "malformed LLM response", - args: []string{"run", "dnd-session", "--config", configPath, "--input", inputPath}, - factory: fakeLLMFactory(newMalformedRunLLMClient(), nil), - wantCode: 1, - wantStderr: "spell_casts", + name: "malformed LLM response carried as raw output", + args: []string{"run", "dnd-session", "--config", configPath, "--input", inputPath}, + factory: fakeLLMFactory(newMalformedRunLLMClient(), nil), + wantCode: 0, + wantOutputStatus: "approved", }, { name: "invalid source reference raw output", diff --git a/internal/modules/extract/dnd/spells/extractor.go b/internal/modules/extract/dnd/spells/extractor.go index aa21ba1..135b599 100644 --- a/internal/modules/extract/dnd/spells/extractor.go +++ b/internal/modules/extract/dnd/spells/extractor.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "fmt" + "strings" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" @@ -90,22 +91,24 @@ func (e *Extractor) Extract(ctx context.Context, req contracts.ExtractionRequest } var response extractionResponse - if _, err := req.LLMClient.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ + completion, err := req.LLMClient.CompleteStructured(ctx, contracts.StructuredCompletionRequest{ StageName: Key, PromptID: PromptID, PromptVersion: SchemaVersion, ProfileID: req.LLMProfile, SessionID: req.SessionID, Inputs: dnd.PromptInputs(req.SourceInput, req.References), - }, &response); err != nil { + }, &response) + if err != nil { return contracts.ExtractionResult{}, extractorErrorf("complete structured output: %w", err) } - if response.SpellCasts == nil { - return contracts.ExtractionResult{}, extractorErrorf("malformed structured output: spell_casts must be present") - } - content, err := json.Marshal(response) - if err != nil { - return contracts.ExtractionResult{}, extractorErrorf("marshal raw output: %w", err) + content := append([]byte(nil), completion.Content...) + if len(strings.TrimSpace(string(content))) == 0 { + var err error + content, err = json.Marshal(response) + if err != nil { + return contracts.ExtractionResult{}, extractorErrorf("marshal raw output: %w", err) + } } return contracts.ExtractionResult{ Output: contracts.ExtractOutput{ diff --git a/internal/modules/extract/dnd/spells/extractor_test.go b/internal/modules/extract/dnd/spells/extractor_test.go index 50efa14..4f2a0d6 100644 --- a/internal/modules/extract/dnd/spells/extractor_test.go +++ b/internal/modules/extract/dnd/spells/extractor_test.go @@ -24,6 +24,7 @@ func TestExtractReturnsRawOutputFromStructuredResponse(t *testing.T) { }, }, }, + content: []byte(`{"spell_casts":[{"caster":" Aria ","spell":" Cure Wounds ","effect":" Heals an injured ally. ","narrative_description":" Aria restores the fighter after the fight. ","source_refs":[{"source_id":"session-alpha","start_unit_id":1,"end_unit_id":2}]}],"raw_marker":true}`), } result, err := New().Extract(context.Background(), extractionRequestWithClient(client)) @@ -58,6 +59,9 @@ func TestExtractReturnsRawOutputFromStructuredResponse(t *testing.T) { if result.Output.Schema.ID != ResponseSchemaID || result.Output.Schema.Name != ResponseSchemaName || result.Output.Schema.Version != SchemaVersion { t.Fatalf("schema = %#v, want response schema provenance", result.Output.Schema) } + if got := string(result.Output.Payload.Content); got != string(client.content) { + t.Fatalf("content = %q, want exact raw completion content", got) + } var payload extractionResponse if err := json.Unmarshal(result.Output.Payload.Content, &payload); err != nil { @@ -179,15 +183,15 @@ func TestExtractReturnsRawOutputForEmptyResponse(t *testing.T) { } } -func TestExtractRejectsMissingSpellCasts(t *testing.T) { +func TestExtractCarriesMalformedStructuredContentAsRawOutput(t *testing.T) { client := &fakeSpellsLLMClient{response: extractionResponse{}} - _, err := New().Extract(context.Background(), extractionRequestWithClient(client)) - if err == nil { - t.Fatal("Extract() error = nil, want malformed output error") + result, err := New().Extract(context.Background(), extractionRequestWithClient(client)) + if err != nil { + t.Fatalf("Extract() error = %v, want nil", err) } - if !strings.Contains(err.Error(), "dnd spells") || !strings.Contains(err.Error(), "spell_casts") { - t.Fatalf("Extract() error = %q, want spell_casts context", err.Error()) + if string(result.Output.Payload.Content) != `{"spell_casts":null}` { + t.Fatalf("content = %s, want raw structured output", result.Output.Payload.Content) } } @@ -329,6 +333,7 @@ func emptyChunkRequest(req contracts.ExtractionRequest) contracts.ExtractionRequ type fakeSpellsLLMClient struct { response extractionResponse + content []byte err error requests []contracts.StructuredCompletionRequest } @@ -344,9 +349,13 @@ func (client *fakeSpellsLLMClient) CompleteStructured(ctx context.Context, req c return contracts.StructuredCompletionResponse{}, errors.New("unexpected output target") } *target = client.response - content, err := json.Marshal(client.response) - if err != nil { - return contracts.StructuredCompletionResponse{}, err + content := append([]byte(nil), client.content...) + if len(content) == 0 { + var err error + content, err = json.Marshal(client.response) + if err != nil { + return contracts.StructuredCompletionResponse{}, err + } } return contracts.StructuredCompletionResponse{Content: content}, nil } diff --git a/internal/modules/extract/dnd/spells/runner_test.go b/internal/modules/extract/dnd/spells/runner_test.go index 4832dab..fb9d1a2 100644 --- a/internal/modules/extract/dnd/spells/runner_test.go +++ b/internal/modules/extract/dnd/spells/runner_test.go @@ -279,7 +279,7 @@ func dndSpellsReferenceSet(party string, glossary string) contracts.ReferenceSet return contracts.ReferenceSet{Slots: slots} } -func TestRunnerFailsWhenDNDSpellsExtractorReturnsMalformedOutput(t *testing.T) { +func TestRunnerCarriesMalformedDNDSpellsExtractorOutput(t *testing.T) { raw := readDNDSpellsFixture(t) resolved := resolveDNDSpellsPipeline(t) llmClient := &fakeSpellsLLMClient{response: extractionResponse{}} @@ -289,16 +289,17 @@ func TestRunnerFailsWhenDNDSpellsExtractorReturnsMalformedOutput(t *testing.T) { RawInput: raw, LLMClient: llmClient, }) - if err == nil { - t.Fatal("Run() error = nil, want malformed extraction error") + if err != nil { + t.Fatalf("Run() error = %v, want nil", err) } - if !strings.Contains(err.Error(), "extract lane") || - !strings.Contains(err.Error(), "dnd spells") || - !strings.Contains(err.Error(), "spell_casts") { - t.Fatalf("Run() error = %q, want D&D spells extraction context", err.Error()) + if len(output.NormalizeOutputs) != 1 { + t.Fatalf("len(NormalizeOutputs) = %d, want raw output", len(output.NormalizeOutputs)) } - if output.Manifest.ValidationStatus != "failed" { - t.Fatalf("ValidationStatus = %q, want failed", output.Manifest.ValidationStatus) + if string(output.NormalizeOutputs[0].Payload.Content) != `{"spell_casts":null}` { + t.Fatalf("content = %s, want raw structured output", output.NormalizeOutputs[0].Payload.Content) + } + if output.Manifest.ValidationStatus != "approved" { + t.Fatalf("ValidationStatus = %q, want approved", output.Manifest.ValidationStatus) } } diff --git a/internal/modules/merge/appendorder/merger.go b/internal/modules/merge/appendorder/merger.go index 8bba6cc..431d4ef 100644 --- a/internal/modules/merge/appendorder/merger.go +++ b/internal/modules/merge/appendorder/merger.go @@ -4,6 +4,9 @@ import ( "context" "encoding/json" "fmt" + "mime" + "sort" + "strings" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" @@ -34,20 +37,24 @@ func (m *Merger) Merge(ctx context.Context, req contracts.MergeRequest) (contrac return contracts.MergeResult{}, mergerErrorf("context error before merge: %w", err) } - if len(req.ExtractOutputs) == 1 { - payload := cloneRawPayload(req.ExtractOutputs[0].Payload) + 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: req.ExtractOutputs[0].SourceID, - Schema: req.ExtractOutputs[0].Schema, + SourceID: outputs[0].SourceID, + Schema: outputs[0].Schema, Payload: payload, }, }, nil } - content, err := orderedContent(req.ExtractOutputs) + content, err := mergedContent(outputs) if err != nil { return contracts.MergeResult{}, err } @@ -55,7 +62,8 @@ func (m *Merger) Merge(ctx context.Context, req contracts.MergeRequest) (contrac Output: contracts.MergeOutput{ LaneID: req.LaneID, MergerKey: Key, - SourceID: sourceID(req.ExtractOutputs), + SourceID: sourceID(outputs), + Schema: commonSchema(outputs), Payload: contracts.RawPayload{ Content: content, MediaType: "application/json", @@ -78,30 +86,94 @@ func Register(registry *pipeline.MergerRegistry) error { }) } -func orderedContent(outputs []contracts.ExtractOutput) ([]byte, error) { - items := make([]map[string]any, 0, len(outputs)) +func orderedOutputs(outputs []contracts.ExtractOutput) ([]contracts.ExtractOutput, error) { + ordered := make([]contracts.ExtractOutput, 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 !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) { - item["content"] = json.RawMessage(append([]byte(nil), output.Payload.Content...)) - } else { - item["content"] = string(output.Payload.Content) + if !json.Valid(output.Payload.Content) { + return nil, mergerErrorf("extract output for chunk %q contains invalid JSON", output.ChunkID) } - items = append(items, item) + ordered = append(ordered, cloneExtractOutput(output)) } - content, err := json.Marshal(struct { - Outputs []map[string]any `json:"outputs"` - }{Outputs: items}) + 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 != "" { @@ -111,6 +183,24 @@ func sourceID(outputs []contracts.ExtractOutput) string { 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 output.Schema != schema { + return contracts.ResponseSchema{} + } + } + return schema +} + +func cloneExtractOutput(output contracts.ExtractOutput) contracts.ExtractOutput { + output.Payload = cloneRawPayload(output.Payload) + return output +} + func cloneRawPayload(payload contracts.RawPayload) contracts.RawPayload { return contracts.RawPayload{ Content: append([]byte(nil), payload.Content...), diff --git a/internal/modules/merge/appendorder/merger_test.go b/internal/modules/merge/appendorder/merger_test.go index 7cd5f06..f298d4b 100644 --- a/internal/modules/merge/appendorder/merger_test.go +++ b/internal/modules/merge/appendorder/merger_test.go @@ -4,6 +4,7 @@ import ( "context" "encoding/json" "reflect" + "strings" "testing" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" @@ -77,12 +78,12 @@ func TestMergeDefensivelyCopiesRawPayload(t *testing.T) { } } -func TestMergeWrapsMultipleOutputsInChunkOrder(t *testing.T) { +func TestMergeConcatenatesCommonTopLevelArrayFieldInChunkOrder(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"}`), + extractOutput("chunk-1", 1, `{"events":[{"name":"second"}]}`), + extractOutput("chunk-0", 0, `{"events":[{"name":"first"}]}`), }, }) if err != nil { @@ -93,27 +94,83 @@ func TestMergeWrapsMultipleOutputsInChunkOrder(t *testing.T) { } var decoded struct { - Outputs []struct { - ChunkID string `json:"chunk_id"` - ChunkIndex int `json:"chunk_index"` - Content json.RawMessage `json:"content"` - } `json:"outputs"` + 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.Outputs) != 2 || decoded.Outputs[0].ChunkID != "chunk-0" || decoded.Outputs[1].ChunkID != "chunk-1" { - t.Fatalf("outputs = %#v, want chunk order", decoded.Outputs) + 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 TestMergeHandlesEmptyInput(t *testing.T) { - result, err := New().Merge(context.Background(), contracts.MergeRequest{LaneID: "events"}) +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) } - if string(result.Output.Payload.Content) != `{"outputs":[]}` { - t.Fatalf("content = %s, want empty outputs", result.Output.Payload.Content) + + 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) + } + }) } }