diff --git a/docs/integrations/dnd-spell-artifacts.md b/docs/integrations/dnd-spell-artifacts.md index 3bac4b8..775d14e 100644 --- a/docs/integrations/dnd-spell-artifacts.md +++ b/docs/integrations/dnd-spell-artifacts.md @@ -1,7 +1,7 @@ -# D&D Spell Raw Output +# D&D Spell Artifact -This document is the durable raw output contract for the production D&D spell -extractor. Selectable extractor keys are cataloged in +This document is the durable serialized artifact contract for the production +D&D spell extractor. Selectable extractor keys are cataloged in [Configuration](../config.md#implemented-production-modules). ## Identity diff --git a/docs/integrations/json-output.md b/docs/integrations/json-output.md index 9c6cd1b..572351b 100644 --- a/docs/integrations/json-output.md +++ b/docs/integrations/json-output.md @@ -15,7 +15,7 @@ The encoder writes: - `index.json` - `manifest.json` -- `lanes/.json`, one file per normalized raw lane output +- `lanes/.json`, one file per normalized serialized artifact - `rejected.json` - `warnings.json` @@ -112,8 +112,8 @@ Reference `stage` is `chunk`, `extract`, `merge`, or `normalize`. `lane_id` is omitted for chunk references and present for extract, merge, and normalize references. -`validation_status` is `approved` when no raw outputs were rejected and -`rejected` when one or more raw outputs were rejected. +`validation_status` is `approved` when no outputs were rejected and `rejected` +when one or more outputs were rejected. `validator_chains` records the resolved validator chain for each validation point. Entries include stage, lane ID when applicable, module key, and validators @@ -131,12 +131,13 @@ message, attempt count, and optional diagnostic artifact path. ## Output Payload Files -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. +Each normalized serialized artifact is written to +`lanes/.json`. The JSON output encoder is domain-neutral and +accepts only artifacts whose codec media type is `application/json`. The file +contains the codec-owned JSON bytes pretty-printed. The schema of each lane payload is owned by that artifact contract. For the -current D&D spell lane, see [D&D Spell Raw Output](dnd-spell-artifacts.md). +current D&D spell lane, see [D&D Spell Artifact](dnd-spell-artifacts.md). ## `rejected.json` @@ -148,7 +149,7 @@ Shape: } ``` -When raw output validation rejects an output, each entry contains `stage` and +When output validation rejects an output, each entry contains `stage` and `message`. It includes `lane_id`, `module_key`, `chunk_id`, `chunk_index`, `validator_name`, `reason_code`, `attempt_count`, and `diagnostic_artifact_path` when applicable. diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index dc6d96d..9b41878 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -127,7 +127,8 @@ debug boundaries retain that reference, and the canonical source digest covers it deterministically. Chunks use the same source model and carry one canonical reference spanning the first selected unit through the last. -`pipeline.RunOutput` carries the run manifest, accepted normalized results, +`pipeline.RunOutput` carries the run manifest, accepted normalized serialized +artifacts with lane and normalizer provenance, rejected results, warnings, checkpoint events, and logical files returned by the output encoder. The CLI owns diagnostics and durable filesystem writes after the runner returns. @@ -199,15 +200,20 @@ The runner depends on recorder and loader interfaces, using no-op implementations when collaborators are absent. Each checkpointed workflow boundary records a running, succeeded, or failed transition. Reuse decisions are consulted in workflow order and accepted payloads are cloned before -entering the normal handoff path. The typed D&D executor uses explicit -migration-only codec adapters to read and write the existing raw artifact -checkpoint envelopes. Dependency fingerprints connect later checkpoints to the -exact accepted results on which they depend. +entering the normal handoff path. Typed extract, merge, and normalize +checkpoints store codec bytes with artifact kind, schema ID and version, exact +schema digest, and media type. Reuse compares that identity with the prepared +codec and decodes through the codec; missing identity, mismatches, corrupt +bytes, and decode failures become explicit reuse misses and execute the step +normally. Dependency fingerprints and debug content digests use the same stable +codec bytes that cross those boundaries. Debug instrumentation wraps run, stage, attempt, validator, and structured LLM boundaries. Context scopes associate nested LLM calls with the module or validator attempt that made them. Debug-write failures are framework errors; -debug data is never used as a checkpoint source. +debug data is never used as a checkpoint source. Typed artifact debug envelopes +are domain-neutral, redact sensitive metadata and bytes through the common +debug policy, and record codec identity plus schema and content digests. Checkpoint identity, physical layout, reuse behavior, and debug artifact handling are operator contracts in [Operations](../operations.md). Serialization @@ -224,7 +230,7 @@ Module metadata providers may add non-secret singleton or lane-scoped metadata. Execution errors include stage, module, lane, or validator context. Once a manifest exists, a failing run returns it with failed status and completion -time. Successful status reflects whether any raw result was rejected. The +time. Successful status reflects whether any result was rejected. The durable manifest and logical file schemas are defined in the [JSON output contract](../integrations/json-output.md). diff --git a/docs/operations.md b/docs/operations.md index a59e66f..fff7ded 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -91,6 +91,12 @@ digests match the current invocation. Changes to input bytes, the resolved pipeline, selected lanes, the runtime LLM profile override, or bound reference content invalidate reuse. +Typed artifact checkpoints additionally record codec-owned bytes, artifact +kind, schema ID and version, exact schema digest, and media type. A missing or +mismatched codec identity, or bytes the current codec cannot decode, is reported +as a checkpoint reuse miss. The affected operation executes normally and, when +checkpoint writing is enabled, replaces the incompatible checkpoint. + Current checkpoint manifests use workspace schema `notarius.workspace.v2`. Manifests written with `notarius.workspace.v1` are incompatible because their chunk provenance has an older shape. On the first explicit resume after an @@ -126,7 +132,9 @@ the attempt `llm_calls` array. Prompt content is written inline in the prompt artifact. The response metadata and body use the paired files described above; the body is pretty-printed JSON when possible and raw text otherwise. Debug artifacts may contain source material, reference material, prompt inputs, model -outputs, and other sensitive data. API keys are not written, and obvious +outputs, and other sensitive data. Typed artifact envelopes include +domain-neutral codec identity, redacted metadata and content, and digests of +the stable codec bytes. API keys are not written, and obvious credential-shaped values and sensitive map keys are redacted, but debug directories should still be protected as sensitive local state. diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index a4d8a64..e919629 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -114,6 +114,28 @@ func (l *WorkspaceLoader) Extract(laneID string, moduleKey string, dependencies }, reusedDecision() } +func (l *WorkspaceLoader) ArtifactExtract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ArtifactExtractCheckpoint, pipeline.CheckpointDecision) { + var manifest coreworkspace.ExtractLaneManifest + if d := l.readJSON(laneManifestPath("extract", laneID), &manifest); !d.Reused { + return pipeline.ArtifactExtractCheckpoint{}, d + } + if d := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageExtract, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded, coreworkspace.StatusSucceededWithRejections); !d.Reused { + return pipeline.ArtifactExtractCheckpoint{}, d + } + var payload artifactExtractEnvelope + if d := l.readJSON(lanePayloadPath("extract", laneID, "outputs.json"), &payload); !d.Reused { + return pipeline.ArtifactExtractCheckpoint{}, d + } + outputs, err := artifactCheckpointOutputs(payload.Outputs) + if err != nil { + return pipeline.ArtifactExtractCheckpoint{}, invalidDecision("extract artifact checkpoint payload is invalid: %v", err) + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) { + return pipeline.ArtifactExtractCheckpoint{}, invalidDecision("extract artifact checkpoint output digests do not match payload") + } + return pipeline.ArtifactExtractCheckpoint{Outputs: outputs, Rejected: cloneRejectedOutputs(payload.Rejected), Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() +} + func (l *WorkspaceLoader) Merge(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) { var manifest coreworkspace.MergeLaneManifest if decision := l.readJSON(laneManifestPath("merge", laneID), &manifest); !decision.Reused { @@ -136,6 +158,28 @@ func (l *WorkspaceLoader) Merge(laneID string, moduleKey string, dependencies [] return pipeline.MergeCheckpoint{Output: output, Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } +func (l *WorkspaceLoader) ArtifactMerge(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ArtifactMergeCheckpoint, pipeline.CheckpointDecision) { + var manifest coreworkspace.MergeLaneManifest + if d := l.readJSON(laneManifestPath("merge", laneID), &manifest); !d.Reused { + return pipeline.ArtifactMergeCheckpoint{}, d + } + if d := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageMerge, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !d.Reused { + return pipeline.ArtifactMergeCheckpoint{}, d + } + var payload artifactSingleEnvelope + if d := l.readJSON(lanePayloadPath("merge", laneID, "output.json"), &payload); !d.Reused { + return pipeline.ArtifactMergeCheckpoint{}, d + } + values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output}) + if err != nil { + return pipeline.ArtifactMergeCheckpoint{}, invalidDecision("merge artifact checkpoint payload is invalid: %v", err) + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { + return pipeline.ArtifactMergeCheckpoint{}, invalidDecision("merge artifact checkpoint output digest does not match payload") + } + return pipeline.ArtifactMergeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() +} + func (l *WorkspaceLoader) Normalize(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) { var manifest coreworkspace.NormalizeLaneManifest if decision := l.readJSON(laneManifestPath("normalize", laneID), &manifest); !decision.Reused { @@ -158,6 +202,46 @@ func (l *WorkspaceLoader) Normalize(laneID string, moduleKey string, dependencie return pipeline.NormalizeCheckpoint{Output: output, Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } +func (l *WorkspaceLoader) ArtifactNormalize(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ArtifactNormalizeCheckpoint, pipeline.CheckpointDecision) { + var manifest coreworkspace.NormalizeLaneManifest + if d := l.readJSON(laneManifestPath("normalize", laneID), &manifest); !d.Reused { + return pipeline.ArtifactNormalizeCheckpoint{}, d + } + if d := l.validateLaneManifest(manifest.StageManifest, coreworkspace.StageNormalize, laneID, moduleKey, dependencies, coreworkspace.StatusSucceeded); !d.Reused { + return pipeline.ArtifactNormalizeCheckpoint{}, d + } + var payload artifactSingleEnvelope + if d := l.readJSON(lanePayloadPath("normalize", laneID, "output.json"), &payload); !d.Reused { + return pipeline.ArtifactNormalizeCheckpoint{}, d + } + values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output}) + if err != nil { + return pipeline.ArtifactNormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint payload is invalid: %v", err) + } + if !fingerprintsEqual(coreworkspaceToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { + return pipeline.ArtifactNormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint output digest does not match payload") + } + return pipeline.ArtifactNormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() +} + +func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline.ArtifactCheckpointOutput, error) { + if len(values) == 0 { + return nil, nil + } + out := make([]pipeline.ArtifactCheckpointOutput, 0, len(values)) + for _, v := range values { + content, err := contentFromEnvelope(v.Content) + if err != nil { + return nil, err + } + if strings.TrimSpace(string(v.Kind)) == "" || strings.TrimSpace(v.Schema.ID) == "" || strings.TrimSpace(v.Schema.Version) == "" || strings.TrimSpace(v.SchemaDigest) == "" { + return nil, fmt.Errorf("artifact codec identity is incomplete") + } + out = append(out, pipeline.ArtifactCheckpointOutput{LaneID: v.LaneID, ModuleKey: v.ModuleKey, SourceID: v.SourceID, ChunkID: v.ChunkID, ChunkIndex: v.ChunkIndex, ChunkRef: v.ChunkRef, SchemaDigest: v.SchemaDigest, Artifact: contracts.SerializedArtifact{Kind: v.Kind, Schema: v.Schema, MediaType: v.Content.MediaType, Content: content, Metadata: cloneMetadata(v.Content.Metadata)}}) + } + return out, nil +} + func (l *WorkspaceLoader) readJSON(name string, out any) pipeline.CheckpointDecision { if !l.Enabled() { return pipeline.CheckpointDecision{Reason: "checkpoint loading disabled"} diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go index ae2e661..9ff378f 100644 --- a/internal/framework/checkpoint/recorder.go +++ b/internal/framework/checkpoint/recorder.go @@ -140,6 +140,19 @@ func (r *WorkspaceRecorder) ExtractSucceeded(laneID string, moduleKey string, de }) } +func (r *WorkspaceRecorder) ArtifactExtractSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, outputs []pipeline.ArtifactCheckpointOutput, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + payload := artifactExtractEnvelope{Outputs: artifactCheckpointEnvelopes(outputs), Rejected: cloneRejectedOutputs(rejected), Warnings: cloneWarnings(warnings)} + if err := r.writePayload(lanePayloadPath("extract", laneID, "outputs.json"), payload); err != nil { + return err + } + manifest := r.laneManifest(coreworkspace.StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies) + manifest.OutputDigests = workspaceFingerprints(artifactOutputDigests(outputs)) + manifest.ValidationStatus = validationStatusString(warnings, rejected) + manifest.Rejections = rejectionSummaries(rejected) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("extract", laneID), coreworkspace.ExtractLaneManifest{StageManifest: manifest, ChunkCount: len(outputs) + len(rejected), OutputCount: len(outputs)}) +} + func (r *WorkspaceRecorder) ExtractFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { manifest := r.laneManifest(coreworkspace.StageExtract, coreworkspace.StatusFailed, laneID, moduleKey, dependencies) manifest.CompletedAt = timePtr(r.timestamp()) @@ -168,6 +181,17 @@ func (r *WorkspaceRecorder) MergeSucceeded(laneID string, moduleKey string, depe }) } +func (r *WorkspaceRecorder) ArtifactMergeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.ArtifactCheckpointOutput, warnings []contracts.Warning) error { + if err := r.writePayload(lanePayloadPath("merge", laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { + return err + } + manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) + manifest.OutputDigests = workspaceFingerprints(artifactOutputDigests([]pipeline.ArtifactCheckpointOutput{output})) + manifest.ValidationStatus = validationStatusString(warnings, nil) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("merge", laneID), coreworkspace.MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) +} + func (r *WorkspaceRecorder) MergeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { manifest := r.laneManifest(coreworkspace.StageMerge, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" @@ -201,6 +225,17 @@ func (r *WorkspaceRecorder) NormalizeSucceeded(laneID string, moduleKey string, return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } +func (r *WorkspaceRecorder) ArtifactNormalizeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.ArtifactCheckpointOutput, warnings []contracts.Warning) error { + if err := r.writePayload(lanePayloadPath("normalize", laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { + return err + } + manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceeded, laneID, moduleKey, dependencies) + manifest.OutputDigests = workspaceFingerprints(artifactOutputDigests([]pipeline.ArtifactCheckpointOutput{output})) + manifest.ValidationStatus = validationStatusString(warnings, nil) + manifest.CompletedAt = timePtr(r.timestamp()) + return r.writeManifest(laneManifestPath("normalize", laneID), coreworkspace.NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) +} + func (r *WorkspaceRecorder) NormalizeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { manifest := r.laneManifest(coreworkspace.StageNormalize, coreworkspace.StatusSucceededWithRejections, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" @@ -323,6 +358,51 @@ type binaryEnvelope struct { Warnings []contracts.Warning `json:"warnings,omitempty"` } +type artifactCheckpointEnvelope struct { + LaneID string `json:"lane_id"` + ModuleKey string `json:"module_key"` + SourceID string `json:"source_id,omitempty"` + ChunkID string `json:"chunk_id,omitempty"` + ChunkIndex int `json:"chunk_index,omitempty"` + ChunkRef source.SourceRef `json:"chunk_ref,omitempty"` + Kind contracts.ArtifactKind `json:"artifact_kind"` + Schema contracts.ArtifactSchema `json:"schema"` + SchemaDigest string `json:"schema_digest"` + Content binaryEnvelope `json:"content"` +} +type artifactExtractEnvelope struct { + Outputs []artifactCheckpointEnvelope `json:"outputs"` + Rejected []contracts.RejectedOutput `json:"rejected,omitempty"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} +type artifactSingleEnvelope struct { + Output artifactCheckpointEnvelope `json:"output"` + Warnings []contracts.Warning `json:"warnings,omitempty"` +} + +func artifactCheckpointEnvelopeFromOutput(output pipeline.ArtifactCheckpointOutput) artifactCheckpointEnvelope { + schema := contracts.CloneArtifactSchema(output.Artifact.Schema) + schema.JSONSchema = nil + return artifactCheckpointEnvelope{LaneID: output.LaneID, ModuleKey: output.ModuleKey, SourceID: output.SourceID, ChunkID: output.ChunkID, ChunkIndex: output.ChunkIndex, ChunkRef: output.ChunkRef, Kind: output.Artifact.Kind, Schema: schema, SchemaDigest: output.SchemaDigest, Content: binaryEnvelopeFromContent(output.Artifact.Content, output.Artifact.MediaType, output.Artifact.Metadata, nil)} +} +func artifactCheckpointEnvelopes(outputs []pipeline.ArtifactCheckpointOutput) []artifactCheckpointEnvelope { + if len(outputs) == 0 { + return nil + } + out := make([]artifactCheckpointEnvelope, 0, len(outputs)) + for _, v := range outputs { + out = append(out, artifactCheckpointEnvelopeFromOutput(v)) + } + return out +} +func artifactOutputDigests(outputs []pipeline.ArtifactCheckpointOutput) []pipeline.CheckpointFingerprint { + values := make([]pipeline.CheckpointFingerprint, 0, len(outputs)) + for i, v := range outputs { + values = append(values, pipeline.CheckpointFingerprint{Name: fmt.Sprintf("artifact[%d]", i), Value: contentDigest(v.Artifact.Content)}) + } + return normalizeFingerprints(values) +} + func chunkEnvelopes(chunks []source.Chunk) []chunkEnvelope { if len(chunks) == 0 { return nil diff --git a/internal/framework/checkpoint/recorder_test.go b/internal/framework/checkpoint/recorder_test.go index 089858a..e94aef8 100644 --- a/internal/framework/checkpoint/recorder_test.go +++ b/internal/framework/checkpoint/recorder_test.go @@ -177,6 +177,46 @@ func TestWorkspaceLoaderReusesSuccessfulCheckpointFiles(t *testing.T) { } } +func TestWorkspaceArtifactCheckpointsRoundTripCodecIdentityAndBytes(t *testing.T) { + root := t.TempDir() + recorder := newTestRecorder(t, root) + loader := &WorkspaceLoader{root: root} + schema := contracts.ArtifactSchema{ID: "dnd.spell_response", Name: "spell response", Version: "v1", JSONSchema: []byte(`{"type":"object"}`)} + artifact := contracts.SerializedArtifact{Kind: "dnd.spells", Schema: schema, MediaType: "application/json", Content: []byte(`{"spell_casts":[]}`), Metadata: map[string]any{"spell_cast_count": float64(0)}} + stored := pipeline.ArtifactCheckpointOutput{LaneID: "spells", ModuleKey: "dnd/spells", SourceID: "source-1", ChunkID: "chunk-1", ChunkIndex: 2, ChunkRef: source.SourceRef{SourceID: "source-1", StartUnitID: 4, EndUnitID: 8}, Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(schema)} + + extractDeps := []pipeline.CheckpointFingerprint{{Name: "chunks", Value: "sha256:chunks"}} + if err := recorder.ArtifactExtractSucceeded("spells", "dnd/spells", extractDeps, []pipeline.ArtifactCheckpointOutput{stored}, nil, nil); err != nil { + t.Fatalf("ArtifactExtractSucceeded: %v", err) + } + extracted, decision := loader.ArtifactExtract("spells", "dnd/spells", extractDeps) + if !decision.Reused || len(extracted.Outputs) != 1 { + t.Fatalf("extract decision=%#v checkpoint=%#v, want reused", decision, extracted) + } + got := extracted.Outputs[0] + if got.Artifact.Kind != artifact.Kind || got.Artifact.Schema.ID != schema.ID || got.Artifact.Schema.Version != schema.Version || got.SchemaDigest != stored.SchemaDigest || string(got.Artifact.Content) != string(artifact.Content) || got.ChunkRef != stored.ChunkRef { + t.Fatalf("artifact checkpoint = %#v, want codec identity, bytes, and provenance", got) + } + + mergeDeps := artifactOutputDigests([]pipeline.ArtifactCheckpointOutput{stored}) + if err := recorder.ArtifactMergeSucceeded("spells", "merge", mergeDeps, stored, nil); err != nil { + t.Fatalf("ArtifactMergeSucceeded: %v", err) + } + merged, decision := loader.ArtifactMerge("spells", "merge", mergeDeps) + if !decision.Reused || string(merged.Output.Artifact.Content) != string(artifact.Content) { + t.Fatalf("merge decision=%#v checkpoint=%#v, want reused", decision, merged) + } + + normalizeDeps := artifactOutputDigests([]pipeline.ArtifactCheckpointOutput{stored}) + if err := recorder.ArtifactNormalizeSucceeded("spells", "normalize", normalizeDeps, stored, nil); err != nil { + t.Fatalf("ArtifactNormalizeSucceeded: %v", err) + } + normalized, decision := loader.ArtifactNormalize("spells", "normalize", normalizeDeps) + if !decision.Reused || normalized.Output.SchemaDigest != stored.SchemaDigest { + t.Fatalf("normalize decision=%#v checkpoint=%#v, want reused", decision, normalized) + } +} + func TestWorkspaceLoaderInvalidatesMissingCorruptAndMismatchedCheckpoints(t *testing.T) { t.Run("missing", func(t *testing.T) { loader := &WorkspaceLoader{root: t.TempDir()} diff --git a/internal/framework/contracts/artifact.go b/internal/framework/contracts/artifact.go index 0ba3873..1d3c309 100644 --- a/internal/framework/contracts/artifact.go +++ b/internal/framework/contracts/artifact.go @@ -27,6 +27,15 @@ type SerializedArtifact struct { Metadata map[string]any `json:"metadata,omitempty"` } +// SerializedOutput associates a domain-neutral artifact with the pipeline +// operation that produced it. Provenance remains outside codec-owned bytes. +type SerializedOutput struct { + LaneID string `json:"lane_id"` + NormalizerKey string `json:"normalizer_key"` + SourceID string `json:"source_id,omitempty"` + Artifact SerializedArtifact `json:"artifact"` +} + // ArtifactCodec owns the stable encoding for one concrete artifact type. type ArtifactCodec[T any] interface { Kind() ArtifactKind @@ -55,6 +64,11 @@ func CloneSerializedArtifact(artifact SerializedArtifact) SerializedArtifact { return artifact } +func CloneSerializedOutput(output SerializedOutput) SerializedOutput { + output.Artifact = CloneSerializedArtifact(output.Artifact) + return output +} + func cloneArtifactMetadata(metadata map[string]any) map[string]any { if len(metadata) == 0 { return nil diff --git a/internal/framework/contracts/composition_test.go b/internal/framework/contracts/composition_test.go index f1430c0..a305ed0 100644 --- a/internal/framework/contracts/composition_test.go +++ b/internal/framework/contracts/composition_test.go @@ -85,7 +85,7 @@ func TestContractsComposeAcrossPackages(t *testing.T) { output, err := encoder.Encode(ctx, contracts.OutputRequest{ Manifest: artifacts.RunManifest{RunID: "run-1"}, - NormalizeOutputs: []contracts.NormalizeOutput{normalize.Output}, + NormalizeOutputs: []contracts.SerializedOutput{{LaneID: normalize.Output.LaneID, NormalizerKey: normalize.Output.NormalizerKey, SourceID: normalize.Output.SourceID, Artifact: contracts.SerializedArtifact{Schema: contracts.ArtifactSchema{ID: normalize.Output.Schema.ID, Name: normalize.Output.Schema.Name, Version: normalize.Output.Schema.Version}, MediaType: normalize.Output.Payload.MediaType, Content: append([]byte(nil), normalize.Output.Payload.Content...)}}}, }) if err != nil { t.Fatalf("Encode() error = %v, want nil", err) diff --git a/internal/framework/contracts/contracts.go b/internal/framework/contracts/contracts.go index 4964764..b238a88 100644 --- a/internal/framework/contracts/contracts.go +++ b/internal/framework/contracts/contracts.go @@ -370,7 +370,7 @@ type Warning struct { type OutputRequest struct { Manifest artifacts.RunManifest `json:"manifest"` - NormalizeOutputs []NormalizeOutput `json:"normalize_outputs,omitempty"` + NormalizeOutputs []SerializedOutput `json:"normalize_outputs,omitempty"` Rejected []RejectedOutput `json:"rejected,omitempty"` Warnings []Warning `json:"warnings,omitempty"` LLMProfile string `json:"llm_profile,omitempty"` diff --git a/internal/framework/contracts/contracts_test.go b/internal/framework/contracts/contracts_test.go index 230a092..861ca49 100644 --- a/internal/framework/contracts/contracts_test.go +++ b/internal/framework/contracts/contracts_test.go @@ -394,7 +394,7 @@ func TestFakeMergeNormalizeAndOutputContracts(t *testing.T) { encoded, err := encoder.Encode(context.Background(), OutputRequest{ Manifest: artifacts.RunManifest{RunID: "run-1"}, - NormalizeOutputs: []NormalizeOutput{normalized.Output}, + NormalizeOutputs: []SerializedOutput{serializedTestOutput(normalized.Output)}, }) if err != nil { t.Fatalf("Encode() error = %v, want nil", err) @@ -413,6 +413,10 @@ func TestFakeMergeNormalizeAndOutputContracts(t *testing.T) { } } +func serializedTestOutput(output NormalizeOutput) SerializedOutput { + return SerializedOutput{LaneID: output.LaneID, NormalizerKey: output.NormalizerKey, SourceID: output.SourceID, Artifact: SerializedArtifact{Schema: ArtifactSchema{ID: output.Schema.ID, Name: output.Schema.Name, Version: output.Schema.Version, JSONSchema: append([]byte(nil), output.Schema.JSONSchema...)}, MediaType: output.Payload.MediaType, Content: append([]byte(nil), output.Payload.Content...), Metadata: cloneArtifactMetadata(output.Payload.Metadata)}} +} + func TestOutputFileJSONShapeOmitsBytes(t *testing.T) { file := OutputFile{ Name: "artifacts/events.json", diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 200f248..f19939a 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -75,6 +75,46 @@ type NormalizeCheckpoint struct { Warnings []contracts.Warning } +// ArtifactCheckpointOutput is the durable, domain-neutral value stored at a +// typed lane checkpoint boundary. +type ArtifactCheckpointOutput struct { + LaneID string + ModuleKey string + SourceID string + ChunkID string + ChunkIndex int + ChunkRef source.SourceRef + Artifact contracts.SerializedArtifact + SchemaDigest string +} + +type ArtifactExtractCheckpoint struct { + Outputs []ArtifactCheckpointOutput + Rejected []contracts.RejectedOutput + Warnings []contracts.Warning +} + +type ArtifactMergeCheckpoint struct { + Output ArtifactCheckpointOutput + Warnings []contracts.Warning +} +type ArtifactNormalizeCheckpoint struct { + Output ArtifactCheckpointOutput + Warnings []contracts.Warning +} + +type ArtifactCheckpointRecorder interface { + ArtifactExtractSucceeded(string, string, []CheckpointFingerprint, []ArtifactCheckpointOutput, []contracts.RejectedOutput, []contracts.Warning) error + ArtifactMergeSucceeded(string, string, []CheckpointFingerprint, ArtifactCheckpointOutput, []contracts.Warning) error + ArtifactNormalizeSucceeded(string, string, []CheckpointFingerprint, ArtifactCheckpointOutput, []contracts.Warning) error +} + +type ArtifactCheckpointLoader interface { + ArtifactExtract(string, string, []CheckpointFingerprint) (ArtifactExtractCheckpoint, CheckpointDecision) + ArtifactMerge(string, string, []CheckpointFingerprint) (ArtifactMergeCheckpoint, CheckpointDecision) + ArtifactNormalize(string, string, []CheckpointFingerprint) (ArtifactNormalizeCheckpoint, CheckpointDecision) +} + type CheckpointLoader interface { Enabled() bool Source(moduleKey string) (SourceCheckpoint, CheckpointDecision) diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index 1a4fc0c..d4374e3 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -139,6 +139,16 @@ type debugNormalizeOutput struct { Payload debugBinaryEnvelope `json:"payload"` } +type debugSerializedOutput struct { + LaneID string `json:"lane_id"` + NormalizerKey string `json:"normalizer_key"` + SourceID string `json:"source_id,omitempty"` + Kind contracts.ArtifactKind `json:"artifact_kind,omitempty"` + Schema contracts.ArtifactSchema `json:"schema"` + SchemaDigest string `json:"schema_digest"` + Content debugBinaryEnvelope `json:"content"` +} + type debugLLMInputMaterial struct { Name string `json:"name"` MediaType string `json:"media_type,omitempty"` @@ -213,7 +223,7 @@ type debugValidationRequest struct { type debugValidationCall struct { ValidatorName string `json:"validator_name"` - Request debugValidationRequest `json:"request"` + Request any `json:"request"` Result contracts.ValidationResult `json:"result,omitempty"` Error string `json:"error,omitempty"` } @@ -536,6 +546,30 @@ func debugNormalizeOutputEnvelopes(outputs []contracts.NormalizeOutput) []debugN return out } +func debugSerializedOutputEnvelope(output contracts.SerializedOutput) debugSerializedOutput { + schema := contracts.CloneArtifactSchema(output.Artifact.Schema) + digest := contracts.DigestArtifactSchema(schema) + schema.JSONSchema = nil + content := debugContentEnvelope(output.Artifact.Content, output.Artifact.MediaType, output.Artifact.Metadata, nil) + content.ContentDigest = debugContentDigest(output.Artifact.Content) + return debugSerializedOutput{ + LaneID: output.LaneID, NormalizerKey: output.NormalizerKey, SourceID: output.SourceID, + Kind: output.Artifact.Kind, Schema: schema, SchemaDigest: digest, + Content: content, + } +} + +func debugSerializedOutputEnvelopes(outputs []contracts.SerializedOutput) []debugSerializedOutput { + if len(outputs) == 0 { + return nil + } + out := make([]debugSerializedOutput, 0, len(outputs)) + for _, output := range outputs { + out = append(out, debugSerializedOutputEnvelope(output)) + } + return out +} + type debugOutputFile struct { Name string `json:"name"` ContentType string `json:"content_type,omitempty"` diff --git a/internal/framework/pipeline/registry_integration_test.go b/internal/framework/pipeline/registry_integration_test.go index f73a385..1e50954 100644 --- a/internal/framework/pipeline/registry_integration_test.go +++ b/internal/framework/pipeline/registry_integration_test.go @@ -256,7 +256,7 @@ func integrationSourceDocument() *source.SourceDocument { } } -func normalizeOutputKeys(outputs []contracts.NormalizeOutput) []string { +func normalizeOutputKeys(outputs []contracts.SerializedOutput) []string { keys := make([]string, 0, len(outputs)) for _, output := range outputs { keys = append(keys, output.NormalizerKey) diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index a62d07b..3327c11 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -56,12 +56,12 @@ type RunInput struct { } type RunOutput struct { - Manifest artifacts.RunManifest `json:"manifest"` - NormalizeOutputs []contracts.NormalizeOutput `json:"normalize_outputs,omitempty"` - Rejected []contracts.RejectedOutput `json:"rejected,omitempty"` - Warnings []contracts.Warning `json:"warnings,omitempty"` - OutputFiles []contracts.OutputFile `json:"-"` - CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"` + Manifest artifacts.RunManifest `json:"manifest"` + NormalizeOutputs []contracts.SerializedOutput `json:"normalize_outputs,omitempty"` + Rejected []contracts.RejectedOutput `json:"rejected,omitempty"` + Warnings []contracts.Warning `json:"warnings,omitempty"` + OutputFiles []contracts.OutputFile `json:"-"` + CheckpointEvents []CheckpointEvent `json:"checkpoint_events,omitempty"` } func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err error) { @@ -335,7 +335,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err StartedAt: outputStarted, Payload: map[string]any{ "manifest": output.Manifest, - "normalize_outputs": debugNormalizeOutputEnvelopes(output.NormalizeOutputs), + "normalize_outputs": debugSerializedOutputEnvelopes(output.NormalizeOutputs), "rejected": debugRejectedOutputEnvelopes(output.Rejected), "warnings": output.Warnings, "options": redactSensitiveMap(input.pipeline.Output.Options), @@ -346,7 +346,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } encoded, err := encoder.Encode(ctx, contracts.OutputRequest{ Manifest: output.Manifest, - NormalizeOutputs: cloneNormalizeOutputs(output.NormalizeOutputs), + NormalizeOutputs: cloneSerializedOutputs(output.NormalizeOutputs), Rejected: cloneRejectedOutputs(output.Rejected), Warnings: output.Warnings, LLMProfile: input.pipeline.Output.LLMProfile, @@ -865,7 +865,7 @@ func (r *Runner) runLegacyLane(ctx context.Context, input RunInput, checkpoints }); err != nil { return fmt.Errorf("write normalize debug artifact for lane %q: %w", lane.ID, err) } - output.NormalizeOutputs = append(output.NormalizeOutputs, acceptedNormalize) + output.NormalizeOutputs = append(output.NormalizeOutputs, serializedOutputFromLegacy(acceptedNormalize)) return nil } @@ -1282,7 +1282,7 @@ func populateRawOutputManifest(output *RunOutput) { output.Manifest.RejectedOutputs = rejectedOutputManifests(output.Rejected) } -func normalizedOutputManifests(outputs []contracts.NormalizeOutput) []artifacts.NormalizedOutputManifest { +func normalizedOutputManifests(outputs []contracts.SerializedOutput) []artifacts.NormalizedOutputManifest { if len(outputs) == 0 { return nil } @@ -1292,11 +1292,11 @@ func normalizedOutputManifests(outputs []contracts.NormalizeOutput) []artifacts. LaneID: output.LaneID, ModuleKey: output.NormalizerKey, SourceID: output.SourceID, - MediaType: output.Payload.MediaType, + MediaType: output.Artifact.MediaType, Schema: artifacts.OutputSchemaProvenance{ - ID: output.Schema.ID, - Name: output.Schema.Name, - Version: output.Schema.Version, + ID: output.Artifact.Schema.ID, + Name: output.Artifact.Schema.Name, + Version: output.Artifact.Schema.Version, }, }) } @@ -1637,17 +1637,27 @@ func cloneNormalizeOutput(output contracts.NormalizeOutput) contracts.NormalizeO return output } -func cloneNormalizeOutputs(outputs []contracts.NormalizeOutput) []contracts.NormalizeOutput { +func cloneSerializedOutputs(outputs []contracts.SerializedOutput) []contracts.SerializedOutput { if len(outputs) == 0 { return nil } - out := make([]contracts.NormalizeOutput, 0, len(outputs)) + out := make([]contracts.SerializedOutput, 0, len(outputs)) for _, output := range outputs { - out = append(out, cloneNormalizeOutput(output)) + out = append(out, contracts.CloneSerializedOutput(output)) } return out } +func serializedOutputFromLegacy(output contracts.NormalizeOutput) contracts.SerializedOutput { + return contracts.SerializedOutput{ + LaneID: output.LaneID, NormalizerKey: output.NormalizerKey, SourceID: output.SourceID, + Artifact: contracts.SerializedArtifact{ + Schema: contracts.ArtifactSchema{ID: output.Schema.ID, Name: output.Schema.Name, Version: output.Schema.Version, JSONSchema: append([]byte(nil), output.Schema.JSONSchema...)}, + MediaType: output.Payload.MediaType, Content: append([]byte(nil), output.Payload.Content...), Metadata: cloneMetadata(output.Payload.Metadata), + }, + } +} + func cloneRejectedOutputs(rejected []contracts.RejectedOutput) []contracts.RejectedOutput { if len(rejected) == 0 { return nil diff --git a/internal/framework/pipeline/runner_test.go b/internal/framework/pipeline/runner_test.go index 42c8d4b..d3aed3a 100644 --- a/internal/framework/pipeline/runner_test.go +++ b/internal/framework/pipeline/runner_test.go @@ -29,7 +29,7 @@ func TestNewAndDataTypes(t *testing.T) { } output := RunOutput{ Manifest: artifacts.RunManifest{PipelineID: "pipeline-1"}, - NormalizeOutputs: []contracts.NormalizeOutput{{NormalizerKey: "normalize"}}, + NormalizeOutputs: []contracts.SerializedOutput{{NormalizerKey: "normalize"}}, Rejected: []contracts.RejectedOutput{{ValidatorName: "validator"}}, Warnings: []contracts.Warning{{ReasonCode: "note", Message: "message"}}, OutputFiles: []contracts.OutputFile{{Name: "outputs/generic.json", ContentType: "application/json", Bytes: []byte(`{}`)}}, @@ -1177,7 +1177,7 @@ func TestRunReusesCheckpointedWorkflowOutputs(t *testing.T) { if len(modules.input.requests) != 0 || len(modules.chunker.requests) != 0 || len(modules.extractors["extract-alpha"].requests) != 0 || len(modules.mergers["merge"].requests) != 0 || len(modules.normalizers["normalize"].requests) != 0 { t.Fatalf("module requests = input:%d chunk:%d extract:%d merge:%d normalize:%d, want all skipped", len(modules.input.requests), len(modules.chunker.requests), len(modules.extractors["extract-alpha"].requests), len(modules.mergers["merge"].requests), len(modules.normalizers["normalize"].requests)) } - if len(output.NormalizeOutputs) != 1 || string(output.NormalizeOutputs[0].Payload.Content) != `{"cached_normalize":true}` { + if len(output.NormalizeOutputs) != 1 || string(output.NormalizeOutputs[0].Artifact.Content) != `{"cached_normalize":true}` { t.Fatalf("NormalizeOutputs = %#v, want cached normalize output", output.NormalizeOutputs) } if len(output.CheckpointEvents) != 5 { diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 251f1ef..1b521de 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -2,6 +2,8 @@ package pipeline import ( "context" + "crypto/sha256" + "encoding/hex" "fmt" "path" "sort" @@ -11,50 +13,133 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) -// migrationRawArtifact serializes a typed value into the existing raw -// checkpoint, debug, and output envelopes. It is removed when those boundaries -// consume SerializedArtifact directly. -func migrationRawArtifact(codec artifactCodecEntry, value any) (contracts.ResponseSchema, contracts.RawPayload, error) { - content, err := codec.encodeCandidate(value) +func loadArtifactExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ArtifactExtractCheckpoint, CheckpointDecision) { + typed, ok := loader.(ArtifactCheckpointLoader) + if !ok { + return ArtifactExtractCheckpoint{}, CheckpointDecision{Reason: "artifact checkpoint loading is unavailable"} + } + return typed.ArtifactExtract(laneID, moduleKey, deps) +} +func loadArtifactMerge(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ArtifactMergeCheckpoint, CheckpointDecision) { + typed, ok := loader.(ArtifactCheckpointLoader) + if !ok { + return ArtifactMergeCheckpoint{}, CheckpointDecision{Reason: "artifact checkpoint loading is unavailable"} + } + return typed.ArtifactMerge(laneID, moduleKey, deps) +} +func loadArtifactNormalize(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ArtifactNormalizeCheckpoint, CheckpointDecision) { + typed, ok := loader.(ArtifactCheckpointLoader) + if !ok { + return ArtifactNormalizeCheckpoint{}, CheckpointDecision{Reason: "artifact checkpoint loading is unavailable"} + } + return typed.ArtifactNormalize(laneID, moduleKey, deps) +} +func recordArtifactExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []ArtifactCheckpointOutput, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + typed, ok := recorder.(ArtifactCheckpointRecorder) + if !ok { + return nil + } + return typed.ArtifactExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings) +} +func recordArtifactMerge(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, output ArtifactCheckpointOutput, warnings []contracts.Warning) error { + typed, ok := recorder.(ArtifactCheckpointRecorder) + if !ok { + return nil + } + return typed.ArtifactMergeSucceeded(laneID, moduleKey, deps, output, warnings) +} +func recordArtifactNormalize(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, output ArtifactCheckpointOutput, warnings []contracts.Warning) error { + typed, ok := recorder.(ArtifactCheckpointRecorder) + if !ok { + return nil + } + return typed.ArtifactNormalizeSucceeded(laneID, moduleKey, deps, output, warnings) +} +func cloneArtifactCheckpointOutput(output ArtifactCheckpointOutput) ArtifactCheckpointOutput { + output.Artifact = contracts.CloneSerializedArtifact(output.Artifact) + return output +} + +func hydrateCheckpointArtifact(codec artifactCodecEntry, output ArtifactCheckpointOutput, value any) ArtifactCheckpointOutput { + output.Artifact.Schema = contracts.CloneArtifactSchema(codec.spec.Schema) + if codec.metadata != nil { + output.Artifact.Metadata = cloneMetadata(codec.metadata(value)) + } else { + output.Artifact.Metadata = nil + } + return output +} +func artifactCheckpointDigests(outputs []ArtifactCheckpointOutput) []CheckpointFingerprint { + values := make([]CheckpointFingerprint, 0, len(outputs)) + for i, output := range outputs { + sum := sha256.Sum256(output.Artifact.Content) + values = append(values, CheckpointFingerprint{Name: fmt.Sprintf("artifact[%d]", i), Value: "sha256:" + hex.EncodeToString(sum[:])}) + } + return normalizeCheckpointFingerprints(values) +} +func debugArtifactCheckpointOutput(output ArtifactCheckpointOutput) map[string]any { + artifact := output.Artifact + schema := contracts.CloneArtifactSchema(artifact.Schema) + digest := output.SchemaDigest + if digest == "" { + digest = contracts.DigestArtifactSchema(schema) + } + schema.JSONSchema = nil + content := debugContentEnvelope(artifact.Content, artifact.MediaType, artifact.Metadata, nil) + content.ContentDigest = debugContentDigest(artifact.Content) + return map[string]any{"lane_id": output.LaneID, "module_key": output.ModuleKey, "source_id": output.SourceID, "chunk_id": output.ChunkID, "chunk_index": output.ChunkIndex, "chunk_ref": output.ChunkRef, "artifact_kind": artifact.Kind, "schema": schema, "schema_digest": digest, "content": content} +} +func debugArtifactCheckpointOutputs(outputs []ArtifactCheckpointOutput) []map[string]any { + if len(outputs) == 0 { + return nil + } + out := make([]map[string]any, 0, len(outputs)) + for _, output := range outputs { + out = append(out, debugArtifactCheckpointOutput(output)) + } + return out +} + +func serializeArtifact(codec artifactCodecEntry, value any, candidate bool) (contracts.SerializedArtifact, error) { + encode := codec.encode + if candidate { + encode = codec.encodeCandidate + } + content, err := encode(value) if err != nil { - return contracts.ResponseSchema{}, contracts.RawPayload{}, err + return contracts.SerializedArtifact{}, err } schema := codec.spec.Schema metadata := map[string]any(nil) if codec.metadata != nil { metadata = codec.metadata(value) } - return contracts.ResponseSchema{ID: schema.ID, Name: schema.Name, Version: schema.Version, JSONSchema: append([]byte(nil), schema.JSONSchema...)}, contracts.RawPayload{Content: content, MediaType: codec.spec.MediaType, Metadata: metadata}, nil + return contracts.SerializedArtifact{Kind: codec.spec.Kind, Schema: contracts.CloneArtifactSchema(schema), MediaType: codec.spec.MediaType, Content: append([]byte(nil), content...), Metadata: cloneMetadata(metadata)}, nil } -// migrationDecodeArtifact restores a typed value from an existing raw -// checkpoint envelope through the registered codec. -func migrationDecodeArtifact(codec artifactCodecEntry, payload contracts.RawPayload) (any, error) { - return codec.decode(append([]byte(nil), payload.Content...)) -} - -func migrationExtractOutput(codec artifactCodecEntry, artifact erasedExtractArtifact) (contracts.ExtractOutput, error) { - schema, payload, err := migrationRawArtifact(codec, artifact.Value) - if err != nil { - return contracts.ExtractOutput{}, err +func decodeCheckpointArtifact(codec artifactCodecEntry, artifact ArtifactCheckpointOutput) (any, error) { + expectedDigest := contracts.DigestArtifactSchema(codec.spec.Schema) + if artifact.Artifact.Kind != codec.spec.Kind { + return nil, fmt.Errorf("artifact kind %q does not match codec %q", artifact.Artifact.Kind, codec.spec.Kind) } - return contracts.ExtractOutput{LaneID: artifact.LaneID, ExtractorKey: artifact.ExtractorKey, SourceID: artifact.SourceID, ChunkID: artifact.ChunkID, ChunkIndex: artifact.ChunkIndex, Schema: schema, Payload: payload}, nil + if artifact.Artifact.Schema.ID != codec.spec.Schema.ID || artifact.Artifact.Schema.Version != codec.spec.Schema.Version { + return nil, fmt.Errorf("artifact schema %q version %q does not match codec schema %q version %q", artifact.Artifact.Schema.ID, artifact.Artifact.Schema.Version, codec.spec.Schema.ID, codec.spec.Schema.Version) + } + if artifact.SchemaDigest != expectedDigest { + return nil, fmt.Errorf("artifact schema digest %q does not match codec schema digest %q", artifact.SchemaDigest, expectedDigest) + } + if artifact.Artifact.MediaType != codec.spec.MediaType { + return nil, fmt.Errorf("artifact media type %q does not match codec media type %q", artifact.Artifact.MediaType, codec.spec.MediaType) + } + return codec.decode(append([]byte(nil), artifact.Artifact.Content...)) } -func migrationMergeOutput(codec artifactCodecEntry, artifact erasedMergeArtifact) (contracts.MergeOutput, error) { - schema, payload, err := migrationRawArtifact(codec, artifact.Value) +func checkpointArtifact(codec artifactCodecEntry, laneID, moduleKey, sourceID string, value any) (ArtifactCheckpointOutput, error) { + serialized, err := serializeArtifact(codec, value, false) if err != nil { - return contracts.MergeOutput{}, err + return ArtifactCheckpointOutput{}, err } - return contracts.MergeOutput{LaneID: artifact.LaneID, MergerKey: artifact.MergerKey, SourceID: artifact.SourceID, Schema: schema, Payload: payload}, nil -} - -func migrationNormalizeOutput(codec artifactCodecEntry, laneID, key, sourceID string, value any) (contracts.NormalizeOutput, error) { - schema, payload, err := migrationRawArtifact(codec, value) - if err != nil { - return contracts.NormalizeOutput{}, err - } - return contracts.NormalizeOutput{LaneID: laneID, NormalizerKey: key, SourceID: sourceID, Schema: schema, Payload: payload}, nil + return ArtifactCheckpointOutput{LaneID: laneID, ModuleKey: moduleKey, SourceID: sourceID, Artifact: serialized, SchemaDigest: contracts.DigestArtifactSchema(serialized.Schema)}, nil } func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, prepared preparedLaneExecutor, output *RunOutput) error { @@ -65,7 +150,7 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer) values := make([]erasedExtractArtifact, 0, len(chunks)) - rawExtracts := make([]contracts.ExtractOutput, 0, len(chunks)) + serializedExtracts := make([]ArtifactCheckpointOutput, 0, len(chunks)) extractWarnings := []contracts.Warning{} rejectedStart := len(output.Rejected) chunksDigest, err := joinedChunkDigest(chunks) @@ -73,23 +158,32 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C return fmt.Errorf("digest chunks for lane %q: %w", lane.ID, err) } extractDeps := digestFingerprints("chunks", chunksDigest) - cp, decision := loader.Extract(lane.ID, lane.Extract.Module, extractDeps) + cp, decision := loadArtifactExtract(loader, lane.ID, lane.Extract.Module, extractDeps) + if decision.Reused { + for _, stored := range cp.Outputs { + if _, decodeErr := decodeCheckpointArtifact(typed.codec, stored); decodeErr != nil { + decision = CheckpointDecision{Reason: "extract artifact checkpoint codec is incompatible: " + decodeErr.Error()} + break + } + } + } recordCheckpointEvent(output, loader, string(StageExtract), lane.ID, lane.Extract.Module, decision) if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": decision.Reused, "decision": decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return err } if decision.Reused { - for _, raw := range cp.Outputs { - value, decodeErr := migrationDecodeArtifact(typed.codec, raw.Payload) + for _, stored := range cp.Outputs { + value, decodeErr := decodeCheckpointArtifact(typed.codec, stored) if decodeErr != nil { return fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr) } - artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: raw.ChunkID, ChunkIndex: raw.ChunkIndex, Value: value} - if raw.ChunkIndex >= 0 && raw.ChunkIndex < len(chunks) { - artifact.ChunkRef = chunks[raw.ChunkIndex].Ref + stored = hydrateCheckpointArtifact(typed.codec, stored, value) + artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: stored.ChunkID, ChunkIndex: stored.ChunkIndex, ChunkRef: stored.ChunkRef, Value: value} + if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) { + artifact.ChunkRef = chunks[stored.ChunkIndex].Ref } values = append(values, artifact) - rawExtracts = append(rawExtracts, cloneExtractOutput(raw)) + serializedExtracts = append(serializedExtracts, cloneArtifactCheckpointOutput(stored)) } extractWarnings = cloneWarnings(cp.Warnings) output.Warnings = append(output.Warnings, extractWarnings...) @@ -101,7 +195,7 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C for i := range chunks { chunk := chunks[i] var accepted erasedExtractArtifact - var rawAccepted contracts.ExtractOutput + var serializedAccepted ArtifactCheckpointOutput var acceptedWarnings []contracts.Warning ok, rejection, runErr := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { started := time.Now().UTC() @@ -117,14 +211,14 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C if validateErr != nil || rejected != nil { return false, rejected, validateErr } - raw, encodeErr := migrationExtractOutput(typed.codec, artifact) + stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value) if encodeErr != nil { return false, nil, encodeErr } - raw.Payload.Warnings = append(raw.Payload.Warnings, cloneWarnings(result.Warnings)...) - accepted, rawAccepted = artifact, raw + stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef + accepted, serializedAccepted = artifact, stored acceptedWarnings = append(cloneWarnings(result.Warnings), warnings...) - if debugErr := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugExtractOutputEnvelope(raw), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope)); debugErr != nil { + if debugErr := writeDebugTimed(input.Debug, attemptPath+".json", debugEnvelopeWithLLMCalls(debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugArtifactCheckpointOutput(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope)); debugErr != nil { return false, nil, debugErr } return true, nil, nil @@ -138,17 +232,17 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C continue } values = append(values, accepted) - rawExtracts = append(rawExtracts, rawAccepted) + serializedExtracts = append(serializedExtracts, serializedAccepted) extractWarnings = append(extractWarnings, acceptedWarnings...) output.Warnings = append(output.Warnings, acceptedWarnings...) } - if err := checkpoints.ExtractSucceeded(lane.ID, lane.Extract.Module, extractDeps, rawExtracts, cloneRejectedOutputs(output.Rejected[rejectedStart:]), extractWarnings); err != nil { + if err := recordArtifactExtract(checkpoints, lane.ID, lane.Extract.Module, extractDeps, serializedExtracts, cloneRejectedOutputs(output.Rejected[rejectedStart:]), extractWarnings); err != nil { return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err) } } sort.SliceStable(values, func(i, j int) bool { return values[i].ChunkIndex < values[j].ChunkIndex }) - sort.SliceStable(rawExtracts, func(i, j int) bool { return rawExtracts[i].ChunkIndex < rawExtracts[j].ChunkIndex }) - if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": decision.Reused, "outputs": debugExtractOutputEnvelopes(rawExtracts), "rejected": debugRejectedOutputEnvelopes(output.Rejected[rejectedStart:]), "warnings": debugWarningEnvelopes(extractWarnings)}}); err != nil { + sort.SliceStable(serializedExtracts, func(i, j int) bool { return serializedExtracts[i].ChunkIndex < serializedExtracts[j].ChunkIndex }) + if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": decision.Reused, "outputs": debugArtifactCheckpointOutputs(serializedExtracts), "rejected": debugRejectedOutputEnvelopes(output.Rejected[rejectedStart:]), "warnings": debugWarningEnvelopes(extractWarnings)}}); err != nil { return err } if len(values) == 0 { @@ -159,22 +253,27 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C for i, value := range values { mergeInputs[i] = contracts.ExtractArtifact[any]{LaneID: value.LaneID, ExtractorKey: value.ExtractorKey, SourceID: value.SourceID, ChunkID: value.ChunkID, ChunkIndex: value.ChunkIndex, ChunkRef: value.ChunkRef, Value: value.Value} } - mergeDeps := rawOutputDigests(extractPayloads(rawExtracts)) - mergeCP, mergeDecision := loader.Merge(lane.ID, lane.Merge.Module, mergeDeps) + mergeDeps := artifactCheckpointDigests(serializedExtracts) + mergeCP, mergeDecision := loadArtifactMerge(loader, lane.ID, lane.Merge.Module, mergeDeps) + if mergeDecision.Reused { + if _, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil { + mergeDecision = CheckpointDecision{Reason: "merge artifact checkpoint codec is incompatible: " + decodeErr.Error()} + } + } recordCheckpointEvent(output, loader, string(StageMerge), lane.ID, lane.Merge.Module, mergeDecision) - if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugExtractOutputEnvelopes(rawExtracts), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugArtifactCheckpointOutputs(serializedExtracts), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return err } var merged erasedMergeArtifact - var rawMerge contracts.MergeOutput + var serializedMerge ArtifactCheckpointOutput var mergeWarnings []contracts.Warning if mergeDecision.Reused { - value, decodeErr := migrationDecodeArtifact(typed.codec, mergeCP.Output.Payload) + value, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output) if decodeErr != nil { return fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr) } merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: value} - rawMerge = cloneMergeOutput(mergeCP.Output) + serializedMerge = hydrateCheckpointArtifact(typed.codec, cloneArtifactCheckpointOutput(mergeCP.Output), value) mergeWarnings = cloneWarnings(mergeCP.Warnings) output.Warnings = append(output.Warnings, mergeWarnings...) } else { @@ -191,12 +290,11 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C if validateErr != nil || rejected != nil { return false, rejected, validateErr } - raw, encodeErr := migrationMergeOutput(typed.codec, candidate) + stored, encodeErr := checkpointArtifact(typed.codec, candidate.LaneID, candidate.MergerKey, candidate.SourceID, candidate.Value) if encodeErr != nil { return false, nil, encodeErr } - raw.Payload.Warnings = append(raw.Payload.Warnings, cloneWarnings(result.Warnings)...) - merged, rawMerge = candidate, raw + merged, serializedMerge = candidate, stored mergeWarnings = append(cloneWarnings(result.Warnings), warnings...) return true, nil, nil }) @@ -212,28 +310,33 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C return nil } output.Warnings = append(output.Warnings, mergeWarnings...) - if err := checkpoints.MergeSucceeded(lane.ID, lane.Merge.Module, mergeDeps, rawMerge, mergeWarnings); err != nil { + if err := recordArtifactMerge(checkpoints, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil { return err } } - if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugMergeOutputEnvelope(rawMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugArtifactCheckpointOutput(serializedMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil { return err } - normalizeDeps := rawOutputDigests([]contracts.RawPayload{rawMerge.Payload}) - normalizeCP, normalizeDecision := loader.Normalize(lane.ID, lane.Normalize.Module, normalizeDeps) + normalizeDeps := artifactCheckpointDigests([]ArtifactCheckpointOutput{serializedMerge}) + normalizeCP, normalizeDecision := loadArtifactNormalize(loader, lane.ID, lane.Normalize.Module, normalizeDeps) + if normalizeDecision.Reused { + if _, decodeErr := decodeCheckpointArtifact(typed.codec, normalizeCP.Output); decodeErr != nil { + normalizeDecision = CheckpointDecision{Reason: "normalize artifact checkpoint codec is incompatible: " + decodeErr.Error()} + } + } recordCheckpointEvent(output, loader, string(StageNormalize), lane.ID, lane.Normalize.Module, normalizeDecision) - if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugMergeOutputEnvelope(rawMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugArtifactCheckpointOutput(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return err } - var rawNormalize contracts.NormalizeOutput + var serializedNormalize ArtifactCheckpointOutput var normalizeWarnings []contracts.Warning if normalizeDecision.Reused { - _, decodeErr := migrationDecodeArtifact(typed.codec, normalizeCP.Output.Payload) + value, decodeErr := decodeCheckpointArtifact(typed.codec, normalizeCP.Output) if decodeErr != nil { return fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr) } - rawNormalize, normalizeWarnings = cloneNormalizeOutput(normalizeCP.Output), cloneWarnings(normalizeCP.Warnings) + serializedNormalize, normalizeWarnings = hydrateCheckpointArtifact(typed.codec, cloneArtifactCheckpointOutput(normalizeCP.Output), value), cloneWarnings(normalizeCP.Warnings) output.Warnings = append(output.Warnings, normalizeWarnings...) } else { if err := checkpoints.NormalizeRunning(lane.ID, lane.Normalize.Module, normalizeDeps); err != nil { @@ -248,12 +351,11 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C if validateErr != nil || rejected != nil { return false, rejected, validateErr } - raw, encodeErr := migrationNormalizeOutput(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value) + stored, encodeErr := checkpointArtifact(typed.codec, lane.ID, lane.Normalize.Module, doc.ID, result.Value) if encodeErr != nil { return false, nil, encodeErr } - raw.Payload.Warnings = append(raw.Payload.Warnings, cloneWarnings(result.Warnings)...) - rawNormalize = raw + serializedNormalize = stored normalizeWarnings = append(cloneWarnings(result.Warnings), warnings...) return true, nil, nil }) @@ -269,14 +371,14 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C return nil } output.Warnings = append(output.Warnings, normalizeWarnings...) - if err := checkpoints.NormalizeSucceeded(lane.ID, lane.Normalize.Module, normalizeDeps, rawNormalize, normalizeWarnings); err != nil { + if err := recordArtifactNormalize(checkpoints, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil { return err } } - if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugNormalizeOutputEnvelope(rawNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugArtifactCheckpointOutput(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil { return err } - output.NormalizeOutputs = append(output.NormalizeOutputs, rawNormalize) + output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(serializedNormalize.Artifact)}) return nil } @@ -318,18 +420,17 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE target.llmProfile = binding.LLMProfile result, err = item.typedValidate(validatorCtx, item.typed, target) case ValidatorTargetSerialized: - schema, payload, encodeErr := migrationRawArtifact(codec, target.value) + artifact, encodeErr := serializeArtifact(codec, target.value, true) if encodeErr != nil { err = encodeErr break } - result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: contracts.ArtifactSchema{ID: schema.ID, Name: schema.Name, Version: schema.Version, JSONSchema: append([]byte(nil), schema.JSONSchema...)}, MediaType: payload.MediaType, Content: append([]byte(nil), payload.Content...)}) + result, err = item.serialized.Validate(validatorCtx, contracts.SerializedValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: contracts.CloneArtifactSchema(artifact.Schema), MediaType: artifact.MediaType, Content: append([]byte(nil), artifact.Content...)}) default: return nil, nil, fmt.Errorf("validator %q is incompatible with typed artifact validation", binding.Module) } - schema, payload, _ := migrationRawArtifact(codec, target.value) - debugRequest := contracts.ValidationRequest{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Source: target.source, SourceID: target.sourceID, SourceInput: target.sourceInput.Clone(), SessionID: target.sessionID, References: CloneReferenceSet(target.references), LLMProfile: binding.LLMProfile, Metadata: cloneMetadata(target.metadata), Chunk: cloneSourceChunkPtr(target.chunk), Chunks: cloneSourceChunks(target.chunks), Schema: schema, Payload: payload} - debugCall := debugValidationCall{ValidatorName: binding.Module, Request: debugValidationRequestEnvelope(debugRequest), Result: debugValidationResultEnvelope(result)} + artifact, _ := serializeArtifact(codec, target.value, true) + debugCall := debugValidationCall{ValidatorName: binding.Module, Request: map[string]any{"stage": string(target.stage), "lane_id": target.laneID, "module_key": target.moduleKey, "source_id": target.sourceID, "artifact": debugArtifactCheckpointOutput(ArtifactCheckpointOutput{Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(artifact.Schema)}), "metadata": redactSensitiveMap(target.metadata)}, Result: debugValidationResultEnvelope(result)} if err != nil { debugCall.Error = err.Error() } @@ -342,11 +443,11 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE if !result.Approved { reason := result.ReasonCode if reason == "" { - reason = "raw_output_rejected" + reason = "artifact_rejected" } message := result.Message if message == "" { - message = "raw output rejected" + message = "artifact rejected" } return nil, &contracts.RejectedOutput{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, ChunkID: func() string { if target.chunk != nil { diff --git a/internal/framework/pipeline/runner_typed_checkpoint_test.go b/internal/framework/pipeline/runner_typed_checkpoint_test.go new file mode 100644 index 0000000..b668831 --- /dev/null +++ b/internal/framework/pipeline/runner_typed_checkpoint_test.go @@ -0,0 +1,45 @@ +package pipeline + +import ( + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +func TestDecodeCheckpointArtifactRejectsIncompatibleCodecIdentityAndBytes(t *testing.T) { + registry := NewArtifactCodecRegistry() + if err := RegisterArtifactCodec(registry, notesCodec()); err != nil { + t.Fatalf("RegisterArtifactCodec: %v", err) + } + codec, _, err := registry.entry("test/notes") + if err != nil { + t.Fatalf("entry: %v", err) + } + artifact, err := serializeArtifact(codec, codecNotes{Items: []string{"one"}}, false) + if err != nil { + t.Fatalf("serializeArtifact: %v", err) + } + base := ArtifactCheckpointOutput{Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(artifact.Schema)} + + tests := []struct { + name string + mutate func(*ArtifactCheckpointOutput) + want string + }{ + {name: "missing kind", mutate: func(v *ArtifactCheckpointOutput) { v.Artifact.Kind = "" }, want: "artifact kind"}, + {name: "schema version", mutate: func(v *ArtifactCheckpointOutput) { v.Artifact.Schema.Version = "v999" }, want: "does not match codec schema"}, + {name: "schema digest", mutate: func(v *ArtifactCheckpointOutput) { v.SchemaDigest = "sha256:different" }, want: "schema digest"}, + {name: "media type", mutate: func(v *ArtifactCheckpointOutput) { v.Artifact.MediaType = "text/plain" }, want: "media type"}, + {name: "decode failure", mutate: func(v *ArtifactCheckpointOutput) { v.Artifact.Content = []byte(`{"items":[`) }, want: "unexpected EOF"}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + stored := cloneArtifactCheckpointOutput(base) + test.mutate(&stored) + if _, err := decodeCheckpointArtifact(codec, stored); err == nil || !strings.Contains(err.Error(), test.want) { + t.Fatalf("decode error = %v, want %q", err, test.want) + } + }) + } +} diff --git a/internal/framework/pipeline/walking_skeleton_test.go b/internal/framework/pipeline/walking_skeleton_test.go index 6a3de10..359a7fe 100644 --- a/internal/framework/pipeline/walking_skeleton_test.go +++ b/internal/framework/pipeline/walking_skeleton_test.go @@ -398,9 +398,9 @@ func (output walkingSkeletonOutput) Encode(ctx context.Context, req contracts.Ou LaneID: output.LaneID, NormalizerKey: output.NormalizerKey, SourceID: output.SourceID, - Schema: output.Schema, - MediaType: output.Payload.MediaType, - Content: json.RawMessage(output.Payload.Content), + Schema: contracts.ResponseSchema{ID: output.Artifact.Schema.ID, Name: output.Artifact.Schema.Name, Version: output.Artifact.Schema.Version}, + MediaType: output.Artifact.MediaType, + Content: json.RawMessage(output.Artifact.Content), }) } encoded, err := json.Marshal(struct { diff --git a/internal/modules/generic/output/json/encoder.go b/internal/modules/generic/output/json/encoder.go index 2d4891e..17d7498 100644 --- a/internal/modules/generic/output/json/encoder.go +++ b/internal/modules/generic/output/json/encoder.go @@ -136,14 +136,14 @@ func logicalFiles(req contracts.OutputRequest) ([]contracts.OutputFile, error) { usedOutputFiles[name] = output.LaneID outputIndexes = append(outputIndexes, outputFileIndex{ LaneID: output.LaneID, - MediaType: output.Payload.MediaType, + MediaType: output.Artifact.MediaType, File: name, ModuleKey: output.NormalizerKey, - SchemaID: output.Schema.ID, - SchemaName: output.Schema.Name, - SchemaVer: output.Schema.Version, + SchemaID: output.Artifact.Schema.ID, + SchemaName: output.Artifact.Schema.Name, + SchemaVer: output.Artifact.Schema.Version, }) - file, err := rawOutputFile(name, output.Payload) + file, err := serializedOutputFile(name, output.Artifact) if err != nil { return nil, err } @@ -175,12 +175,12 @@ func logicalFiles(req contracts.OutputRequest) ([]contracts.OutputFile, error) { return files, nil } -func rawOutputFile(name string, payload contracts.RawPayload) (contracts.OutputFile, error) { - content := append([]byte(nil), payload.Content...) +func serializedOutputFile(name string, artifact contracts.SerializedArtifact) (contracts.OutputFile, error) { + content := append([]byte(nil), artifact.Content...) if len(content) == 0 { content = []byte("null") } - mediaType := strings.TrimSpace(payload.MediaType) + mediaType := strings.TrimSpace(artifact.MediaType) if mediaType == "" { mediaType = "application/octet-stream" } @@ -242,14 +242,13 @@ func outputFileName(laneID string) (string, error) { return "lanes/" + sanitized + ".json", nil } -func cloneNormalizeOutputs(outputs []contracts.NormalizeOutput) []contracts.NormalizeOutput { +func cloneNormalizeOutputs(outputs []contracts.SerializedOutput) []contracts.SerializedOutput { if len(outputs) == 0 { return nil } - out := make([]contracts.NormalizeOutput, 0, len(outputs)) + out := make([]contracts.SerializedOutput, 0, len(outputs)) for _, output := range outputs { - output.Payload = cloneRawPayload(output.Payload) - out = append(out, output) + out = append(out, contracts.CloneSerializedOutput(output)) } return out } diff --git a/internal/modules/generic/output/json/encoder_test.go b/internal/modules/generic/output/json/encoder_test.go index 41c21fd..08386b9 100644 --- a/internal/modules/generic/output/json/encoder_test.go +++ b/internal/modules/generic/output/json/encoder_test.go @@ -42,7 +42,7 @@ func TestModuleSpecAndRegister(t *testing.T) { func TestEncodeReturnsLogicalFilesForNormalizedOutputs(t *testing.T) { req := contracts.OutputRequest{ Manifest: artifacts.RunManifest{RunID: "run-1", PipelineID: "pipeline-1"}, - NormalizeOutputs: []contracts.NormalizeOutput{ + NormalizeOutputs: []contracts.SerializedOutput{ normalizeOutput("spells", `{"spell_casts":[{"spell":"Cure Wounds"}]}`), normalizeOutput("notes/items", `{"items":[{"name":"Torch"}]}`), }, @@ -234,7 +234,7 @@ func TestEncodeIncludesManifestRawOutputProvenance(t *testing.T) { func TestEncodeRejectsLaneIDWithoutSafeFileName(t *testing.T) { _, err := New().Encode(context.Background(), contracts.OutputRequest{ - NormalizeOutputs: []contracts.NormalizeOutput{normalizeOutput("///", `{"value":true}`)}, + NormalizeOutputs: []contracts.SerializedOutput{normalizeOutput("///", `{"value":true}`)}, }) if err == nil { t.Fatal("Encode() error = nil, want unsafe lane id error") @@ -246,7 +246,7 @@ func TestEncodeRejectsLaneIDWithoutSafeFileName(t *testing.T) { func TestEncodeSanitizesParentPathSequences(t *testing.T) { result, err := New().Encode(context.Background(), contracts.OutputRequest{ - NormalizeOutputs: []contracts.NormalizeOutput{normalizeOutput("dnd..spell.", `{"value":true}`)}, + NormalizeOutputs: []contracts.SerializedOutput{normalizeOutput("dnd..spell.", `{"value":true}`)}, }) if err != nil { t.Fatalf("Encode() error = %v, want nil", err) @@ -260,7 +260,7 @@ func TestEncodeSanitizesParentPathSequences(t *testing.T) { func TestEncodeRejectsInvalidJSONAndUnsupportedMediaTypes(t *testing.T) { tests := []struct { name string - output contracts.NormalizeOutput + output contracts.SerializedOutput want string }{ { @@ -270,9 +270,9 @@ func TestEncodeRejectsInvalidJSONAndUnsupportedMediaTypes(t *testing.T) { }, { name: "unsupported media type", - output: func() contracts.NormalizeOutput { + output: func() contracts.SerializedOutput { output := normalizeOutput("spells", `{"spell_casts":[]}`) - output.Payload.MediaType = "text/plain" + output.Artifact.MediaType = "text/plain" return output }(), want: "unsupported media type", @@ -282,7 +282,7 @@ func TestEncodeRejectsInvalidJSONAndUnsupportedMediaTypes(t *testing.T) { for _, test := range tests { t.Run(test.name, func(t *testing.T) { _, err := New().Encode(context.Background(), contracts.OutputRequest{ - NormalizeOutputs: []contracts.NormalizeOutput{test.output}, + NormalizeOutputs: []contracts.SerializedOutput{test.output}, }) if err == nil { t.Fatal("Encode() error = nil, want error") @@ -296,7 +296,7 @@ func TestEncodeRejectsInvalidJSONAndUnsupportedMediaTypes(t *testing.T) { func TestEncodeRejectsSanitizedFilenameCollisions(t *testing.T) { _, err := New().Encode(context.Background(), contracts.OutputRequest{ - NormalizeOutputs: []contracts.NormalizeOutput{ + NormalizeOutputs: []contracts.SerializedOutput{ normalizeOutput("a/b", `{"value":"slash"}`), normalizeOutput("a?b", `{"value":"question"}`), }, @@ -312,7 +312,7 @@ func TestEncodeRejectsSanitizedFilenameCollisions(t *testing.T) { func TestEncodeDoesNotMutateInputs(t *testing.T) { req := contracts.OutputRequest{ Manifest: artifacts.RunManifest{RunID: "run-1"}, - NormalizeOutputs: []contracts.NormalizeOutput{ + NormalizeOutputs: []contracts.SerializedOutput{ normalizeOutput("spells", `{"name":"original"}`), }, Rejected: []contracts.RejectedOutput{ @@ -331,8 +331,8 @@ func TestEncodeDoesNotMutateInputs(t *testing.T) { t.Fatalf("request mutated:\nbefore: %s\nafter: %s", before, after) } - req.NormalizeOutputs[0].Payload.Content[0] = '[' - req.NormalizeOutputs[0].Payload.Metadata["name"] = "changed" + req.NormalizeOutputs[0].Artifact.Content[0] = '[' + req.NormalizeOutputs[0].Artifact.Metadata["name"] = "changed" req.Rejected[0].Message = "changed" req.Warnings[0].Message = "changed" @@ -348,7 +348,7 @@ func TestEncodeDoesNotMutateInputs(t *testing.T) { func TestOutputFilesDoNotContainWarnings(t *testing.T) { result, err := New().Encode(context.Background(), contracts.OutputRequest{ - NormalizeOutputs: []contracts.NormalizeOutput{normalizeOutput("spells", `{"spell":"Shield"}`)}, + NormalizeOutputs: []contracts.SerializedOutput{normalizeOutput("spells", `{"spell":"Shield"}`)}, Warnings: []contracts.Warning{ {ReasonCode: "pipeline-warning", Message: "warning"}, }, @@ -363,17 +363,16 @@ func TestOutputFilesDoNotContainWarnings(t *testing.T) { } } -func normalizeOutput(laneID string, content string) contracts.NormalizeOutput { - return contracts.NormalizeOutput{ +func normalizeOutput(laneID string, content string) contracts.SerializedOutput { + return contracts.SerializedOutput{ LaneID: laneID, NormalizerKey: "noop", SourceID: "source-1", - Schema: contracts.ResponseSchema{ + Artifact: contracts.SerializedArtifact{Schema: contracts.ArtifactSchema{ ID: "schema-id", Name: "schema-name", Version: "v1", }, - Payload: contracts.RawPayload{ Content: []byte(content), MediaType: contentTypeJSON, Metadata: map[string]any{"name": laneID}, diff --git a/internal/modules/integration/dnd_spells_runner_test.go b/internal/modules/integration/dnd_spells_runner_test.go index 2a45643..c2e1a5f 100644 --- a/internal/modules/integration/dnd_spells_runner_test.go +++ b/internal/modules/integration/dnd_spells_runner_test.go @@ -61,10 +61,10 @@ func TestRunnerProcessesSeriatimInputWithDNDSpellsExtractor(t *testing.T) { t.Fatalf("len(NormalizeOutputs) = %d, want 1", len(output.NormalizeOutputs)) } rawOutput := output.NormalizeOutputs[0] - if rawOutput.LaneID != "spells" || rawOutput.Schema.ID != spells.ResponseSchemaID || rawOutput.Schema.Version != spells.SchemaVersion { + if rawOutput.LaneID != "spells" || rawOutput.Artifact.Schema.ID != spells.ResponseSchemaID || rawOutput.Artifact.Schema.Version != spells.SchemaVersion { t.Fatalf("raw output envelope = %#v, want dnd spells schema on spells lane", rawOutput) } - response := decodeRunnerSpellResponse(t, rawOutput.Payload.Content) + response := decodeRunnerSpellResponse(t, rawOutput.Artifact.Content) if len(response.SpellCasts) != 2 { t.Fatalf("len(spell_casts) = %d, want 2", len(response.SpellCasts)) } @@ -185,7 +185,7 @@ func TestRunnerDoesNotExtractSpellMentionedOnlyInPartyReference(t *testing.T) { if len(output.NormalizeOutputs) != 1 { t.Fatalf("len(NormalizeOutputs) = %d, want empty spell response output", len(output.NormalizeOutputs)) } - response := decodeRunnerSpellResponse(t, output.NormalizeOutputs[0].Payload.Content) + response := decodeRunnerSpellResponse(t, output.NormalizeOutputs[0].Artifact.Content) if len(response.SpellCasts) != 0 { t.Fatalf("spell_casts = %#v, want no party-reference-only spell casts", response.SpellCasts) } @@ -230,7 +230,7 @@ func TestRunnerCarriesDNDSpellCastWithInvalidSourceRefAsRawOutput(t *testing.T) if len(output.NormalizeOutputs) != 1 { t.Fatalf("len(NormalizeOutputs) = %d, want 1", len(output.NormalizeOutputs)) } - response := decodeRunnerSpellResponse(t, output.NormalizeOutputs[0].Payload.Content) + response := decodeRunnerSpellResponse(t, output.NormalizeOutputs[0].Artifact.Content) if len(response.SpellCasts) != 1 { t.Fatalf("len(spell_casts) = %d, want 1", len(response.SpellCasts)) } @@ -296,8 +296,8 @@ func TestRunnerCarriesMalformedDNDSpellsExtractorOutput(t *testing.T) { if len(output.NormalizeOutputs) != 1 { t.Fatalf("len(NormalizeOutputs) = %d, want raw output", len(output.NormalizeOutputs)) } - if string(output.NormalizeOutputs[0].Payload.Content) != `{"spell_casts":null}` { - t.Fatalf("content = %s, want canonical structured output", output.NormalizeOutputs[0].Payload.Content) + if string(output.NormalizeOutputs[0].Artifact.Content) != `{"spell_casts":null}` { + t.Fatalf("content = %s, want canonical structured output", output.NormalizeOutputs[0].Artifact.Content) } if output.Manifest.ValidationStatus != "approved" { t.Fatalf("ValidationStatus = %q, want approved", output.Manifest.ValidationStatus) diff --git a/internal/modules/seriatim/input/transcript/runner_test.go b/internal/modules/seriatim/input/transcript/runner_test.go index eb66d2f..c87b7c2 100644 --- a/internal/modules/seriatim/input/transcript/runner_test.go +++ b/internal/modules/seriatim/input/transcript/runner_test.go @@ -56,14 +56,14 @@ func TestRunnerProcessesSeriatimInputWithFakeModules(t *testing.T) { } rawOutput := output.NormalizeOutputs[0] - if rawOutput.LaneID != "events" || rawOutput.NormalizerKey != pipeline.DefaultNormalizeModule || rawOutput.Schema.ID != "fake.event" || rawOutput.Schema.Version != "v1" { + if rawOutput.LaneID != "events" || rawOutput.NormalizerKey != pipeline.DefaultNormalizeModule || rawOutput.Artifact.Schema.ID != "fake.event" || rawOutput.Artifact.Schema.Version != "v1" { t.Fatalf("raw output envelope = %#v, want fake extractor envelope", rawOutput) } var payload struct { Value string `json:"value"` SourceRefs []source.SourceRef `json:"source_refs"` } - if err := json.Unmarshal(rawOutput.Payload.Content, &payload); err != nil { + if err := json.Unmarshal(rawOutput.Artifact.Content, &payload); err != nil { t.Fatalf("Unmarshal(raw output) error = %v, want nil", err) } if len(payload.SourceRefs) != 1 {