diff --git a/docs/internal/state.md b/docs/internal/state.md index 18fa305..5333c80 100644 --- a/docs/internal/state.md +++ b/docs/internal/state.md @@ -43,6 +43,17 @@ not permit stale downstream reuse. The CLI's selective recomputation policy records `forced_recompute` for the selected step and its dependents while requiring compatible predecessor checkpoints. +The loader assigns a typed category and reason code at each validation site; +diagnostic prose is not classified after the fact. The runner then applies +forced-execution policy, validates reusable artifact bytes through the prepared +codec, and records the final decision before enforcing a required-predecessor +failure. That failure names only the step, lane, and stable reason code. Decision +detail passes through one UTF-8-safe bounded sanitizer and contains only +allowlisted diagnostic context, never payloads, references, credentials, +environment values, or physical paths. Typed categories and codes remain intact +through pipeline events and become strings only in manifest and debug-summary +JSON. + `internal/core/fileio` provides confined atomic file writes used by state collaborators. The chunk-plan store retains its stronger entry validation. diff --git a/docs/operations.md b/docs/operations.md index 7339b05..1fa02ee 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -173,6 +173,33 @@ then inspect checkpoint decision categories and reason codes. A `dependency_invalidated` decision means the stored producer, codec identity, schema, or canonical content no longer matches; a missing or rejected producer requires rerunning it rather than copying an artifact into the checkpoint root. +The decision that caused a required-predecessor failure is retained before the +run returns, and the CLI error identifies its step, lane, and reason code. + +Checkpoint reason codes are stable diagnostic identifiers: + +| Reason code | Operator meaning | +| --- | --- | +| `loading_disabled` | This invocation did not enable checkpoint loading. | +| `checkpoint_missing` | The requested checkpoint file does not exist. | +| `checkpoint_path_invalid` | The requested checkpoint location failed confinement validation. | +| `checkpoint_read_failed` | An existing checkpoint could not be read. | +| `checkpoint_decode_failed` | Checkpoint JSON could not be decoded. | +| `workspace_schema_incompatible` | The stored workspace schema is not supported by this build. | +| `identity_mismatch` | The stored invocation identity differs from the current invocation. | +| `stage_mismatch`, `step_mismatch`, `lane_mismatch`, `module_mismatch` | Stored scope does not match the requested pipeline scope. | +| `status_not_reusable` | The stored operation did not finish in a reusable status. | +| `dependency_mismatch` | Stored dependencies differ; the category is `dependency_invalidated`. | +| `artifact_payload_invalid` | Stored artifact payload structure or encoding is invalid. | +| `artifact_digest_mismatch` | Stored artifact bytes do not match their recorded digest. | +| `artifact_codec_incompatible` | Stored artifact identity is incomplete or incompatible with the codec contract. | +| `artifact_not_canonical` | The codec can decode the artifact, but its bytes are not canonical. | +| `checkpoint_reused` | The stored checkpoint passed validation and was reused. | +| `recompute_step` | Selective recomputation forced execution of this lane. | + +Decision detail is bounded explanatory text, not a data-recovery channel. It +never contains checkpoint paths, artifact or reference content, source content, +credentials, or environment values. ## Debug Bundles diff --git a/internal/framework/checkpoint/filesystem_test.go b/internal/framework/checkpoint/filesystem_test.go index 117b804..fef4ae0 100644 --- a/internal/framework/checkpoint/filesystem_test.go +++ b/internal/framework/checkpoint/filesystem_test.go @@ -8,6 +8,7 @@ import ( "runtime" "strings" "testing" + "unicode/utf8" "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" @@ -137,7 +138,7 @@ func TestFilesystemCheckpointRejectsMissingAndCorruptState(t *testing.T) { if err := os.Remove(stage.manifest(fixture)); err != nil { t.Fatal(err) } - assertStageNotReused(t, stage, fixture, "missing") + assertStageNotReused(t, stage, fixture, pipeline.CheckpointReasonMissing) }) t.Run(stage.name+" malformed manifest", func(t *testing.T) { @@ -145,7 +146,7 @@ func TestFilesystemCheckpointRejectsMissingAndCorruptState(t *testing.T) { if err := os.WriteFile(stage.manifest(fixture), []byte("{"), 0o600); err != nil { t.Fatal(err) } - assertStageNotReused(t, stage, fixture, "decode") + assertStageNotReused(t, stage, fixture, pipeline.CheckpointReasonDecodeFailed) }) } } @@ -154,18 +155,18 @@ func TestFilesystemCheckpointRejectsIncompatibleManifests(t *testing.T) { for _, tt := range []struct { name string edit func(map[string]any) - want string + want pipeline.CheckpointReasonCode }{ - {"v1 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV1 }, "workspace schema"}, - {"v2 schema", func(m map[string]any) { m["workspace_schema_version"] = WorkspaceSchemaVersionV2 }, "workspace schema"}, - {"unknown schema", func(m map[string]any) { m["workspace_schema_version"] = "notarius.workspace.future" }, "workspace schema"}, - {"identity", func(m map[string]any) { m["metadata"].(map[string]any)["checkpoint_identity_digest"] = "sha256:other" }, "identity"}, - {"stage", func(m map[string]any) { m["stage"] = string(StageMerge) }, "stage"}, - {"lane", func(m map[string]any) { m["lane_id"] = "lane-other" }, "lane"}, - {"module", func(m map[string]any) { m["module_key"] = "module-other" }, "module"}, + {"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}, + {"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}, + {"lane", func(m map[string]any) { m["lane_id"] = "lane-other" }, pipeline.CheckpointReasonLaneMismatch}, + {"module", func(m map[string]any) { m["module_key"] = "module-other" }, pipeline.CheckpointReasonModuleMismatch}, {"dependency", func(m map[string]any) { m["dependency_fingerprints"] = []map[string]string{{"name": "input", "value": "other"}} - }, "dependency"}, + }, pipeline.CheckpointReasonDependencyMismatch}, } { t.Run(tt.name, func(t *testing.T) { fixture := seedFilesystemCheckpoints(t) @@ -180,7 +181,7 @@ func TestFilesystemCheckpointRejectsNonTerminalStatuses(t *testing.T) { t.Run(string(status), func(t *testing.T) { fixture := seedFilesystemCheckpoints(t) editManifest(t, checkpointStages()[1].manifest(fixture), func(m map[string]any) { m["status"] = string(status) }) - assertStageNotReused(t, checkpointStages()[1], fixture, "status") + assertStageNotReused(t, checkpointStages()[1], fixture, pipeline.CheckpointReasonStatusNotReusable) }) } } @@ -189,20 +190,20 @@ func TestFilesystemCheckpointRejectsIncompleteArtifactsAndContent(t *testing.T) for _, tt := range []struct { name string edit func(map[string]any) - want string + want pipeline.CheckpointReasonCode }{ - {"artifact kind", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["artifact_kind"] = "" }, "artifact codec identity"}, - {"schema id", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["schema"].(map[string]any)["id"] = "" }, "artifact codec identity"}, + {"artifact kind", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["artifact_kind"] = "" }, pipeline.CheckpointReasonArtifactCodecIncompatible}, + {"schema id", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["schema"].(map[string]any)["id"] = "" }, pipeline.CheckpointReasonArtifactCodecIncompatible}, {"schema version", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["schema"].(map[string]any)["version"] = "" - }, "artifact codec identity"}, - {"schema digest", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["schema_digest"] = "" }, "artifact codec identity"}, + }, pipeline.CheckpointReasonArtifactCodecIncompatible}, + {"schema digest", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["schema_digest"] = "" }, pipeline.CheckpointReasonArtifactCodecIncompatible}, {"base64", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["content"].(map[string]any)["content_base64"] = "%" - }, "base64"}, + }, pipeline.CheckpointReasonArtifactPayloadInvalid}, {"content digest", func(m map[string]any) { m["outputs"].([]any)[0].(map[string]any)["content"].(map[string]any)["content_digest"] = "sha256:other" - }, "content digest"}, + }, pipeline.CheckpointReasonArtifactDigestMismatch}, } { t.Run(tt.name, func(t *testing.T) { fixture := seedFilesystemCheckpoints(t) @@ -218,7 +219,7 @@ func TestFilesystemCheckpointRejectsSourceAndOutputDigestMismatches(t *testing.T editJSON(t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "source", "source-document.json"), func(m map[string]any) { m["document"].(map[string]any)["units"].([]any)[0].(map[string]any)["text"] = "" }) - assertStageNotReused(t, checkpointStages()[0], fixture, "source checkpoint document") + assertStageNotReused(t, checkpointStages()[0], fixture, pipeline.CheckpointReasonArtifactPayloadInvalid) }) t.Run("source output digest", func(t *testing.T) { @@ -226,7 +227,7 @@ func TestFilesystemCheckpointRejectsSourceAndOutputDigestMismatches(t *testing.T editJSON(t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "source", "source-document.json"), func(m map[string]any) { m["document"].(map[string]any)["digest"] = "sha256:other" }) - assertStageNotReused(t, checkpointStages()[0], fixture, "output digest") + assertStageNotReused(t, checkpointStages()[0], fixture, pipeline.CheckpointReasonArtifactDigestMismatch) }) for _, stage := range checkpointStages()[1:] { @@ -235,7 +236,7 @@ func TestFilesystemCheckpointRejectsSourceAndOutputDigestMismatches(t *testing.T editManifest(t, stage.manifest(fixture), func(m map[string]any) { m["output_digests"].([]any)[0].(map[string]any)["value"] = "sha256:other" }) - assertStageNotReused(t, stage, fixture, "output digest") + assertStageNotReused(t, stage, fixture, pipeline.CheckpointReasonArtifactDigestMismatch) }) } } @@ -259,6 +260,68 @@ func TestFilesystemCheckpointDependencyDecisionIsBoundedAndCategorized(t *testin } } +func TestFilesystemCheckpointDecisionFamiliesAreStableAndSafe(t *testing.T) { + const secretSentinel = "do-not-expose-checkpoint-secret" + tests := []struct { + name string + prepare func(*testing.T, filesystemCheckpointFixture) pipeline.CheckpointDecision + category pipeline.CheckpointDecisionCategory + code pipeline.CheckpointReasonCode + }{ + {"loading disabled", func(t *testing.T, _ filesystemCheckpointFixture) pipeline.CheckpointDecision { + _, decision := pipeline.NoopCheckpointLoader().Source("source") + return decision + }, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLoadingDisabled}, + {"checkpoint unavailable", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision { + if err := os.Remove(checkpointStages()[1].manifest(fixture)); err != nil { + t.Fatal(err) + } + return checkpointStages()[1].load(fixture) + }, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonMissing}, + {"workspace identity", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision { + editManifest(t, checkpointStages()[1].manifest(fixture), func(m map[string]any) { + m["workspace_schema_version"] = secretSentinel + }) + return checkpointStages()[1].load(fixture) + }, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible}, + {"manifest scope", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision { + editManifest(t, checkpointStages()[1].manifest(fixture), func(m map[string]any) { m["module_key"] = secretSentinel }) + return checkpointStages()[1].load(fixture) + }, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch}, + {"dependency invalidated", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision { + _, decision := fixture.loader.Extract("lane-a", "extract-module", []pipeline.CheckpointFingerprint{{Name: "input", Value: secretSentinel}}) + return decision + }, pipeline.CheckpointDecisionDependencyInvalidated, pipeline.CheckpointReasonDependencyMismatch}, + {"artifact payload", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision { + editJSON(t, filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), "extract", "lane-a", "outputs.json"), func(m map[string]any) { + m["outputs"].([]any)[0].(map[string]any)["artifact_kind"] = "" + m["outputs"].([]any)[0].(map[string]any)["source_id"] = secretSentinel + }) + return checkpointStages()[1].load(fixture) + }, pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactCodecIncompatible}, + {"checkpoint reused", func(t *testing.T, fixture filesystemCheckpointFixture) pipeline.CheckpointDecision { + return checkpointStages()[1].load(fixture) + }, pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fixture := seedFilesystemCheckpoints(t) + decision := test.prepare(t, fixture) + if decision.Category != test.category || decision.ReasonCode != test.code { + t.Fatalf("decision = %#v, want category %q and code %q", decision, test.category, test.code) + } + if len([]byte(decision.Detail)) > 512 || !utf8.ValidString(decision.Detail) { + t.Fatalf("decision detail is not bounded valid UTF-8: %#v", decision) + } + for _, forbidden := range []string{secretSentinel, fixture.root} { + if strings.Contains(decision.Detail, forbidden) || strings.Contains(decision.Reason, forbidden) { + t.Fatalf("decision leaked %q: %#v", forbidden, decision) + } + } + }) + } +} + type checkpointStage struct { name string manifest func(filesystemCheckpointFixture) string @@ -367,11 +430,11 @@ func checkpointArtifact(module, content string) pipeline.CheckpointArtifact { } } -func assertStageNotReused(t *testing.T, stage checkpointStage, fixture filesystemCheckpointFixture, want string) { +func assertStageNotReused(t *testing.T, stage checkpointStage, fixture filesystemCheckpointFixture, want pipeline.CheckpointReasonCode) { t.Helper() decision := stage.load(fixture) - if decision.Reused || !strings.Contains(strings.ToLower(decision.Reason), strings.ToLower(want)) { - t.Fatalf("%s decision=%#v, want non-reused reason containing %q", stage.name, decision, want) + if decision.Reused || decision.ReasonCode != want { + t.Fatalf("%s decision=%#v, want non-reused reason code %q", stage.name, decision, want) } } diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index 4bc5140..c7a52b4 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -52,13 +52,13 @@ func (l *FilesystemLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, } doc := cloneSourceDocument(payload.Document) if err := source.ValidateDocument(&doc); err != nil { - return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint document is invalid: %v", err) + return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "source checkpoint document is invalid") } if strings.TrimSpace(manifest.SourceID) != "" && manifest.SourceID != doc.ID { - return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint source id does not match payload") + return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "source checkpoint identity does not match its payload") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), digestFingerprints("source_document", doc.Digest)) { - return pipeline.SourceCheckpoint{}, invalidDecision("source checkpoint output digest does not match payload") + return pipeline.SourceCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "source checkpoint output digest does not match its payload") } return pipeline.SourceCheckpoint{Document: &doc}, reusedDecision() } @@ -81,10 +81,10 @@ func (l *FilesystemLoader) ExtractForStep(stepID, laneID, moduleKey string, depe } outputs, err := artifactCheckpointOutputs(payload.Outputs) if err != nil { - return pipeline.ExtractCheckpoint{}, invalidDecision("extract artifact checkpoint payload is invalid: %v", err) + return pipeline.ExtractCheckpoint{}, artifactDecision(err, "extract checkpoint artifact is invalid") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(outputs)) { - return pipeline.ExtractCheckpoint{}, invalidDecision("extract artifact checkpoint output digests do not match payload") + return pipeline.ExtractCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "extract checkpoint output digest does not match its payload") } return pipeline.ExtractCheckpoint{Outputs: outputs, Rejected: cloneRejectedOutputs(payload.Rejected), Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } @@ -107,10 +107,10 @@ func (l *FilesystemLoader) MergeForStep(stepID, laneID, moduleKey string, depend } values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output}) if err != nil { - return pipeline.MergeCheckpoint{}, invalidDecision("merge artifact checkpoint payload is invalid: %v", err) + return pipeline.MergeCheckpoint{}, artifactDecision(err, "merge checkpoint artifact is invalid") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { - return pipeline.MergeCheckpoint{}, invalidDecision("merge artifact checkpoint output digest does not match payload") + return pipeline.MergeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "merge checkpoint output digest does not match its payload") } return pipeline.MergeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } @@ -133,10 +133,10 @@ func (l *FilesystemLoader) NormalizeForStep(stepID, laneID, moduleKey string, de } values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output}) if err != nil { - return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint payload is invalid: %v", err) + return pipeline.NormalizeCheckpoint{}, artifactDecision(err, "normalize checkpoint artifact is invalid") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { - return pipeline.NormalizeCheckpoint{}, invalidDecision("normalize artifact checkpoint output digest does not match payload") + return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "normalize checkpoint output digest does not match its payload") } return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } @@ -152,7 +152,7 @@ func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline. 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") + 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)}}) } @@ -161,21 +161,21 @@ func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline. func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision { if !l.Enabled() { - return pipeline.CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLoadingDisabled, "checkpoint loading disabled") } target, err := fileio.SafePath(l.root, name) if err != nil { - return invalidDecision("checkpoint path is invalid: %v", err) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonPathInvalid, "checkpoint path is invalid") } data, err := os.ReadFile(target) if err != nil { if os.IsNotExist(err) { - return pipeline.CheckpointDecision{Category: "executed", ReasonCode: "checkpoint_missing", Reason: "checkpoint artifact is missing"} + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonMissing, "checkpoint artifact is missing") } - return invalidDecision("read checkpoint artifact: %v", err) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonReadFailed, "checkpoint artifact could not be read") } if err := json.Unmarshal(data, out); err != nil { - return invalidDecision("decode checkpoint artifact: %v", err) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonDecodeFailed, "checkpoint artifact could not be decoded") } return reusedDecision() } @@ -186,28 +186,28 @@ func (l *FilesystemLoader) validateManifest(manifest StageManifest, stage StageN func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage StageName, stepID string, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...StageStatus) pipeline.CheckpointDecision { if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV1 { - return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, WorkspaceSchemaVersion) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") } if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV2 { - return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, WorkspaceSchemaVersion) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") } if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion { - return invalidDecision("checkpoint workspace schema version %q is not supported", manifest.WorkspaceSchemaVersion) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") } if strings.TrimSpace(l.identityDigest) != "" && manifest.Metadata["checkpoint_identity_digest"] != l.identityDigest { - return invalidDecision("checkpoint identity digest does not match current invocation") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonIdentityMismatch, "checkpoint identity does not match the current invocation") } if manifest.Stage != stage { - return invalidDecision("checkpoint stage %q does not match %q", manifest.Stage, stage) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStageMismatch, "checkpoint stage does not match the requested stage") } if strings.TrimSpace(stepID) != "" && manifest.StepID != stepID { - return invalidDecision("checkpoint step does not match requested step") + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStepMismatch, "checkpoint step does not match the requested step") } if strings.TrimSpace(laneID) != "" && manifest.LaneID != laneID { - return invalidDecision("checkpoint lane %q does not match %q", manifest.LaneID, laneID) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLaneMismatch, "checkpoint lane does not match the requested lane") } if strings.TrimSpace(moduleKey) != "" && manifest.ModuleKey != moduleKey { - return invalidDecision("checkpoint module %q does not match %q", manifest.ModuleKey, moduleKey) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch, "checkpoint module does not match the requested module") } statusOK := false for _, status := range statuses { @@ -217,10 +217,10 @@ func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage St } } if !statusOK { - return invalidDecision("checkpoint status %q cannot be reused", manifest.Status) + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStatusNotReusable, "checkpoint status cannot be reused") } if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.DependencyFingerprints), dependencies) { - return invalidDecision("checkpoint dependency fingerprints do not match") + return decision(pipeline.CheckpointDecisionDependencyInvalidated, pipeline.CheckpointReasonDependencyMismatch, "checkpoint dependencies do not match") } return reusedDecision() } @@ -228,10 +228,10 @@ func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage St func contentFromEnvelope(value binaryEnvelope) ([]byte, error) { content, err := base64.StdEncoding.DecodeString(value.ContentBase64) if err != nil { - return nil, fmt.Errorf("decode content_base64: %w", err) + return nil, &artifactPayloadError{code: pipeline.CheckpointReasonArtifactPayloadInvalid, err: fmt.Errorf("decode content_base64: %w", err)} } if digest := strings.TrimSpace(value.ContentDigest); digest != "" && digest != contentDigest(content) { - return nil, fmt.Errorf("content digest mismatch") + return nil, &artifactPayloadError{code: pipeline.CheckpointReasonArtifactDigestMismatch, err: fmt.Errorf("content digest mismatch")} } return content, nil } @@ -262,65 +262,24 @@ func fingerprintsEqual(a []pipeline.CheckpointFingerprint, b []pipeline.Checkpoi } func reusedDecision() pipeline.CheckpointDecision { - return pipeline.CheckpointDecision{Reused: true, Category: "reused", ReasonCode: "checkpoint_valid", Reason: "checkpoint is valid"} + return decision(pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonReused, "checkpoint is reusable") } -func invalidDecision(format string, args ...any) pipeline.CheckpointDecision { - reason := fmt.Sprintf(format, args...) - code := "checkpoint_invalid" - category := "executed" - switch { - case strings.Contains(reason, "dependency fingerprints"): - code, category = "dependency_mismatch", "dependency_invalidated" - case strings.Contains(reason, "missing"): - code = "checkpoint_missing" - case strings.Contains(reason, "schema"): - code = "workspace_schema_incompatible" - case strings.Contains(reason, "codec"): - code = "artifact_codec_incompatible" - case strings.Contains(reason, "content"): - code = "artifact_content_invalid" - case strings.Contains(reason, "identity"): - code = "identity_mismatch" - } - safe := safeReasonText(reason, code) - return pipeline.CheckpointDecision{Category: category, ReasonCode: code, Detail: safe, Reason: safe} +func decision(category pipeline.CheckpointDecisionCategory, code pipeline.CheckpointReasonCode, detail string) pipeline.CheckpointDecision { + return pipeline.NewCheckpointDecision(category, code, detail) } -func safeReasonText(reason, code string) string { - lower := strings.ToLower(reason) - switch { - case strings.Contains(lower, "workspace schema"): - return "checkpoint workspace schema is incompatible" - case strings.Contains(lower, "source checkpoint document"): - return "source checkpoint document is invalid" - case strings.Contains(lower, "artifact codec identity"): - return "artifact codec identity is incomplete" - case strings.Contains(lower, "base64"): - return "checkpoint content base64 is invalid" - case strings.Contains(lower, "content digest"): - return "checkpoint content digest is invalid" - case strings.Contains(lower, "output digest"): - return "checkpoint output digest does not match payload" - case strings.Contains(lower, "identity"): - return "checkpoint identity does not match" - case strings.Contains(lower, "dependency"): - return "checkpoint dependency fingerprints do not match" - case strings.Contains(lower, "stage"): - return "checkpoint stage does not match" - case strings.Contains(lower, "step"): - return "checkpoint step does not match" - case strings.Contains(lower, "lane"): - return "checkpoint lane does not match" - case strings.Contains(lower, "module"): - return "checkpoint module does not match" - case strings.Contains(lower, "status"): - return "checkpoint status cannot be reused" - case strings.Contains(lower, "payload"): - return "checkpoint artifact payload is invalid" - case strings.Contains(lower, "decode"): - return "checkpoint artifact decode failed" - default: - return "checkpoint is not reusable (" + code + ")" - } +type artifactPayloadError struct { + code pipeline.CheckpointReasonCode + err error +} + +func (e *artifactPayloadError) Error() string { return e.err.Error() } + +func artifactDecision(err error, detail string) pipeline.CheckpointDecision { + code := pipeline.CheckpointReasonArtifactPayloadInvalid + if payloadErr, ok := err.(*artifactPayloadError); ok { + code = payloadErr.code + } + return decision(pipeline.CheckpointDecisionExecuted, code, detail) } diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 743d52a..2aa1bb3 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -57,26 +57,96 @@ type StepCheckpointRecorder interface { NormalizeFailedForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error } +type CheckpointDecisionCategory string + +const ( + CheckpointDecisionExecuted CheckpointDecisionCategory = "executed" + CheckpointDecisionReused CheckpointDecisionCategory = "reused" + CheckpointDecisionForcedRecompute CheckpointDecisionCategory = "forced_recompute" + CheckpointDecisionDependencyInvalidated CheckpointDecisionCategory = "dependency_invalidated" +) + +type CheckpointReasonCode string + +const ( + CheckpointReasonLoadingDisabled CheckpointReasonCode = "loading_disabled" + CheckpointReasonMissing CheckpointReasonCode = "checkpoint_missing" + CheckpointReasonPathInvalid CheckpointReasonCode = "checkpoint_path_invalid" + CheckpointReasonReadFailed CheckpointReasonCode = "checkpoint_read_failed" + CheckpointReasonDecodeFailed CheckpointReasonCode = "checkpoint_decode_failed" + CheckpointReasonWorkspaceSchemaIncompatible CheckpointReasonCode = "workspace_schema_incompatible" + CheckpointReasonIdentityMismatch CheckpointReasonCode = "identity_mismatch" + CheckpointReasonStageMismatch CheckpointReasonCode = "stage_mismatch" + CheckpointReasonStepMismatch CheckpointReasonCode = "step_mismatch" + CheckpointReasonLaneMismatch CheckpointReasonCode = "lane_mismatch" + CheckpointReasonModuleMismatch CheckpointReasonCode = "module_mismatch" + CheckpointReasonStatusNotReusable CheckpointReasonCode = "status_not_reusable" + CheckpointReasonDependencyMismatch CheckpointReasonCode = "dependency_mismatch" + CheckpointReasonArtifactPayloadInvalid CheckpointReasonCode = "artifact_payload_invalid" + CheckpointReasonArtifactDigestMismatch CheckpointReasonCode = "artifact_digest_mismatch" + CheckpointReasonArtifactCodecIncompatible CheckpointReasonCode = "artifact_codec_incompatible" + CheckpointReasonArtifactNotCanonical CheckpointReasonCode = "artifact_not_canonical" + CheckpointReasonReused CheckpointReasonCode = "checkpoint_reused" + CheckpointReasonAcceptedArtifactReused CheckpointReasonCode = "accepted_artifact_reused" + CheckpointReasonRecomputeStep CheckpointReasonCode = "recompute_step" +) + type CheckpointDecision struct { - Reused bool `json:"reused"` - Category string `json:"category,omitempty"` - ReasonCode string `json:"reason_code,omitempty"` - Detail string `json:"detail,omitempty"` + Reused bool `json:"reused"` + Category CheckpointDecisionCategory `json:"category,omitempty"` + ReasonCode CheckpointReasonCode `json:"reason_code,omitempty"` + Detail string `json:"detail,omitempty"` // Reason is retained as a compatibility/debug field for existing callers. // New checkpoint stores should put bounded, non-sensitive text in Detail. Reason string `json:"reason,omitempty"` } +const checkpointDecisionDetailLimit = 512 + +func NewCheckpointDecision(category CheckpointDecisionCategory, reasonCode CheckpointReasonCode, detail string) CheckpointDecision { + return checkpointDecision(category, reasonCode, detail) +} + +func checkpointDecision(category CheckpointDecisionCategory, reasonCode CheckpointReasonCode, detail string) CheckpointDecision { + detail = sanitizeCheckpointDecisionDetail(detail) + return CheckpointDecision{ + Reused: category == CheckpointDecisionReused, + Category: category, + ReasonCode: reasonCode, + Detail: detail, + Reason: detail, + } +} + +func sanitizeCheckpointDecisionDetail(detail string) string { + detail = strings.TrimSpace(strings.ToValidUTF8(detail, "?")) + lower := strings.ToLower(detail) + if strings.ContainsAny(detail, `/\\`) || strings.Contains(lower, "secret") || strings.Contains(lower, "token") || strings.Contains(lower, "password") || strings.Contains(lower, "credential") || strings.Contains(lower, "environment") { + return "checkpoint decision detail redacted" + } + var b strings.Builder + for _, r := range detail { + if r < 0x20 || r == 0x7f { + r = ' ' + } + if b.Len()+len(string(r)) > checkpointDecisionDetailLimit { + break + } + b.WriteRune(r) + } + return strings.Join(strings.Fields(b.String()), " ") +} + type CheckpointEvent struct { - Stage string `json:"stage"` - StepID string `json:"step_id,omitempty"` - LaneID string `json:"lane_id,omitempty"` - ModuleKey string `json:"module_key,omitempty"` - Action string `json:"action"` - Category string `json:"category,omitempty"` - ReasonCode string `json:"reason_code,omitempty"` - Detail string `json:"detail,omitempty"` - Reason string `json:"reason,omitempty"` + Stage string `json:"stage"` + StepID string `json:"step_id,omitempty"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + Action CheckpointDecisionCategory `json:"action"` + Category CheckpointDecisionCategory `json:"category,omitempty"` + ReasonCode CheckpointReasonCode `json:"reason_code,omitempty"` + Detail string `json:"detail,omitempty"` + Reason string `json:"reason,omitempty"` } type CheckpointExecutionPolicy struct { @@ -100,14 +170,14 @@ func (policy CheckpointExecutionPolicy) requiresReusable(stepID, laneID string) func forceCheckpointDecision(policy CheckpointExecutionPolicy, stepID, laneID string, decision CheckpointDecision) CheckpointDecision { if policy.forced(stepID, laneID) { - return CheckpointDecision{Category: "forced_recompute", ReasonCode: "recompute_step", Detail: "selected step requires execution"} + return checkpointDecision(CheckpointDecisionForcedRecompute, CheckpointReasonRecomputeStep, "selected step requires execution") } return decision } func requireReusableCheckpoint(policy CheckpointExecutionPolicy, stepID, laneID string, decision CheckpointDecision) error { if policy.requiresReusable(stepID, laneID) && !decision.Reused { - return fmt.Errorf("required reusable checkpoint unavailable for step %q lane %q", strings.TrimSpace(stepID), strings.TrimSpace(laneID)) + return fmt.Errorf("required reusable checkpoint unavailable for step %q lane %q (%s)", strings.TrimSpace(stepID), strings.TrimSpace(laneID), decision.ReasonCode) } return nil } @@ -116,23 +186,20 @@ func requireReusableCheckpoint(policy CheckpointExecutionPolicy, stepID, laneID // validation at the single point where a stage's observable decision is made. func resolveCheckpointDecision(output *RunOutput, loader CheckpointLoader, policy CheckpointExecutionPolicy, stage ModuleStage, stepID, laneID, moduleKey string, decision CheckpointDecision, codec artifactCodecEntry, artifacts []CheckpointArtifact) (CheckpointDecision, error) { decision = forceCheckpointDecision(policy, stepID, laneID, decision) - if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil { - return decision, err - } if decision.Reused { for _, artifact := range artifacts { if _, _, err := decodeCanonicalCheckpointArtifact(codec, artifact); err != nil { - decision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored " + string(stage) + " artifact failed canonical codec validation", Reason: string(stage) + " artifact checkpoint is not canonical"} + decision = checkpointDecision(CheckpointDecisionExecuted, checkpointArtifactReasonCode(err), "stored "+string(stage)+" artifact failed canonical codec validation") break } } } - if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil { - return decision, err - } if output != nil { recordCheckpointEvent(output, loader, string(stage), stepID, laneID, moduleKey, decision) } + if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil { + return decision, err + } return decision, nil } @@ -228,16 +295,16 @@ func (noopCheckpointRecorder) NormalizeFailed(string, string, []CheckpointFinger func (noopCheckpointLoader) Enabled() bool { return false } func (noopCheckpointLoader) Source(string) (SourceCheckpoint, CheckpointDecision) { - return SourceCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} + return SourceCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") } func (noopCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { - return ExtractCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} + return ExtractCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") } func (noopCheckpointLoader) Merge(string, string, []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { - return MergeCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} + return MergeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") } func (noopCheckpointLoader) Normalize(string, string, []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { - return NormalizeCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} + return NormalizeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") } func checkpointExtractRunning(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) error { diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 30f6867..66daf49 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -623,14 +623,14 @@ func recordCheckpointEvent(output *RunOutput, loader CheckpointLoader, stage str }) } -func checkpointDecisionCategory(decision CheckpointDecision) string { +func checkpointDecisionCategory(decision CheckpointDecision) CheckpointDecisionCategory { if decision.Category != "" { return decision.Category } if decision.Reused { - return "reused" + return CheckpointDecisionReused } - return "executed" + return CheckpointDecisionExecuted } func populateOutputManifest(output *RunOutput) { @@ -642,7 +642,7 @@ func populateOutputManifest(output *RunOutput) { if len(output.CheckpointEvents) > 0 { decisions := make([]artifacts.CheckpointDecisionManifest, 0, len(output.CheckpointEvents)) for _, event := range output.CheckpointEvents { - decisions = append(decisions, artifacts.CheckpointDecisionManifest{Stage: event.Stage, StepID: event.StepID, LaneID: event.LaneID, ModuleKey: event.ModuleKey, Category: event.Category, ReasonCode: event.ReasonCode, Detail: event.Detail}) + decisions = append(decisions, artifacts.CheckpointDecisionManifest{Stage: event.Stage, StepID: event.StepID, LaneID: event.LaneID, ModuleKey: event.ModuleKey, Category: string(event.Category), ReasonCode: string(event.ReasonCode), Detail: event.Detail}) } output.Manifest.CheckpointDecisions = decisions } diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 01e8c3b..d815923 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -100,7 +100,7 @@ func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints if err := setTypedLaneManifestMetadata(output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil { return nil, err } - state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared) + state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared, output) if err != nil { return nil, err } @@ -175,6 +175,13 @@ func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoin }() } + completedOutputs, runErrors := collectLaneResults(ctx, cancel, input, checkpoints, chunks, states, results, completions, continuations) + close(continuations) + continuationWorkers.Wait() + return completedOutputs, runErrors +} + +func collectLaneResults(ctx context.Context, cancel context.CancelFunc, input RunInput, checkpoints CheckpointRecorder, chunks []source.Chunk, states []*laneExtractState, results <-chan extractJobResult, completions <-chan laneCompletion, continuations chan<- *laneExtractState) ([]RunOutput, []orderedRunError) { completedOutputs := make([]RunOutput, len(states)) var runErrors []orderedRunError var pendingContinuations []*laneExtractState @@ -184,7 +191,6 @@ func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoin pendingContinuations = append(pendingContinuations, state) } } - resultChannel := results for resultChannel != nil || len(pendingContinuations) > 0 || completed < launched { var continuationChannel chan<- *laneExtractState @@ -232,8 +238,6 @@ func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoin } } } - close(continuations) - continuationWorkers.Wait() return completedOutputs, runErrors } @@ -246,7 +250,7 @@ func mergeCompletedLanes(output *RunOutput, completedOutputs []RunOutput) error return nil } -func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, chunks []source.Chunk, index int, prepared preparedLaneExecutor) (*laneExtractState, error) { +func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, chunks []source.Chunk, index int, prepared preparedLaneExecutor, output *RunOutput) (*laneExtractState, error) { lane, typed := prepared.resolved, prepared.typed digest, err := joinedChunkDigest(chunks) if err != nil { @@ -256,7 +260,7 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou deps := append(digestFingerprints("chunks", digest), generatedReferenceDependencies(extractReferences)...) state := &laneExtractState{index: index, prepared: prepared, deps: normalizeCheckpointFingerprints(deps), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))} cp, decision := loadExtract(loader, input.stepID, lane.ID, lane.Extract.Module, state.deps) - decision, err = resolveCheckpointDecision(nil, loader, input.CheckpointPolicy, StageExtract, input.stepID, lane.ID, lane.Extract.Module, decision, typed.codec, cp.Outputs) + decision, err = resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageExtract, input.stepID, lane.ID, lane.Extract.Module, decision, typed.codec, cp.Outputs) if err != nil { return nil, err } @@ -385,7 +389,6 @@ func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints C } local.Warnings = append(local.Warnings, cloneWarnings(results.warnings)...) local.Rejected = append(local.Rejected, cloneRejectedOutputs(results.rejected)...) - recordCheckpointEvent(&local, loader, string(StageExtract), input.stepID, lane.ID, lane.Extract.Module, results.decision) if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "decision": results.decision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return local, &laneRunError{stage: StageExtract, err: err} } diff --git a/internal/framework/pipeline/runner_extract_handoff_test.go b/internal/framework/pipeline/runner_extract_handoff_test.go index 2ba4fd6..bd92796 100644 --- a/internal/framework/pipeline/runner_extract_handoff_test.go +++ b/internal/framework/pipeline/runner_extract_handoff_test.go @@ -252,7 +252,7 @@ func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) { } } -func assertExtractDecision(t *testing.T, events []CheckpointEvent, action string, reason string) { +func assertExtractDecision(t *testing.T, events []CheckpointEvent, action CheckpointDecisionCategory, reason string) { t.Helper() for _, event := range events { if event.Stage == string(StageExtract) { diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 3bd9b98..475ad67 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -100,21 +100,45 @@ func serializeArtifact(codec artifactCodecEntry, value any, candidate bool) (con return contracts.SerializedArtifact{Kind: codec.spec.Kind, Schema: contracts.CloneArtifactSchema(schema), MediaType: codec.spec.MediaType, Content: append([]byte(nil), content...), Metadata: metadata}, nil } +type checkpointArtifactValidationError struct { + code CheckpointReasonCode + err error +} + +func (e *checkpointArtifactValidationError) Error() string { return e.err.Error() } +func (e *checkpointArtifactValidationError) Unwrap() error { return e.err } + +func checkpointArtifactValidationFailure(code CheckpointReasonCode, format string, args ...any) error { + return &checkpointArtifactValidationError{code: code, err: fmt.Errorf(format, args...)} +} + +func checkpointArtifactReasonCode(err error) CheckpointReasonCode { + var validationErr *checkpointArtifactValidationError + if errors.As(err, &validationErr) { + return validationErr.code + } + return CheckpointReasonArtifactPayloadInvalid +} + func decodeCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtifact) (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 nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "artifact kind %q does not match codec %q", artifact.Artifact.Kind, codec.spec.Kind) } 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) + return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "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) + return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "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 nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "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...)) + value, err := codec.decode(append([]byte(nil), artifact.Artifact.Content...)) + if err != nil { + return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactPayloadInvalid, "decode artifact payload: %w", err) + } + return value, nil } func decodeCanonicalCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtifact) (any, CheckpointArtifact, error) { @@ -124,14 +148,14 @@ func decodeCanonicalCheckpointArtifact(codec artifactCodecEntry, artifact Checkp } canonical, err := serializeArtifact(codec, value, false) if err != nil { - return nil, CheckpointArtifact{}, err + return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactPayloadInvalid, "encode canonical artifact: %w", err) } if canonical.Kind != artifact.Artifact.Kind || canonical.MediaType != artifact.Artifact.MediaType || checkpointContentDigest(canonical.Content) != checkpointContentDigest(artifact.Artifact.Content) { - return nil, CheckpointArtifact{}, fmt.Errorf("stored artifact is not canonical") + return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactNotCanonical, "stored artifact is not canonical") } hydrated, err := hydrateCheckpointArtifact(codec, cloneCheckpointArtifact(artifact), value) if err != nil { - return nil, CheckpointArtifact{}, err + return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactPayloadInvalid, "hydrate checkpoint artifact: %w", err) } return value, hydrated, nil } diff --git a/internal/framework/pipeline/runner_typed_checkpoint_test.go b/internal/framework/pipeline/runner_typed_checkpoint_test.go index 50d4e9c..e6dbdaa 100644 --- a/internal/framework/pipeline/runner_typed_checkpoint_test.go +++ b/internal/framework/pipeline/runner_typed_checkpoint_test.go @@ -1,12 +1,94 @@ package pipeline import ( + "context" + "encoding/json" "strings" "testing" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) +type requiredCheckpointLoader struct { + CheckpointLoader + checkpoint ExtractCheckpoint + decision CheckpointDecision +} + +func (l requiredCheckpointLoader) Enabled() bool { return true } +func (l requiredCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + return l.checkpoint, l.decision +} + +func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { + const unsafeDetail = "unsafe-loader-detail-/private/checkpoint/path" + tests := []struct { + name string + decision CheckpointDecision + corrupt bool + wantCode CheckpointReasonCode + wantAction CheckpointDecisionCategory + }{ + {"missing", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, unsafeDetail), false, CheckpointReasonMissing, CheckpointDecisionExecuted}, + {"corrupt", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonReused, "checkpoint reusable"), true, CheckpointReasonArtifactNotCanonical, CheckpointDecisionExecuted}, + {"incompatible", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactCodecIncompatible, unsafeDetail), false, CheckpointReasonArtifactCodecIncompatible, CheckpointDecisionExecuted}, + {"dependency invalidated", NewCheckpointDecision(CheckpointDecisionDependencyInvalidated, CheckpointReasonDependencyMismatch, unsafeDetail), false, CheckpointReasonDependencyMismatch, CheckpointDecisionDependencyInvalidated}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + prepared := preparedConcurrentPipeline(t, 1) + step := prepared.Steps[0] + lane := step.lanes[0] + checkpoint := ExtractCheckpoint{} + if test.corrupt { + stored, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Extract.Module, "source", codecNotes{Items: []string{"stored"}}) + if err != nil { + t.Fatal(err) + } + stored.Artifact.Content = []byte(`{"items": ["stored"]}`) + checkpoint.Outputs = []CheckpointArtifact{stored} + } + loader := requiredCheckpointLoader{CheckpointLoader: NoopCheckpointLoader(), checkpoint: checkpoint, decision: test.decision} + policy := CheckpointExecutionPolicy{RequireReusableLanes: map[string]struct{}{CheckpointLaneKey(step.ID, lane.resolved.ID): {}}} + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}) + if err == nil || !strings.Contains(err.Error(), step.ID) || !strings.Contains(err.Error(), lane.resolved.ID) || !strings.Contains(err.Error(), string(test.wantCode)) { + t.Fatalf("Run() error = %v, want step, lane, and reason code %q", err, test.wantCode) + } + if strings.Contains(err.Error(), unsafeDetail) { + t.Fatalf("Run() error leaked loader detail: %v", err) + } + var found bool + for _, event := range output.CheckpointEvents { + if event.Stage == string(StageExtract) && event.StepID == step.ID && event.LaneID == lane.resolved.ID { + found = true + if event.Action != test.wantAction || event.ReasonCode != test.wantCode { + t.Fatalf("checkpoint event = %#v, want action %q and reason %q", event, test.wantAction, test.wantCode) + } + if strings.Contains(event.Detail, unsafeDetail) { + t.Fatalf("checkpoint event leaked loader detail: %#v", event) + } + } + } + if !found { + t.Fatalf("required checkpoint decision missing from %#v", output.CheckpointEvents) + } + var manifestFound bool + for _, decision := range output.Manifest.CheckpointDecisions { + if decision.Stage == string(StageExtract) && decision.StepID == step.ID && decision.LaneID == lane.resolved.ID { + manifestFound = decision.Category == string(test.wantAction) && decision.ReasonCode == string(test.wantCode) + } + } + if !manifestFound { + t.Fatalf("manifest checkpoint decision missing category %q and code %q: %#v", test.wantAction, test.wantCode, output.Manifest.CheckpointDecisions) + } + encoded, marshalErr := json.Marshal(output.CheckpointEvents) + if marshalErr != nil || !strings.Contains(string(encoded), `"category":"`+string(test.wantAction)+`"`) || !strings.Contains(string(encoded), `"reason_code":"`+string(test.wantCode)+`"`) { + t.Fatalf("checkpoint event JSON = %s, error = %v", encoded, marshalErr) + } + }) + } +} + func TestDecodeCheckpointArtifactRejectsIncompatibleCodecIdentityAndBytes(t *testing.T) { registry := NewArtifactCodecRegistry() if err := RegisterArtifactCodec(registry, notesCodec()); err != nil {