diff --git a/docs/integrations/dnd-spell-artifacts.md b/docs/integrations/dnd-spell-artifacts.md index 3fdeb0c..7bdc573 100644 --- a/docs/integrations/dnd-spell-artifacts.md +++ b/docs/integrations/dnd-spell-artifacts.md @@ -21,7 +21,7 @@ merge output through unchanged. ## Output Shape -For a single chunk, `outputs/spells.json` has this shape: +For a single chunk, `lanes/spells.json` has this shape: ```json { diff --git a/docs/integrations/json-output.md b/docs/integrations/json-output.md index 4233462..5e3cff0 100644 --- a/docs/integrations/json-output.md +++ b/docs/integrations/json-output.md @@ -20,7 +20,7 @@ The `json` output module writes: - `index.json` - `manifest.json` -- `outputs/.json`, one file per normalized raw lane output +- `lanes/.json`, one file per normalized raw lane output - `rejected.json` - `warnings.json` @@ -37,7 +37,7 @@ Shape: { "lane_id": "spells", "media_type": "application/json", - "file": "outputs/spells.json", + "file": "lanes/spells.json", "module_key": "noop", "schema_id": "notarius.dnd.spells", "schema_name": "notarius_dnd_spells_v1", @@ -49,7 +49,7 @@ Shape: } ``` -`output_files` is in normalized output order. Output file names are produced by +`output_files` is sorted by lane ID. Output file names are produced by sanitizing the lane ID: - characters outside `A-Z`, `a-z`, `0-9`, `.`, `_`, and `-` become `_`; @@ -103,14 +103,21 @@ references. `validation_status` is `approved` when no raw outputs were rejected and `rejected` when one or more raw outputs were rejected. +`normalized_outputs` summarizes each normalized lane output without embedding +payload bytes. Entries include lane ID, normalizer module key, source ID, media +type, and response schema provenance where available. + +`rejected_outputs` summarizes rejected module outputs without embedding raw +payload bytes. Entries include stage, lane, module, chunk, validator or reason, +message, attempt count, and optional diagnostic artifact path. + ## Output Payload Files -Each normalized raw output is written to `outputs/.json`. -For `application/json` payloads, the file contains the raw JSON payload -pretty-printed. Other media types are written as raw bytes with the media type -reported in `index.json`. +Each normalized raw output is written to `lanes/.json`. +The JSON output encoder accepts only `application/json` normalized outputs. The +file contains the raw JSON payload pretty-printed. -For the current D&D spell extractor, `outputs/spells.json` has this shape: +For the current D&D spell extractor, `lanes/spells.json` has this shape: ```json { diff --git a/docs/internal/modules.md b/docs/internal/modules.md index 671c3ca..6741eb8 100644 --- a/docs/internal/modules.md +++ b/docs/internal/modules.md @@ -221,7 +221,8 @@ Package: `internal/modules/output/json` The `json` output encoder converts normalized raw outputs, rejected raw outputs, warnings, and the run manifest into logical JSON output files. It writes one -payload file per lane under `outputs/` and sanitizes lane IDs for file names. +payload file per lane under `lanes/` and sanitizes lane IDs for file names. +Normalized output payloads must be valid `application/json`. Requires: diff --git a/docs/operations.md b/docs/operations.md index c94058f..5629a74 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -37,8 +37,8 @@ The `json` output module writes these files: - `manifest.json`: run manifest with resolved pipeline provenance, top-level module metadata, module keys, reference provenance, validation status, and timing. -- `outputs/.json`: normalized raw output payloads, one file per lane. - For the current D&D spell extractor, this includes `outputs/spells.json`. +- `lanes/.json`: normalized raw JSON output payloads, one file per + lane. For the current D&D spell extractor, this includes `lanes/spells.json`. - `rejected.json`: rejected raw output records. - `warnings.json`: warnings reported by pipeline modules or the output encoder. diff --git a/internal/cli/run_test.go b/internal/cli/run_test.go index 5fbfe39..d63bf58 100644 --- a/internal/cli/run_test.go +++ b/internal/cli/run_test.go @@ -1732,8 +1732,8 @@ func TestRunPipelineWritesDurableOutputFiles(t *testing.T) { runOutputDir := onlyChildDir(t, outputDir) for _, name := range []string{ "index.json", + "lanes/spells.json", "manifest.json", - "outputs/spells.json", "rejected.json", "warnings.json", } { @@ -2086,7 +2086,7 @@ func TestExampleFixtureRunWritesExpectedJSON(t *testing.T) { SourceRefs []source.SourceRef `json:"source_refs"` } `json:"spell_casts"` } - readJSONFile(t, filepath.Join(runOutputDir, "outputs", "spells.json"), &spellOutput) + readJSONFile(t, filepath.Join(runOutputDir, "lanes", "spells.json"), &spellOutput) if len(spellOutput.SpellCasts) != 1 { t.Fatalf("spell output = %#v, want one spell cast", spellOutput) } diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index 46a54ff..0dd3f1b 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -60,26 +60,55 @@ type ReferenceProvenance struct { BindingSource string `json:"binding_source,omitempty"` } +type OutputSchemaProvenance struct { + ID string `json:"id,omitempty"` + Name string `json:"name,omitempty"` + Version string `json:"version,omitempty"` +} + +type NormalizedOutputManifest struct { + LaneID string `json:"lane_id"` + ModuleKey string `json:"module_key,omitempty"` + SourceID string `json:"source_id,omitempty"` + MediaType string `json:"media_type,omitempty"` + Schema OutputSchemaProvenance `json:"schema,omitempty"` +} + +type RejectedOutputManifest struct { + Stage string `json:"stage"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + ChunkID string `json:"chunk_id,omitempty"` + ChunkIndex int `json:"chunk_index,omitempty"` + ValidatorName string `json:"validator_name,omitempty"` + ReasonCode string `json:"reason_code,omitempty"` + Message string `json:"message,omitempty"` + AttemptCount int `json:"attempt_count,omitempty"` + DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` +} + type RunManifest struct { - RunID string `json:"run_id,omitempty"` - PipelineID string `json:"pipeline_id,omitempty"` - PipelineDigest string `json:"pipeline_digest,omitempty"` - InputModule string `json:"input_module,omitempty"` - Chunker string `json:"chunker,omitempty"` - SourceDigests []string `json:"source_digests,omitempty"` - Extractors []string `json:"extractors,omitempty"` - Merger string `json:"merger,omitempty"` - Normalizer string `json:"normalizer,omitempty"` - OutputEncoder string `json:"output_encoder,omitempty"` - ModuleMetadata map[string]map[string]any `json:"module_metadata,omitempty"` - ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` - References []ReferenceProvenance `json:"references,omitempty"` - LLMProfiles []LLMProfileManifest `json:"llm_profiles,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` - SchemaVersion string `json:"schema_version,omitempty"` - ValidationStatus string `json:"validation_status,omitempty"` - StartedAt *time.Time `json:"started_at,omitempty"` - CompletedAt *time.Time `json:"completed_at,omitempty"` + RunID string `json:"run_id,omitempty"` + PipelineID string `json:"pipeline_id,omitempty"` + PipelineDigest string `json:"pipeline_digest,omitempty"` + InputModule string `json:"input_module,omitempty"` + Chunker string `json:"chunker,omitempty"` + SourceDigests []string `json:"source_digests,omitempty"` + Extractors []string `json:"extractors,omitempty"` + Merger string `json:"merger,omitempty"` + Normalizer string `json:"normalizer,omitempty"` + OutputEncoder string `json:"output_encoder,omitempty"` + ModuleMetadata map[string]map[string]any `json:"module_metadata,omitempty"` + ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` + References []ReferenceProvenance `json:"references,omitempty"` + NormalizedOutputs []NormalizedOutputManifest `json:"normalized_outputs,omitempty"` + RejectedOutputs []RejectedOutputManifest `json:"rejected_outputs,omitempty"` + LLMProfiles []LLMProfileManifest `json:"llm_profiles,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + SchemaVersion string `json:"schema_version,omitempty"` + ValidationStatus string `json:"validation_status,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` } func ArtifactFromCandidate(candidate ArtifactCandidate) Artifact { diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 8e148ea..95a671a 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -157,6 +157,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } else { output.Manifest.ValidationStatus = "approved" } + populateRawOutputManifest(&output) output.Manifest.CompletedAt = timePtr(time.Now().UTC()) encoder, err := r.registries.Outputs.Build(input.Pipeline.Output.Module) @@ -636,12 +637,64 @@ func manifestFromPipeline(input RunInput) artifacts.RunManifest { func failOutput(output RunOutput) RunOutput { if output.Manifest.PipelineID != "" { + populateRawOutputManifest(&output) output.Manifest.ValidationStatus = "failed" output.Manifest.CompletedAt = timePtr(time.Now().UTC()) } return output } +func populateRawOutputManifest(output *RunOutput) { + if output == nil { + return + } + output.Manifest.NormalizedOutputs = normalizedOutputManifests(output.NormalizeOutputs) + output.Manifest.RejectedOutputs = rejectedOutputManifests(output.Rejected) +} + +func normalizedOutputManifests(outputs []contracts.NormalizeOutput) []artifacts.NormalizedOutputManifest { + if len(outputs) == 0 { + return nil + } + manifests := make([]artifacts.NormalizedOutputManifest, 0, len(outputs)) + for _, output := range outputs { + manifests = append(manifests, artifacts.NormalizedOutputManifest{ + LaneID: output.LaneID, + ModuleKey: output.NormalizerKey, + SourceID: output.SourceID, + MediaType: output.Payload.MediaType, + Schema: artifacts.OutputSchemaProvenance{ + ID: output.Schema.ID, + Name: output.Schema.Name, + Version: output.Schema.Version, + }, + }) + } + return manifests +} + +func rejectedOutputManifests(rejected []contracts.RejectedOutput) []artifacts.RejectedOutputManifest { + if len(rejected) == 0 { + return nil + } + manifests := make([]artifacts.RejectedOutputManifest, 0, len(rejected)) + for _, output := range rejected { + manifests = append(manifests, artifacts.RejectedOutputManifest{ + Stage: output.Stage, + LaneID: output.LaneID, + ModuleKey: output.ModuleKey, + ChunkID: output.ChunkID, + ChunkIndex: output.ChunkIndex, + ValidatorName: output.ValidatorName, + ReasonCode: output.ReasonCode, + Message: output.Message, + AttemptCount: output.AttemptCount, + DiagnosticArtifactPath: output.DiagnosticArtifactPath, + }) + } + return manifests +} + func setLaneManifestMetadata(output *RunOutput, laneID string, modules ...any) { if output == nil { return diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index e753b1f..6fecc02 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -1034,6 +1034,12 @@ func TestRunStopsRetryAfterConfiguredAttemptsAndRecordsAttemptCount(t *testing.T if len(modules.extractors["extract-alpha"].requests) != 4 { t.Fatalf("extract requests = %d, want two attempts per chunk", len(modules.extractors["extract-alpha"].requests)) } + if len(output.Manifest.RejectedOutputs) != 2 { + t.Fatalf("manifest rejected outputs = %#v, want rejected records", output.Manifest.RejectedOutputs) + } + if output.Manifest.RejectedOutputs[0].AttemptCount != 2 || output.Manifest.RejectedOutputs[0].ChunkID != "chunk-0" { + t.Fatalf("manifest rejected output = %#v, want final attempt count and chunk provenance", output.Manifest.RejectedOutputs[0]) + } } func TestRunContextCancellationStopsRetries(t *testing.T) { @@ -1292,6 +1298,13 @@ func TestRunManifestIncludesPipelineAndLaneDetails(t *testing.T) { if manifest.ValidationStatus != "approved" { t.Fatalf("ValidationStatus = %q, want approved", manifest.ValidationStatus) } + if len(manifest.NormalizedOutputs) != 1 { + t.Fatalf("NormalizedOutputs = %#v, want one raw output manifest", manifest.NormalizedOutputs) + } + normalized := manifest.NormalizedOutputs[0] + if normalized.LaneID != "alpha" || normalized.ModuleKey != "normalize" || normalized.MediaType != "application/json" || normalized.Schema.ID != "runner.raw" { + t.Fatalf("normalized output manifest = %#v, want lane/module/media/schema provenance", normalized) + } if len(manifest.ArtifactLanes) != 1 { t.Fatalf("len(ArtifactLanes) = %d, want 1", len(manifest.ArtifactLanes)) } diff --git a/internal/modules/output/json/encoder.go b/internal/modules/output/json/encoder.go index a83c4a0..95a420e 100644 --- a/internal/modules/output/json/encoder.go +++ b/internal/modules/output/json/encoder.go @@ -4,6 +4,7 @@ import ( "context" stdjson "encoding/json" "fmt" + "mime" "regexp" "sort" "strings" @@ -158,27 +159,36 @@ func rawOutputFile(name string, payload contracts.RawPayload) (contracts.OutputF if len(content) == 0 { content = []byte("null") } - if payload.MediaType == contentTypeJSON && stdjson.Valid(content) { - var decoded any - if err := stdjson.Unmarshal(content, &decoded); err == nil { - pretty, err := marshalPretty(decoded) - if err != nil { - return contracts.OutputFile{}, err - } - content = pretty - } - } 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: append([]byte(nil), content...), + 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 { @@ -208,7 +218,7 @@ func outputFileName(laneID string) (string, error) { if sanitized == "" { return "", encoderErrorf("lane id %q cannot produce a safe file name", laneID) } - return "outputs/" + sanitized + ".json", nil + return "lanes/" + sanitized + ".json", nil } func cloneNormalizeOutputs(outputs []contracts.NormalizeOutput) []contracts.NormalizeOutput { diff --git a/internal/modules/output/json/encoder_test.go b/internal/modules/output/json/encoder_test.go index ce06e2b..dd37f21 100644 --- a/internal/modules/output/json/encoder_test.go +++ b/internal/modules/output/json/encoder_test.go @@ -66,9 +66,9 @@ func TestEncodeReturnsLogicalFilesForNormalizedOutputs(t *testing.T) { wantNames := []string{ "index.json", + "lanes/notes_items.json", + "lanes/spells.json", "manifest.json", - "outputs/notes_items.json", - "outputs/spells.json", "rejected.json", "warnings.json", } @@ -85,7 +85,7 @@ func TestEncodeReturnsLogicalFilesForNormalizedOutputs(t *testing.T) { } } - spells := decodeObject(t, fileBytes(t, result.Files, "outputs/spells.json")) + 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) @@ -177,6 +177,58 @@ func TestEncodeIncludesManifestReferences(t *testing.T) { } } +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}`)}, @@ -197,11 +249,48 @@ func TestEncodeSanitizesParentPathSequences(t *testing.T) { t.Fatalf("Encode() error = %v, want nil", err) } - if got := outputFileNames(result.Files); !containsString(got, "outputs/dnd__spell.json") { + 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{ @@ -244,7 +333,7 @@ func TestEncodeDoesNotMutateInputs(t *testing.T) { req.Rejected[0].Message = "changed" req.Warnings[0].Message = "changed" - if !stdjson.Valid(fileBytes(t, result.Files, "outputs/spells.json")) { + 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")) @@ -265,7 +354,7 @@ func TestOutputFilesDoNotContainWarnings(t *testing.T) { t.Fatalf("Encode() error = %v, want nil", err) } - outputFile := decodeObject(t, fileBytes(t, result.Files, "outputs/spells.json")) + outputFile := decodeObject(t, fileBytes(t, result.Files, "lanes/spells.json")) if _, ok := outputFile["warnings"]; ok { t.Fatalf("output file contains warnings: %#v", outputFile) }