eric/notarius
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Update raw output files and manifests

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-07 19:27:28 +00:00
parent aa14faa3cb
commit 7c95791e94
10 changed files with 253 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/integrations/dnd-spell-artifacts.md
View File

@@ -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
{

23
docs/integrations/json-output.md
View File

@@ -20,7 +20,7 @@ The `json` output module writes:
- `index.json`
- `manifest.json`
- `outputs/<lane-id>.json`, one file per normalized raw lane output
- `lanes/<lane-id>.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/<sanitized-lane-id>.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/<sanitized-lane-id>.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
{

3
docs/internal/modules.md
View File

@@ -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:

4
docs/operations.md
View File

@@ -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/<lane-id>.json`: normalized raw output payloads, one file per lane.
For the current D&D spell extractor, this includes `outputs/spells.json`.
- `lanes/<lane-id>.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.

4
internal/cli/run_test.go
View File

@@ -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)
}

67
internal/core/artifacts/artifacts.go
View File

@@ -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 {

53
internal/framework/pipeline/runner.go
View File

@@ -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

13
internal/framework/pipeline/runner_test.go
View File

@@ -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))
}

34
internal/modules/output/json/encoder.go
View File

@@ -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 {

101
internal/modules/output/json/encoder_test.go
View File

@@ -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)
}