diff --git a/docs/internal/state.md b/docs/internal/state.md index 68f18041..fbeaba60 100644 --- a/docs/internal/state.md +++ b/docs/internal/state.md @@ -64,7 +64,7 @@ Ordinary resume loads extract, merge, and normalize checkpoints progressively and may execute later lane stages after an earlier cache miss. Selective recomputation instead asks the loader for the required producer's accepted normalize artifact. That lookup reuses the existing normalize files, requires -workspace schema v3 plus an exact non-empty invocation identity, and deliberately +workspace schema v4 plus an exact non-empty invocation identity, and deliberately does not require extract or merge checkpoint files or dependency fingerprints. The runner performs canonical codec and producer-provenance validation before cloning the artifact into normal step output. Success restores only stored diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 2d15cc4b..62cca12b 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -323,7 +323,7 @@ remove the output-encoder consistency defect. This stage is appropriately sized for one `gpt-5.6-terra` prompt. -## Stage 4 — Version Diagnostic-Bearing Cache And Checkpoint State +## Stage 4 ✅ — Version Diagnostic-Bearing Cache And Checkpoint State ### Goal diff --git a/internal/framework/checkpoint/filesystem_test.go b/internal/framework/checkpoint/filesystem_test.go index 82fcff89..a755e079 100644 --- a/internal/framework/checkpoint/filesystem_test.go +++ b/internal/framework/checkpoint/filesystem_test.go @@ -36,6 +36,7 @@ func TestFilesystemCheckpointRoundTripsAllStages(t *testing.T) { fixture.doc.Metadata["owner"] = "caller mutation" fixture.extract.Artifact.Content[0] = 'x' fixture.extract.Artifact.Metadata["content"] = "caller mutation" + fixture.extract.Diagnostics[0].Diagnostic.Samples[0].Message = "caller mutation" fixture.merge.Artifact.Content[0] = 'x' fixture.normalize.Artifact.Content[0] = 'x' fixture.warnings[0].Message = "caller mutation" @@ -60,11 +61,11 @@ func TestFilesystemCheckpointRoundTripsAllStages(t *testing.T) { t.Run("extract", func(t *testing.T) { got, decision := fixture.loader.Extract("lane-a", "extract-module", fixture.dependencies) - if !decision.Reused || len(got.Outputs) != 1 || len(got.Rejected) != 1 || len(got.Warnings) != 1 { + if !decision.Reused || len(got.Outputs) != 1 || len(got.Rejected) != 1 || len(got.Warnings) != 1 || len(got.Outputs[0].Diagnostics) != 1 { t.Fatalf("extract result=%#v decision=%#v", got, decision) } output := got.Outputs[0] - if !bytes.Equal(output.Artifact.Content, []byte(`{"spell":"fire"}`)) || output.Artifact.Kind != "spell" || output.Artifact.Schema.ID != "spell-schema" || output.Artifact.Schema.Version != "1" || output.Artifact.MediaType != "application/json" || output.Artifact.Metadata["chunk"] != "chunk-a" || output.ChunkRef.StartUnitID != 1 || got.Warnings[0].ReasonCode != "partial" || got.Rejected[0].ReasonCode != "invalid_source" { + if !bytes.Equal(output.Artifact.Content, []byte(`{"spell":"fire"}`)) || output.Artifact.Kind != "spell" || output.Artifact.Schema.ID != "spell-schema" || output.Artifact.Schema.Version != "1" || output.Artifact.MediaType != "application/json" || output.Artifact.Metadata["chunk"] != "chunk-a" || output.ChunkRef.StartUnitID != 1 || output.Diagnostics[0].Diagnostic.ReasonCode != "normalized_record" || got.Warnings[0].ReasonCode != "partial" || got.Rejected[0].ReasonCode != "invalid_source" { t.Fatalf("extract values were not restored: %#v", got) } manifest := readManifest[ExtractLaneManifest](t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "extract", "lane-a", "manifest.json")) @@ -96,7 +97,7 @@ func TestFilesystemCheckpointRoundTripsAllStages(t *testing.T) { } { t.Run(tt.name, func(t *testing.T) { got, warnings, decision := tt.load() - if !decision.Reused || !bytes.Equal(got.Artifact.Content, tt.want) || got.Artifact.Kind != "spell" || got.Artifact.Schema.ID != "spell-schema" || got.Artifact.Schema.Version != "1" || got.Artifact.Metadata["lane"] != "lane-a" || len(warnings) != 1 || warnings[0].ReasonCode != "review" { + if !decision.Reused || !bytes.Equal(got.Artifact.Content, tt.want) || got.Artifact.Kind != "spell" || got.Artifact.Schema.ID != "spell-schema" || got.Artifact.Schema.Version != "1" || got.Artifact.Metadata["lane"] != "lane-a" || len(got.Diagnostics) != 1 || got.Diagnostics[0].Diagnostic.ReasonCode != "normalized_record" || len(warnings) != 1 || warnings[0].ReasonCode != "review" { t.Fatalf("%s result=%#v warnings=%#v decision=%#v", tt.name, got, warnings, decision) } @@ -159,6 +160,7 @@ func TestFilesystemCheckpointRejectsIncompatibleManifests(t *testing.T) { }{ {"v1 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV1 }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible}, {"v2 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV2 }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible}, + {"v3 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV3 }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible}, {"unknown schema", func(m map[string]any) { m["workspace_schema_version"] = "notarius.workspace.future" }, pipeline.CheckpointReasonWorkspaceSchemaIncompatible}, {"identity", func(m map[string]any) { m["metadata"].(map[string]any)["checkpoint_identity_digest"] = "sha256:other" }, pipeline.CheckpointReasonIdentityMismatch}, {"stage", func(m map[string]any) { m["stage"] = string(StageMerge) }, pipeline.CheckpointReasonStageMismatch}, @@ -204,6 +206,11 @@ func TestFilesystemCheckpointRejectsIncompleteArtifactsAndContent(t *testing.T) {"content digest", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["content"].(map[string]any)["content_digest"] = "sha256:other" }, pipeline.CheckpointReasonArtifactDigestMismatch}, + {"diagnostics", func(m map[string]any) { + m["outputs"].([]any)[0].(map[string]any)["diagnostics"] = []any{map[string]any{ + "diagnostic": map[string]any{"disposition": "warning", "category": "configuration", "reason_code": "invalid", "occurrence_count": float64(0)}, + }} + }, pipeline.CheckpointReasonArtifactPayloadInvalid}, } { t.Run(tt.name, func(t *testing.T) { fixture := seedFilesystemCheckpoints(t) @@ -527,7 +534,8 @@ func checkpointArtifact(module, content string) pipeline.CheckpointArtifact { return pipeline.CheckpointArtifact{ LaneID: "lane-a", ModuleKey: module, SourceID: "document-1", ChunkID: "chunk-a", ChunkIndex: 0, ChunkRef: source.SourceRef{SourceID: "document-1", StartUnitID: 1, EndUnitID: 1}, SchemaDigest: "sha256:schema", - Artifact: contracts.SerializedArtifact{Kind: "spell", Schema: contracts.ArtifactSchema{ID: "spell-schema", Name: "Spell", Version: "1", JSONSchema: []byte(`{"type":"object"}`)}, MediaType: "application/json", Content: []byte(content), Metadata: map[string]any{"chunk": "chunk-a", "lane": "lane-a"}}, + Artifact: contracts.SerializedArtifact{Kind: "spell", Schema: contracts.ArtifactSchema{ID: "spell-schema", Name: "Spell", Version: "1", JSONSchema: []byte(`{"type":"object"}`)}, MediaType: "application/json", Content: []byte(content), Metadata: map[string]any{"chunk": "chunk-a", "lane": "lane-a"}}, + Diagnostics: []pipeline.CheckpointDiagnostic{{Diagnostic: contracts.ProducerDiagnostic{Disposition: contracts.DiagnosticDispositionObservation, Category: contracts.DiagnosticCategoryNormalization, ReasonCode: "normalized_record", OccurrenceCount: 1, Samples: []contracts.DiagnosticSample{{Scope: "fixture", Message: "record normalized"}}}}}, } } diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index accfacbb..5426fab7 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -205,11 +205,33 @@ func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline. if strings.TrimSpace(string(v.Kind)) == "" || strings.TrimSpace(v.Schema.ID) == "" || strings.TrimSpace(v.Schema.Version) == "" || strings.TrimSpace(v.SchemaDigest) == "" { return nil, &artifactPayloadError{code: pipeline.CheckpointReasonArtifactCodecIncompatible, err: fmt.Errorf("artifact codec identity is incomplete")} } - out = append(out, pipeline.CheckpointArtifact{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)}}) + diagnostics, err := cloneAndValidateCheckpointDiagnostics(v.Diagnostics) + if err != nil { + return nil, err + } + out = append(out, pipeline.CheckpointArtifact{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)}, Diagnostics: diagnostics}) } return out, nil } +func cloneAndValidateCheckpointDiagnostics(values []pipeline.CheckpointDiagnostic) ([]pipeline.CheckpointDiagnostic, error) { + if len(values) == 0 { + return nil, nil + } + diagnostics := make([]contracts.ProducerDiagnostic, len(values)) + for index, value := range values { + diagnostics[index] = value.Diagnostic + } + if err := contracts.ValidateProducerDiagnostics(diagnostics); err != nil { + return nil, &artifactPayloadError{code: pipeline.CheckpointReasonArtifactPayloadInvalid, err: fmt.Errorf("checkpoint diagnostics: %w", err)} + } + cloned := make([]pipeline.CheckpointDiagnostic, len(values)) + for index, value := range values { + cloned[index] = pipeline.CheckpointDiagnostic{Diagnostic: contracts.CloneProducerDiagnostics([]contracts.ProducerDiagnostic{value.Diagnostic})[0], ValidatorKey: value.ValidatorKey} + } + return cloned, nil +} + func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision { if !l.Enabled() { return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLoadingDisabled) diff --git a/internal/framework/checkpoint/manifest.go b/internal/framework/checkpoint/manifest.go index 6b4d27a5..c97d2ccf 100644 --- a/internal/framework/checkpoint/manifest.go +++ b/internal/framework/checkpoint/manifest.go @@ -3,7 +3,8 @@ package checkpoint import "time" const ( - WorkspaceSchemaVersion = "notarius.workspace.v3" + WorkspaceSchemaVersion = "notarius.workspace.v4" + WorkspaceSchemaVersionV3 = "notarius.workspace.v3" WorkspaceSchemaVersionV2 = "notarius.workspace.v2" WorkspaceSchemaVersionV1 = "notarius.workspace.v1" ) diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go index 7bce66b7..f2cc713c 100644 --- a/internal/framework/checkpoint/recorder.go +++ b/internal/framework/checkpoint/recorder.go @@ -257,16 +257,17 @@ type binaryEnvelope struct { } 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"` + 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"` + Diagnostics []pipeline.CheckpointDiagnostic `json:"diagnostics,omitempty"` } type artifactExtractEnvelope struct { Outputs []artifactCheckpointEnvelope `json:"outputs"` @@ -281,7 +282,7 @@ type artifactSingleEnvelope struct { func artifactCheckpointEnvelopeFromOutput(output pipeline.CheckpointArtifact) 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)} + 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), Diagnostics: cloneCheckpointDiagnostics(output.Diagnostics)} } func artifactCheckpointEnvelopes(outputs []pipeline.CheckpointArtifact) []artifactCheckpointEnvelope { if len(outputs) == 0 { @@ -293,6 +294,20 @@ func artifactCheckpointEnvelopes(outputs []pipeline.CheckpointArtifact) []artifa } return out } + +func cloneCheckpointDiagnostics(values []pipeline.CheckpointDiagnostic) []pipeline.CheckpointDiagnostic { + if len(values) == 0 { + return nil + } + cloned := make([]pipeline.CheckpointDiagnostic, len(values)) + for index, value := range values { + cloned[index] = pipeline.CheckpointDiagnostic{ + Diagnostic: contracts.CloneProducerDiagnostics([]contracts.ProducerDiagnostic{value.Diagnostic})[0], + ValidatorKey: value.ValidatorKey, + } + } + return cloned +} func artifactOutputDigests(outputs []pipeline.CheckpointArtifact) []pipeline.CheckpointFingerprint { values := make([]pipeline.CheckpointFingerprint, 0, len(outputs)) for i, v := range outputs { diff --git a/internal/framework/checkpoint/recorder_test.go b/internal/framework/checkpoint/recorder_test.go index 8ee15d20..769411d5 100644 --- a/internal/framework/checkpoint/recorder_test.go +++ b/internal/framework/checkpoint/recorder_test.go @@ -119,7 +119,7 @@ func TestStepAwareCheckpointPreservesDistinctDotIdentities(t *testing.T) { } func TestCheckpointSchemaCompatibilityIdentifiers(t *testing.T) { - if WorkspaceSchemaVersion != "notarius.workspace.v3" || WorkspaceSchemaVersionV2 != "notarius.workspace.v2" || WorkspaceSchemaVersionV1 != "notarius.workspace.v1" { + if WorkspaceSchemaVersion != "notarius.workspace.v4" || WorkspaceSchemaVersionV3 != "notarius.workspace.v3" || WorkspaceSchemaVersionV2 != "notarius.workspace.v2" || WorkspaceSchemaVersionV1 != "notarius.workspace.v1" { t.Fatal("checkpoint schema identifiers are incorrect") } } diff --git a/internal/framework/chunkplan/store.go b/internal/framework/chunkplan/store.go index 57861a26..817962a8 100644 --- a/internal/framework/chunkplan/store.go +++ b/internal/framework/chunkplan/store.go @@ -14,6 +14,7 @@ import ( "syscall" "gitea.maximumdirect.net/eric/notarius/internal/core/source" + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" ) @@ -340,6 +341,9 @@ func validateRecord(record pipeline.ChunkPlanRecord, requestedDigest string) err if record.SchemaVersion != SchemaVersion { return fmt.Errorf("schema_version %q is not supported", record.SchemaVersion) } + if err := contracts.ValidateProducerDiagnostics(record.Diagnostics); err != nil { + return fmt.Errorf("diagnostics: %w", err) + } if _, err := digestPathSegment(requestedDigest); err != nil { return err } diff --git a/internal/framework/chunkplan/store_test.go b/internal/framework/chunkplan/store_test.go index 72b1e54c..dcdca15d 100644 --- a/internal/framework/chunkplan/store_test.go +++ b/internal/framework/chunkplan/store_test.go @@ -281,7 +281,7 @@ func TestFilesystemStoreReportsInvalidRecordsAsRecoverable(t *testing.T) { return bytes.Replace(data, []byte(`{"schema_version"`), []byte(`{"SENTINEL_UNKNOWN_FIELD":true,"schema_version"`), 1) }}, {name: "truncated JSON", mutate: func(data []byte) []byte { return data[:len(data)/2] }}, - {name: "legacy v1 record", mutate: replaceJSON(`notarius.chunk-plan.v2`, `notarius.chunk-plan.v1`)}, + {name: "legacy v2 record", mutate: replaceJSON(`notarius.chunk-plan.v3`, `notarius.chunk-plan.v2`)}, {name: "source mismatch", mutate: replaceJSON(testSourceDigest, "sha256:"+strings.Repeat("b", 64))}, {name: "plan digest mismatch", mutate: func(data []byte) []byte { prefix := []byte(`"plan_digest":"sha256:`) @@ -498,7 +498,14 @@ func testRecord(t *testing.T, value int) pipeline.ChunkPlanRecord { References: []artifacts.ReferenceProvenance{{Stage: "chunk", SlotName: "guide", OriginType: "file", OriginURI: "file:///guide.txt", Digest: "sha256:reference"}}, Metadata: map[string]any{"prompt_id": "test/prompt", "enabled": true}, }, - Warnings: []contracts.Warning{{Scope: "chunk/test", ReasonCode: "observed", Message: "warning"}}, + Warnings: []contracts.Warning{{Scope: "chunk/test", ReasonCode: "observed", Message: "warning"}}, + Diagnostics: []contracts.ProducerDiagnostic{{ + Disposition: contracts.DiagnosticDispositionWarning, + Category: contracts.DiagnosticCategoryConfiguration, + ReasonCode: "empty_reference", + OccurrenceCount: 1, + Samples: []contracts.DiagnosticSample{{Scope: "reference", Message: "Reference was empty."}}, + }}, CreatedAt: time.Date(2026, 7, 18, 12, 0, 0, 0, time.UTC), } } diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 70d1f64e..73b6ce14 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -279,6 +279,15 @@ type CheckpointArtifact struct { ChunkRef source.SourceRef Artifact contracts.SerializedArtifact SchemaDigest string + Diagnostics []CheckpointDiagnostic +} + +// CheckpointDiagnostic stores a producer-local diagnostic alongside a +// reusable artifact. The runner supplies the current run's origin when it +// promotes this value into a diagnostic group. +type CheckpointDiagnostic struct { + Diagnostic contracts.ProducerDiagnostic `json:"diagnostic"` + ValidatorKey string `json:"validator_key,omitempty"` } type ExtractCheckpoint struct { diff --git a/internal/framework/pipeline/chunk_plan_store.go b/internal/framework/pipeline/chunk_plan_store.go index 9023e56b..cc6a7c51 100644 --- a/internal/framework/pipeline/chunk_plan_store.go +++ b/internal/framework/pipeline/chunk_plan_store.go @@ -8,7 +8,7 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) -const ChunkPlanSchemaVersion = "notarius.chunk-plan.v2" +const ChunkPlanSchemaVersion = "notarius.chunk-plan.v3" type ChunkPlanProducer struct { InputModule string `json:"input_module"` @@ -19,13 +19,14 @@ type ChunkPlanProducer struct { } type ChunkPlanRecord struct { - SchemaVersion string `json:"schema_version"` - SourceDigest string `json:"source_digest"` - PlanDigest string `json:"plan_digest"` - Plan source.ChunkPlan `json:"plan"` - Producer ChunkPlanProducer `json:"producer"` - Warnings []contracts.Warning `json:"warnings,omitempty"` - CreatedAt time.Time `json:"created_at"` + SchemaVersion string `json:"schema_version"` + SourceDigest string `json:"source_digest"` + PlanDigest string `json:"plan_digest"` + Plan source.ChunkPlan `json:"plan"` + Producer ChunkPlanProducer `json:"producer"` + Warnings []contracts.Warning `json:"warnings,omitempty"` + Diagnostics []contracts.ProducerDiagnostic `json:"diagnostics,omitempty"` + CreatedAt time.Time `json:"created_at"` } type ChunkPlanStore interface { diff --git a/internal/framework/pipeline/diagnostics.go b/internal/framework/pipeline/diagnostics.go index 0833e696..cdd8e7f8 100644 --- a/internal/framework/pipeline/diagnostics.go +++ b/internal/framework/pipeline/diagnostics.go @@ -9,33 +9,53 @@ import ( ) func terminalDiagnosticGroups(terminal producerAttemptTerminal, origin contracts.DiagnosticOrigin, chunk *source.Chunk) ([]contracts.DiagnosticGroup, error) { - groups, err := promoteProducerDiagnostics(terminal.Diagnostics, origin, chunk) - if err != nil { - return nil, err + return promoteCheckpointDiagnostics(terminalCheckpointDiagnostics(terminal), origin, chunk) +} + +func terminalCheckpointDiagnostics(terminal producerAttemptTerminal) []CheckpointDiagnostic { + diagnostics := make([]CheckpointDiagnostic, 0, len(terminal.Diagnostics)) + for _, diagnostic := range terminal.Diagnostics { + diagnostics = append(diagnostics, CheckpointDiagnostic{Diagnostic: diagnostic}) } for _, record := range terminal.Validation.Diagnostics() { - validatorOrigin := origin - validatorOrigin.ValidatorKey = record.validatorName - promoted, err := promoteProducerDiagnostics([]contracts.ProducerDiagnostic{record.diagnostic}, validatorOrigin, chunk) - if err != nil { - return nil, fmt.Errorf("validator %q diagnostic: %w", record.validatorName, err) - } - groups = append(groups, promoted...) + diagnostics = append(diagnostics, CheckpointDiagnostic{Diagnostic: record.diagnostic, ValidatorKey: record.validatorName}) } if terminal.Action == producerTerminalIncompleteAccepted { for _, record := range incompleteValidationDiagnostics(terminal.Validation) { - validatorOrigin := origin - validatorOrigin.ValidatorKey = record.validatorName - promoted, err := promoteProducerDiagnostics([]contracts.ProducerDiagnostic{record.diagnostic}, validatorOrigin, chunk) - if err != nil { - return nil, fmt.Errorf("validator %q incomplete diagnostic: %w", record.validatorName, err) - } - groups = append(groups, promoted...) + diagnostics = append(diagnostics, CheckpointDiagnostic{Diagnostic: record.diagnostic, ValidatorKey: record.validatorName}) } } + return cloneCheckpointDiagnostics(diagnostics) +} + +func promoteCheckpointDiagnostics(diagnostics []CheckpointDiagnostic, origin contracts.DiagnosticOrigin, chunk *source.Chunk) ([]contracts.DiagnosticGroup, error) { + groups := make([]contracts.DiagnosticGroup, 0, len(diagnostics)) + for _, record := range diagnostics { + validatorOrigin := origin + validatorOrigin.ValidatorKey = record.ValidatorKey + promoted, err := promoteProducerDiagnostics([]contracts.ProducerDiagnostic{record.Diagnostic}, validatorOrigin, chunk) + if err != nil { + return nil, fmt.Errorf("checkpoint diagnostic: %w", err) + } + groups = append(groups, promoted...) + } return groups, nil } +func cloneCheckpointDiagnostics(diagnostics []CheckpointDiagnostic) []CheckpointDiagnostic { + if len(diagnostics) == 0 { + return nil + } + cloned := make([]CheckpointDiagnostic, len(diagnostics)) + for index, diagnostic := range diagnostics { + cloned[index] = CheckpointDiagnostic{ + Diagnostic: contracts.CloneProducerDiagnostics([]contracts.ProducerDiagnostic{diagnostic.Diagnostic})[0], + ValidatorKey: diagnostic.ValidatorKey, + } + } + return cloned +} + func promoteProducerDiagnostics(diagnostics []contracts.ProducerDiagnostic, origin contracts.DiagnosticOrigin, chunk *source.Chunk) ([]contracts.DiagnosticGroup, error) { if err := contracts.ValidateProducerDiagnostics(diagnostics); err != nil { return nil, err diff --git a/internal/framework/pipeline/runner_chunk_plan.go b/internal/framework/pipeline/runner_chunk_plan.go index ba660d64..836c7d80 100644 --- a/internal/framework/pipeline/runner_chunk_plan.go +++ b/internal/framework/pipeline/runner_chunk_plan.go @@ -85,7 +85,12 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S result.warnings = append(cloneWarnings(record.Warnings), report.Warnings()...) result.accepted = true result.setValidation(report.Warnings(), nil, nil) - cachedTerminal := producerAttemptTerminal{Action: producerTerminalAccepted, Validation: report} + cachedTerminal := producerAttemptTerminal{Action: producerTerminalAccepted, Diagnostics: contracts.CloneProducerDiagnostics(record.Diagnostics), Validation: report} + diagnostics, diagnosticErr := terminalDiagnosticGroups(cachedTerminal, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageChunk, ModuleKey: chunker.Key()}, nil) + if diagnosticErr != nil { + return result, fmt.Errorf("promote reused chunk diagnostics: %w", diagnosticErr) + } + result.diagnostics = diagnostics summary := validationSummary(cachedTerminal, StageChunk, "", "", chunker.Key(), "", 0) result.validation = &summary return result, nil @@ -242,6 +247,7 @@ func (r *Runner) runChunkPlan(ctx context.Context, input RunInput, doc *source.S return result, fmt.Errorf("clone chunk plan record for publication: %w", cloneErr) } record.Warnings = cloneWarnings(candidate.producerWarnings) + record.Diagnostics = contracts.CloneProducerDiagnostics(candidate.producerDiagnostics) if err := input.ChunkPlans.Save(record); err != nil { return result, fmt.Errorf("save chunk plan: %w", err) } @@ -319,6 +325,7 @@ func cloneChunkPlanRecord(record ChunkPlanRecord) (ChunkPlanRecord, error) { } record.Producer.Metadata = metadata record.Warnings = cloneWarnings(record.Warnings) + record.Diagnostics = contracts.CloneProducerDiagnostics(record.Diagnostics) return record, nil } diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go index 33142562..c3fcaa38 100644 --- a/internal/framework/pipeline/runner_chunk_plan_test.go +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -233,6 +233,37 @@ func TestRunnerChunkPlanHitUsesStoredProducerProvenance(t *testing.T) { } } +func TestRunnerReusesChunkPlanDiagnosticsWithCurrentValidatorDiagnostics(t *testing.T) { + prepared, plan := preparedTerminalDebugPipeline(t) + record := chunkPlanRecord(t, prepared, plan) + record.Diagnostics = []contracts.ProducerDiagnostic{{ + Disposition: contracts.DiagnosticDispositionWarning, + Category: contracts.DiagnosticCategoryConfiguration, + ReasonCode: "stored_chunk_diagnostic", + OccurrenceCount: 1, + Samples: []contracts.DiagnosticSample{{Scope: "reference", Message: "Stored configuration signal."}}, + }} + validator := &countingChunkValidator{result: contracts.ValidationResult{Approved: true, Diagnostics: []contracts.ProducerDiagnostic{{ + Disposition: contracts.DiagnosticDispositionAdvisory, + Category: contracts.DiagnosticCategoryDataQuality, + ReasonCode: "current_validator_diagnostic", + OccurrenceCount: 1, + Samples: []contracts.DiagnosticSample{{Scope: "chunk", Message: "Current validator finding."}}, + }}}} + prepared.chunkValidators.validators = []preparedValidator{{resolved: ResolvedValidator{Binding: Binding(validator.Name()), Target: ValidatorTargetChunk}, chunk: validator}} + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ChunkCacheMode: ChunkCacheAuto, ChunkPlans: &recordingChunkPlanStore{record: record, decision: ChunkPlanDecision{Status: ChunkPlanHit}}}) + if err != nil { + t.Fatal(err) + } + if validator.calls != 1 || len(output.Diagnostics.Groups) != 2 { + t.Fatalf("validator calls = %d diagnostics = %#v", validator.calls, output.Diagnostics) + } + if output.Diagnostics.Groups[0].ReasonCode != "stored_chunk_diagnostic" || output.Diagnostics.Groups[1].ReasonCode != "current_validator_diagnostic" || output.Diagnostics.Groups[1].Origin.ValidatorKey != validator.Name() { + t.Fatalf("diagnostic groups = %#v", output.Diagnostics.Groups) + } +} + func TestRunnerProvidesAcceptedChunkMapToOutput(t *testing.T) { for _, test := range []struct { name string diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 64822dbb..58b4aa40 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -357,6 +357,11 @@ func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.So } hydrated := resolution.artifacts[0] local.Warnings = append(local.Warnings, cloneWarnings(checkpoint.Warnings)...) + diagnostics, diagnosticErr := promoteCheckpointDiagnostics(hydrated.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageNormalize, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module}, nil) + if diagnosticErr != nil { + return state, fmt.Errorf("promote reused accepted diagnostics: %w", diagnosticErr) + } + appendDiagnosticGroups(&local, diagnostics) local.NormalizeOutputs = append(local.NormalizeOutputs, contracts.SerializedOutput{ StepID: input.stepID, LaneID: lane.ID, @@ -421,6 +426,12 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou } state.values = append(state.values, artifact) state.serialized = append(state.serialized, cloneCheckpointArtifact(stored)) + chunk := source.Chunk{ID: stored.ChunkID, Index: stored.ChunkIndex} + diagnostics, diagnosticErr := promoteCheckpointDiagnostics(stored.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageExtract, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module}, &chunk) + if diagnosticErr != nil { + return nil, fmt.Errorf("promote reused extract diagnostics: %w", diagnosticErr) + } + state.diagnostics = append(state.diagnostics, diagnostics...) } state.warnings, state.rejected = cloneWarnings(cp.Warnings), cloneRejectedOutputs(cp.Rejected) } @@ -516,6 +527,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. return result } stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = candidate.artifact.ChunkID, candidate.artifact.ChunkIndex, candidate.artifact.ChunkRef + stored.Diagnostics = terminalCheckpointDiagnostics(terminalResult) if debugErr := candidate.terminal.record(payload, nil); debugErr != nil { result.err = debugErr return result diff --git a/internal/framework/pipeline/runner_extract_handoff_test.go b/internal/framework/pipeline/runner_extract_handoff_test.go index 92be174e..fb923cbc 100644 --- a/internal/framework/pipeline/runner_extract_handoff_test.go +++ b/internal/framework/pipeline/runner_extract_handoff_test.go @@ -158,6 +158,13 @@ func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) { return erasedTypedResult{ Value: typedValueForLane(0, request.Chunk.Index), Warnings: []contracts.Warning{{Scope: "extract", ReasonCode: "observed", Message: "accepted extract"}}, + Diagnostics: []contracts.ProducerDiagnostic{{ + Disposition: contracts.DiagnosticDispositionObservation, + Category: contracts.DiagnosticCategoryNormalization, + ReasonCode: "accepted_extract_normalized", + OccurrenceCount: 1, + Samples: []contracts.DiagnosticSample{{Scope: "extract", Message: "accepted extract"}}, + }}, }, nil }) @@ -209,6 +216,9 @@ func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) { if !reflect.DeepEqual(reused.Warnings, fresh.Warnings) { t.Fatalf("reused warnings = %#v, want fresh warnings %#v", reused.Warnings, fresh.Warnings) } + if !reflect.DeepEqual(reused.Diagnostics, fresh.Diagnostics) { + t.Fatalf("reused diagnostics = %#v, want fresh diagnostics %#v", reused.Diagnostics, fresh.Diagnostics) + } } func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) { diff --git a/internal/framework/pipeline/runner_terminal_debug_test.go b/internal/framework/pipeline/runner_terminal_debug_test.go index 52440a2e..18300457 100644 --- a/internal/framework/pipeline/runner_terminal_debug_test.go +++ b/internal/framework/pipeline/runner_terminal_debug_test.go @@ -14,11 +14,12 @@ import ( ) type terminalChunker struct { - key string - plan source.ChunkPlan - warnings []contracts.Warning - err error - calls *int + key string + plan source.ChunkPlan + warnings []contracts.Warning + diagnostics []contracts.ProducerDiagnostic + err error + calls *int } func (c terminalChunker) Key() string { return c.key } @@ -29,7 +30,7 @@ func (c terminalChunker) Plan(context.Context, contracts.ChunkRequest) (contract if c.calls != nil { (*c.calls)++ } - return contracts.ChunkPlanResult{Plan: source.CloneChunkPlan(c.plan), Warnings: cloneWarnings(c.warnings)}, c.err + return contracts.ChunkPlanResult{Plan: source.CloneChunkPlan(c.plan), Warnings: cloneWarnings(c.warnings), Diagnostics: contracts.CloneProducerDiagnostics(c.diagnostics)}, c.err } type terminalChunkValidator struct { diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 63111957..4b99d322 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -37,6 +37,7 @@ func recordNormalize(recorder CheckpointRecorder, stepID, laneID, moduleKey stri } func cloneCheckpointArtifact(output CheckpointArtifact) CheckpointArtifact { output.Artifact = contracts.CloneSerializedArtifact(output.Artifact) + output.Diagnostics = cloneCheckpointDiagnostics(output.Diagnostics) return output } @@ -270,6 +271,11 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints serializedMerge = mergeResolution.artifacts[0] mergeWarnings = cloneWarnings(mergeCP.Warnings) output.Warnings = append(output.Warnings, mergeWarnings...) + diagnostics, diagnosticErr := promoteCheckpointDiagnostics(serializedMerge.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageMerge, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module}, nil) + if diagnosticErr != nil { + return stageResult, fmt.Errorf("promote reused merge diagnostics: %w", diagnosticErr) + } + appendDiagnosticGroups(output, diagnostics) } else { if stageResult.reuseEligible { if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil { @@ -358,6 +364,7 @@ func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints } return stageResult, candidate.terminal.record(payload, attemptErr) } + stored.Diagnostics = terminalCheckpointDiagnostics(terminalResult) if debugErr := candidate.terminal.record(payload, nil); debugErr != nil { if stageResult.reuseEligible { _ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, debugErr) @@ -420,6 +427,11 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi serializedNormalize = normalizeResolution.artifacts[0] normalizeWarnings = cloneWarnings(normalizeCP.Warnings) output.Warnings = append(output.Warnings, normalizeWarnings...) + diagnostics, diagnosticErr := promoteCheckpointDiagnostics(serializedNormalize.Diagnostics, contracts.DiagnosticOrigin{Stage: contracts.DiagnosticOriginStageNormalize, StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module}, nil) + if diagnosticErr != nil { + return stageResult, fmt.Errorf("promote reused normalize diagnostics: %w", diagnosticErr) + } + appendDiagnosticGroups(output, diagnostics) } else { if stageResult.reuseEligible { if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil { @@ -528,6 +540,7 @@ func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoi } return stageResult, candidate.terminal.record(payload, attemptErr) } + stored.Diagnostics = terminalCheckpointDiagnostics(terminalResult) if debugErr := candidate.terminal.record(payload, nil); debugErr != nil { if stageResult.reuseEligible { _ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, debugErr)