diff --git a/internal/cli/recompute_policy_test.go b/internal/cli/recompute_policy_test.go new file mode 100644 index 0000000..a9bc7e7 --- /dev/null +++ b/internal/cli/recompute_policy_test.go @@ -0,0 +1,56 @@ +package cli + +import ( + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/pipeline" +) + +func TestRecomputePolicyIncludesDependentLanesAndReusablePredecessors(t *testing.T) { + producer := pipeline.ResolvedArtifactLane{StepID: "first", ID: "producer"} + unrelated := pipeline.ResolvedArtifactLane{StepID: "first", ID: "unrelated"} + consumer := pipeline.ResolvedArtifactLane{ + StepID: "second", ID: "consumer", + ExtractReferences: pipeline.ResolvedReferenceTarget{Bindings: []pipeline.ReferenceBinding{{Artifact: &pipeline.ArtifactReference{Step: "first", Lane: "producer"}}}}, + } + downstream := pipeline.ResolvedArtifactLane{ + StepID: "third", ID: "downstream", + ExtractReferences: pipeline.ResolvedReferenceTarget{Bindings: []pipeline.ReferenceBinding{{Artifact: &pipeline.ArtifactReference{Step: "second", Lane: "consumer"}}}}, + } + independent := pipeline.ResolvedArtifactLane{StepID: "third", ID: "independent"} + resolved := pipeline.ResolvedPipeline{Steps: []pipeline.ResolvedPipelineStep{ + {ID: "first", ArtifactLanes: []pipeline.ResolvedArtifactLane{producer, unrelated}}, + {ID: "second", ArtifactLanes: []pipeline.ResolvedArtifactLane{consumer}}, + {ID: "third", ArtifactLanes: []pipeline.ResolvedArtifactLane{downstream, independent}}, + }} + + policy, err := recomputePolicy(resolved, "second") + if err != nil { + t.Fatal(err) + } + if _, ok := policy.ForcedLanes[pipeline.CheckpointLaneKey("second", "consumer")]; !ok { + t.Fatal("selected lane was not forced") + } + if _, ok := policy.ForcedLanes[pipeline.CheckpointLaneKey("third", "downstream")]; !ok { + t.Fatal("transitive dependent lane was not forced") + } + if _, ok := policy.ForcedLanes[pipeline.CheckpointLaneKey("first", "producer")]; ok { + t.Fatal("predecessor was implicitly forced") + } + if _, ok := policy.RequireReusableLanes[pipeline.CheckpointLaneKey("first", "producer")]; !ok { + t.Fatal("required predecessor was not marked reusable") + } + if _, ok := policy.ForcedLanes[pipeline.CheckpointLaneKey("first", "unrelated")]; ok { + t.Fatal("unrelated lane was forced") + } + if _, ok := policy.ForcedLanes[pipeline.CheckpointLaneKey("third", "independent")]; ok { + t.Fatal("unrelated later lane was forced") + } +} + +func TestRecomputePolicyRejectsUnknownStep(t *testing.T) { + _, err := recomputePolicy(pipeline.ResolvedPipeline{Steps: []pipeline.ResolvedPipelineStep{{ID: "known"}}}, "missing") + if err == nil { + t.Fatal("unknown step was accepted") + } +} diff --git a/internal/cli/run.go b/internal/cli/run.go index 5554c5b..1f1d18c 100644 --- a/internal/cli/run.go +++ b/internal/cli/run.go @@ -28,7 +28,7 @@ import ( const defaultConfigPath = "/usr/local/etc/notarius/config.yml" const usage = `Usage: notarius help - notarius run --input path/to/source.json [--config path/to/config.yml] [--output-dir path] [--chunk_cache auto|bypass|refresh] [--resume] [--debug [--debug-dir path]] [--only lane-a,lane-b] [--session-id id] [--reference selector=path] [--without-reference selector] + notarius run --input path/to/source.json [--config path/to/config.yml] [--output-dir path] [--chunk_cache auto|bypass|refresh] [--resume] [--recompute-step step-id] [--debug [--debug-dir path]] [--only lane-a,lane-b] [--session-id id] [--reference selector=path] [--without-reference selector] notarius config validate --config path/to/config.yml [--pipeline pipeline-id] [--only lane-a,lane-b] notarius pipelines list --config path/to/config.yml [--json] ` @@ -136,6 +136,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i debugDir := fs.String("debug-dir", "", "debug bundle directory") llmProfile := fs.String("llm-profile", "", "LLM profile override") resume := fs.Bool("resume", false, "reuse compatible recorded checkpoints") + recomputeStep := singleValueFlag{} chunkCache := chunkCacheFlag{} sessionID := sessionIDFlag{} referenceFlags := stringListFlag{} @@ -144,6 +145,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i fs.Var(&chunkCache, "chunk_cache", "chunk plan cache mode: auto, bypass, or refresh") fs.Var(&referenceFlags, "reference", "reference binding, as slot=path, chunk.slot=path, merge.slot=path, lane.slot=path, lane.extract.slot=path, lane.merge.slot=path, or lane.normalize.slot=path") fs.Var(&withoutReferenceFlags, "without-reference", "unbind a reference, using the same selector forms as --reference") + fs.Var(&recomputeStep, "recompute-step", "recompute one ordered pipeline step and dependent lanes") if err := validateRunFlagValues(args); err != nil { fmt.Fprintf(stderr, "notarius: %v\n", err) return 2 @@ -190,6 +192,14 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i fmt.Fprintf(stderr, "notarius: %v\n", err) return 2 } + if recomputeStep.set && strings.TrimSpace(recomputeStep.value) == "" { + fmt.Fprintln(stderr, "notarius: --recompute-step must not be empty") + return 2 + } + if recomputeStep.set && len(only) > 0 { + fmt.Fprintln(stderr, "notarius: --recompute-step cannot be combined with --only") + return 2 + } referenceRequests, err := parseReferenceFlags(referenceFlags) if err != nil { fmt.Fprintf(stderr, "notarius: %v\n", err) @@ -223,6 +233,14 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i fmt.Fprintln(stderr, "notarius: --resume requires cache.checkpoints.enabled: true") return 1 } + if recomputeStep.set && !*resume { + fmt.Fprintln(stderr, "notarius: --recompute-step requires --resume") + return 2 + } + if recomputeStep.set && !cfg.Cache.Checkpoints.Enabled { + fmt.Fprintln(stderr, "notarius: --recompute-step requires cache.checkpoints.enabled: true") + return 1 + } startedAt := opts.Now().UTC() runID, err := opts.RunIDGenerator(startedAt) @@ -267,6 +285,7 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i OnlyLanes: append([]string(nil), only...), ChunkCacheOverride: chunkCache.explicitValue(), Resume: *resume, + RecomputeStep: strings.TrimSpace(recomputeStep.value), RunID: runID, StartedAt: startedAt, } @@ -309,6 +328,13 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i return failPipelineCommand(stderr, commandState, terminalWriter, err) } effective.ResolvedPipeline = materialized + checkpointPolicy := pipeline.CheckpointExecutionPolicy{} + if recomputeStep.set { + checkpointPolicy, err = recomputePolicy(effective.ResolvedPipeline, recomputeStep.value) + if err != nil { + return failPipelineCommand(stderr, commandState, terminalWriter, err) + } + } invocation.PipelineDigest = effective.ResolvedPipeline.Digest if err := writeSummary(summary, func() error { return summary.WriteInvocation(invocation) }); err != nil { return failPipelineCommand(stderr, commandState, terminalWriter, fmt.Errorf("write debug invocation metadata: %w", err)) @@ -358,21 +384,22 @@ func runPipelineCommand(args []string, stdout, stderr io.Writer, opts Options) i } output, err := pipeline.New().Run(ctx, pipeline.RunInput{ - Prepared: prepared, - Path: strings.TrimSpace(*inputPath), - RawInput: rawInput, - SessionID: strings.TrimSpace(sessionID.value), - RunID: runID, - StartedAt: startedAt, - LLMProfiles: llmProfiles, - Metadata: runMetadata(effective.Config.Output.Directory, debugPath), - Warnings: referenceWarnings, - ChunkCacheMode: effective.Config.Cache.ChunkPlans.Mode, - ChunkPlans: chunkPlans, - Checkpoints: checkpointRecorder, - Checkpoint: checkpointLoader, - Debug: debugRecorder, - ExtractWorkers: cfg.Concurrency.StageWorkers["extract"], + Prepared: prepared, + Path: strings.TrimSpace(*inputPath), + RawInput: rawInput, + SessionID: strings.TrimSpace(sessionID.value), + RunID: runID, + StartedAt: startedAt, + LLMProfiles: llmProfiles, + Metadata: runMetadata(effective.Config.Output.Directory, debugPath), + Warnings: referenceWarnings, + ChunkCacheMode: effective.Config.Cache.ChunkPlans.Mode, + ChunkPlans: chunkPlans, + Checkpoints: checkpointRecorder, + Checkpoint: checkpointLoader, + CheckpointPolicy: checkpointPolicy, + Debug: debugRecorder, + ExtractWorkers: cfg.Concurrency.StageWorkers["extract"], }) commandState.observeOutput(output) if err != nil { @@ -481,6 +508,95 @@ func checkpointHandlersForRun( return recorder, loader, nil } +func recomputePolicy(resolved pipeline.ResolvedPipeline, requestedStep string) (pipeline.CheckpointExecutionPolicy, error) { + requestedStep = strings.TrimSpace(requestedStep) + if requestedStep == "" { + return pipeline.CheckpointExecutionPolicy{}, fmt.Errorf("--recompute-step must not be empty") + } + var selected *pipeline.ResolvedPipelineStep + for index := range resolved.Steps { + if strings.TrimSpace(resolved.Steps[index].ID) == requestedStep { + selected = &resolved.Steps[index] + break + } + } + if selected == nil { + return pipeline.CheckpointExecutionPolicy{}, fmt.Errorf("unknown pipeline step %q", requestedStep) + } + + policy := pipeline.CheckpointExecutionPolicy{ForcedLanes: make(map[string]struct{}), RequireReusableLanes: make(map[string]struct{})} + var forced []pipeline.ResolvedArtifactLane + for _, lane := range selected.ArtifactLanes { + lane.StepID = selected.ID + policy.ForcedLanes[pipeline.CheckpointLaneKey(selected.ID, lane.ID)] = struct{}{} + forced = append(forced, lane) + } + + lanes := make(map[string]pipeline.ResolvedArtifactLane) + dependents := make(map[string][]pipeline.ResolvedArtifactLane) + for _, step := range resolved.Steps { + for _, lane := range step.ArtifactLanes { + lane.StepID = step.ID + lanes[pipeline.CheckpointLaneKey(step.ID, lane.ID)] = lane + for _, target := range []pipeline.ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} { + for _, binding := range target.Bindings { + if binding.Artifact != nil { + producerKey := pipeline.CheckpointLaneKey(binding.Artifact.Step, binding.Artifact.Lane) + dependents[producerKey] = append(dependents[producerKey], lane) + } + } + } + } + } + processed := make(map[string]struct{}) + queue := append([]pipeline.ResolvedArtifactLane(nil), forced...) + for len(queue) > 0 { + lane := queue[0] + queue = queue[1:] + key := pipeline.CheckpointLaneKey(lane.StepID, lane.ID) + if _, ok := processed[key]; ok { + continue + } + processed[key] = struct{}{} + for _, dependent := range dependents[key] { + dependentKey := pipeline.CheckpointLaneKey(dependent.StepID, dependent.ID) + if _, alreadyForced := policy.ForcedLanes[dependentKey]; alreadyForced { + continue + } + policy.ForcedLanes[dependentKey] = struct{}{} + forced = append(forced, dependent) + queue = append(queue, dependent) + } + } + + processed = make(map[string]struct{}) + queue = append([]pipeline.ResolvedArtifactLane(nil), forced...) + for len(queue) > 0 { + lane := queue[0] + queue = queue[1:] + key := pipeline.CheckpointLaneKey(lane.StepID, lane.ID) + if _, ok := processed[key]; ok { + continue + } + processed[key] = struct{}{} + for _, target := range []pipeline.ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} { + for _, binding := range target.Bindings { + if binding.Artifact == nil { + continue + } + producerKey := pipeline.CheckpointLaneKey(binding.Artifact.Step, binding.Artifact.Lane) + if _, forcedAlready := policy.ForcedLanes[producerKey]; !forcedAlready { + policy.RequireReusableLanes[producerKey] = struct{}{} + } + if producer, ok := lanes[producerKey]; ok { + queue = append(queue, producer) + } + } + } + } + return policy, nil +} + func checkpointIdentityFingerprints(values []pipeline.CheckpointFingerprint) []checkpoint.Fingerprint { if len(values) == 0 { return nil @@ -661,7 +777,7 @@ func reorderRunArgs(args []string) []string { func runFlagTakesValue(arg string) bool { switch arg { - case "--config", "--input", "--only", "--output-dir", "--debug-dir", "--llm-profile", "--session-id", "--chunk_cache", "--reference", "--without-reference": + case "--config", "--input", "--only", "--output-dir", "--debug-dir", "--llm-profile", "--session-id", "--chunk_cache", "--reference", "--without-reference", "--recompute-step": return true default: return false @@ -1008,6 +1124,27 @@ type sessionIDFlag struct { set bool } +type singleValueFlag struct { + value string + set bool +} + +func (flag *singleValueFlag) String() string { + if flag == nil { + return "" + } + return flag.value +} + +func (flag *singleValueFlag) Set(value string) error { + if flag.set { + return fmt.Errorf("--recompute-step may be specified only once") + } + flag.value = value + flag.set = true + return nil +} + func (flag *sessionIDFlag) String() string { if flag == nil { return "" diff --git a/internal/core/artifacts/artifacts.go b/internal/core/artifacts/artifacts.go index a1ef506..afffd93 100644 --- a/internal/core/artifacts/artifacts.go +++ b/internal/core/artifacts/artifacts.go @@ -82,6 +82,16 @@ type RejectedOutputManifest struct { DiagnosticArtifactPath string `json:"diagnostic_artifact_path,omitempty"` } +type CheckpointDecisionManifest struct { + Stage string `json:"stage"` + StepID string `json:"step_id,omitempty"` + LaneID string `json:"lane_id,omitempty"` + ModuleKey string `json:"module_key,omitempty"` + Category string `json:"category"` + ReasonCode string `json:"reason_code,omitempty"` + Detail string `json:"detail,omitempty"` +} + type ChunkPlanManifest struct { Mode string `json:"mode"` Action string `json:"action,omitempty"` @@ -112,27 +122,28 @@ type ChunkPlanSummary struct { } type RunManifest struct { - RunID string `json:"run_id,omitempty"` - PipelineID string `json:"pipeline_id,omitempty"` - PipelineDigest string `json:"pipeline_digest,omitempty"` - InputModule string `json:"input_module,omitempty"` - Chunker string `json:"chunker,omitempty"` - ChunkPlan *ChunkPlanManifest `json:"chunk_plan,omitempty"` - SourceDigests []string `json:"source_digests,omitempty"` - Extractors []string `json:"extractors,omitempty"` - Merger string `json:"merger,omitempty"` - Normalizer string `json:"normalizer,omitempty"` - OutputEncoder string `json:"output_encoder,omitempty"` - ModuleMetadata map[string]map[string]any `json:"module_metadata,omitempty"` - ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` - ValidatorChains []ValidatorChainManifest `json:"validator_chains,omitempty"` - References []ReferenceProvenance `json:"references,omitempty"` - NormalizedOutputs []NormalizedOutputManifest `json:"normalized_outputs,omitempty"` - RejectedOutputs []RejectedOutputManifest `json:"rejected_outputs,omitempty"` - LLMProfiles []LLMProfileManifest `json:"llm_profiles,omitempty"` - Metadata map[string]any `json:"metadata,omitempty"` - SchemaVersion string `json:"schema_version,omitempty"` - ValidationStatus string `json:"validation_status,omitempty"` - StartedAt *time.Time `json:"started_at,omitempty"` - CompletedAt *time.Time `json:"completed_at,omitempty"` + RunID string `json:"run_id,omitempty"` + PipelineID string `json:"pipeline_id,omitempty"` + PipelineDigest string `json:"pipeline_digest,omitempty"` + InputModule string `json:"input_module,omitempty"` + Chunker string `json:"chunker,omitempty"` + ChunkPlan *ChunkPlanManifest `json:"chunk_plan,omitempty"` + SourceDigests []string `json:"source_digests,omitempty"` + Extractors []string `json:"extractors,omitempty"` + Merger string `json:"merger,omitempty"` + Normalizer string `json:"normalizer,omitempty"` + OutputEncoder string `json:"output_encoder,omitempty"` + ModuleMetadata map[string]map[string]any `json:"module_metadata,omitempty"` + ArtifactLanes []ArtifactLaneManifest `json:"artifact_lanes,omitempty"` + ValidatorChains []ValidatorChainManifest `json:"validator_chains,omitempty"` + References []ReferenceProvenance `json:"references,omitempty"` + NormalizedOutputs []NormalizedOutputManifest `json:"normalized_outputs,omitempty"` + RejectedOutputs []RejectedOutputManifest `json:"rejected_outputs,omitempty"` + CheckpointDecisions []CheckpointDecisionManifest `json:"checkpoint_decisions,omitempty"` + LLMProfiles []LLMProfileManifest `json:"llm_profiles,omitempty"` + Metadata map[string]any `json:"metadata,omitempty"` + SchemaVersion string `json:"schema_version,omitempty"` + ValidationStatus string `json:"validation_status,omitempty"` + StartedAt *time.Time `json:"started_at,omitempty"` + CompletedAt *time.Time `json:"completed_at,omitempty"` } diff --git a/internal/core/debugbundle/summary.go b/internal/core/debugbundle/summary.go index 49b291e..6f957d2 100644 --- a/internal/core/debugbundle/summary.go +++ b/internal/core/debugbundle/summary.go @@ -33,6 +33,7 @@ type Invocation struct { PipelineID string `json:"pipeline_id,omitempty"` PipelineDigest string `json:"pipeline_digest,omitempty"` Resume bool `json:"resume,omitempty"` + RecomputeStep string `json:"recompute_step,omitempty"` InputPath string `json:"input_path,omitempty"` ConfigPath string `json:"config_path,omitempty"` ConfigSource string `json:"config_source,omitempty"` diff --git a/internal/framework/checkpoint/filesystem_test.go b/internal/framework/checkpoint/filesystem_test.go index fd0dbf0..117b804 100644 --- a/internal/framework/checkpoint/filesystem_test.go +++ b/internal/framework/checkpoint/filesystem_test.go @@ -157,6 +157,7 @@ func TestFilesystemCheckpointRejectsIncompatibleManifests(t *testing.T) { want string }{ {"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"}, @@ -247,6 +248,17 @@ func TestFilesystemCheckpointReusesExtractWithRejections(t *testing.T) { } } +func TestFilesystemCheckpointDependencyDecisionIsBoundedAndCategorized(t *testing.T) { + fixture := seedFilesystemCheckpoints(t) + _, decision := fixture.loader.Extract("lane-a", "extract-module", []pipeline.CheckpointFingerprint{{Name: "source", Value: "sha256:changed"}}) + if decision.Reused || decision.Category != "dependency_invalidated" || decision.ReasonCode != "dependency_mismatch" { + t.Fatalf("dependency decision = %#v", decision) + } + if strings.Contains(decision.Reason, fixture.root) || strings.Contains(decision.Detail, fixture.root) { + t.Fatalf("dependency decision leaked checkpoint path: %#v", decision) + } +} + type checkpointStage struct { name string manifest func(filesystemCheckpointFixture) string diff --git a/internal/framework/checkpoint/identity.go b/internal/framework/checkpoint/identity.go index 7faafd0..3ea80a8 100644 --- a/internal/framework/checkpoint/identity.go +++ b/internal/framework/checkpoint/identity.go @@ -31,6 +31,7 @@ type Identity struct { Digest string `json:"digest"` PipelineID string `json:"pipeline_id"` PipelineDigest string `json:"pipeline_digest"` + PipelineTopology []string `json:"pipeline_topology,omitempty"` InputKey string `json:"input_key"` RawInputDigest string `json:"raw_input_digest,omitempty"` SourceDigest string `json:"source_digest,omitempty"` @@ -57,8 +58,8 @@ func NewIdentity(input IdentityInput) (Identity, error) { if strings.TrimSpace(input.RawInputDigest) == "" && strings.TrimSpace(input.SourceDigest) == "" { return Identity{}, fmt.Errorf("checkpoint identity raw input digest or source digest must be set") } - v := Identity{PipelineID: pipelineID, PipelineDigest: pipelineDigest, InputKey: inputKey, RawInputDigest: strings.TrimSpace(input.RawInputDigest), SourceDigest: strings.TrimSpace(input.SourceDigest), SelectedLanes: normalizedLanes(input.SelectedLanes, input.Pipeline.AllArtifactLanes()), RuntimeOverrides: normalizeIdentityFingerprints(input.RuntimeOverrides), ReferenceDigests: referenceFingerprints(input.References), ProvenanceFingerprints: normalizeIdentityFingerprints(input.ProvenanceFingerprints)} - data, err := json.Marshal(Identity{PipelineID: v.PipelineID, PipelineDigest: v.PipelineDigest, InputKey: v.InputKey, RawInputDigest: v.RawInputDigest, SourceDigest: v.SourceDigest, SelectedLanes: v.SelectedLanes, RuntimeOverrides: v.RuntimeOverrides, ReferenceDigests: v.ReferenceDigests, ProvenanceFingerprints: v.ProvenanceFingerprints}) + v := Identity{PipelineID: pipelineID, PipelineDigest: pipelineDigest, PipelineTopology: pipelineTopology(input.Pipeline), InputKey: inputKey, RawInputDigest: strings.TrimSpace(input.RawInputDigest), SourceDigest: strings.TrimSpace(input.SourceDigest), SelectedLanes: normalizedLanes(input.SelectedLanes, input.Pipeline.AllArtifactLanes()), RuntimeOverrides: normalizeIdentityFingerprints(input.RuntimeOverrides), ReferenceDigests: referenceFingerprints(input.References), ProvenanceFingerprints: normalizeIdentityFingerprints(input.ProvenanceFingerprints)} + data, err := json.Marshal(Identity{PipelineID: v.PipelineID, PipelineDigest: v.PipelineDigest, PipelineTopology: v.PipelineTopology, InputKey: v.InputKey, RawInputDigest: v.RawInputDigest, SourceDigest: v.SourceDigest, SelectedLanes: v.SelectedLanes, RuntimeOverrides: v.RuntimeOverrides, ReferenceDigests: v.ReferenceDigests, ProvenanceFingerprints: v.ProvenanceFingerprints}) if err != nil { return Identity{}, fmt.Errorf("marshal checkpoint identity: %w", err) } @@ -66,6 +67,19 @@ func NewIdentity(input IdentityInput) (Identity, error) { v.Digest = "sha256:" + hex.EncodeToString(sum[:]) return v, nil } + +func pipelineTopology(resolved pipeline.ResolvedPipeline) []string { + var topology []string + for _, step := range resolved.Steps { + for _, lane := range step.ArtifactLanes { + topology = append(topology, strings.TrimSpace(step.ID)+"\x00"+strings.TrimSpace(lane.ID)) + } + } + if len(topology) == 0 { + return nil + } + return topology +} func (i Identity) RelativePath() (string, error) { p, err := safeComponent(i.PipelineID) if err != nil { diff --git a/internal/framework/checkpoint/identity_test.go b/internal/framework/checkpoint/identity_test.go index 8db700e..8e03ab7 100644 --- a/internal/framework/checkpoint/identity_test.go +++ b/internal/framework/checkpoint/identity_test.go @@ -32,7 +32,9 @@ func TestNewIdentityNormalizesOrderAndEmptyValues(t *testing.T) { v.ProvenanceFingerprints = []Fingerprint{{Name: "source", Value: "v2"}, {Name: "runner", Value: "v1"}} }}, {"resolved lanes", func(v *IdentityInput) { - v.Pipeline.Steps[0].ArtifactLanes = []pipeline.ResolvedArtifactLane{v.Pipeline.Steps[0].ArtifactLanes[1], v.Pipeline.Steps[0].ArtifactLanes[0]} + steps := append([]pipeline.ResolvedPipelineStep(nil), v.Pipeline.Steps...) + steps[0].ArtifactLanes = []pipeline.ResolvedArtifactLane{steps[0].ArtifactLanes[1], steps[0].ArtifactLanes[0]} + v.Pipeline.Steps = steps }}, } { t.Run(tt.name, func(t *testing.T) { @@ -42,6 +44,12 @@ func TestNewIdentityNormalizesOrderAndEmptyValues(t *testing.T) { if err != nil { t.Fatal(err) } + if tt.name == "resolved lanes" { + if reflect.DeepEqual(identity, got) { + t.Fatal("reordered topology did not change checkpoint identity") + } + return + } if !reflect.DeepEqual(identity, got) { t.Fatalf("reordered identity differs:\nbase=%#v\ngot=%#v", identity, got) } diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index f6a3005..4bc5140 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -64,15 +64,19 @@ func (l *FilesystemLoader) Source(moduleKey string) (pipeline.SourceCheckpoint, } func (l *FilesystemLoader) Extract(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) { + return l.ExtractForStep("", laneID, moduleKey, dependencies) +} + +func (l *FilesystemLoader) ExtractForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.ExtractCheckpoint, pipeline.CheckpointDecision) { var manifest ExtractLaneManifest - if d := l.readJSON(laneManifestPath("extract", laneID), &manifest); !d.Reused { + if d := l.readJSON(laneManifestPath("extract", stepID, laneID), &manifest); !d.Reused { return pipeline.ExtractCheckpoint{}, d } - if d := l.validateLaneManifest(manifest.StageManifest, StageExtract, laneID, moduleKey, dependencies, StatusSucceeded, StatusSucceededWithRejections); !d.Reused { + if d := l.validateLaneManifest(manifest.StageManifest, StageExtract, stepID, laneID, moduleKey, dependencies, StatusSucceeded, StatusSucceededWithRejections); !d.Reused { return pipeline.ExtractCheckpoint{}, d } var payload artifactExtractEnvelope - if d := l.readJSON(lanePayloadPath("extract", laneID, "outputs.json"), &payload); !d.Reused { + if d := l.readJSON(lanePayloadPath("extract", stepID, laneID, "outputs.json"), &payload); !d.Reused { return pipeline.ExtractCheckpoint{}, d } outputs, err := artifactCheckpointOutputs(payload.Outputs) @@ -86,15 +90,19 @@ func (l *FilesystemLoader) Extract(laneID, moduleKey string, dependencies []pipe } func (l *FilesystemLoader) Merge(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) { + return l.MergeForStep("", laneID, moduleKey, dependencies) +} + +func (l *FilesystemLoader) MergeForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.MergeCheckpoint, pipeline.CheckpointDecision) { var manifest MergeLaneManifest - if d := l.readJSON(laneManifestPath("merge", laneID), &manifest); !d.Reused { + if d := l.readJSON(laneManifestPath("merge", stepID, laneID), &manifest); !d.Reused { return pipeline.MergeCheckpoint{}, d } - if d := l.validateLaneManifest(manifest.StageManifest, StageMerge, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused { + if d := l.validateLaneManifest(manifest.StageManifest, StageMerge, stepID, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused { return pipeline.MergeCheckpoint{}, d } var payload artifactSingleEnvelope - if d := l.readJSON(lanePayloadPath("merge", laneID, "output.json"), &payload); !d.Reused { + if d := l.readJSON(lanePayloadPath("merge", stepID, laneID, "output.json"), &payload); !d.Reused { return pipeline.MergeCheckpoint{}, d } values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output}) @@ -108,15 +116,19 @@ func (l *FilesystemLoader) Merge(laneID, moduleKey string, dependencies []pipeli } func (l *FilesystemLoader) Normalize(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) { + return l.NormalizeForStep("", laneID, moduleKey, dependencies) +} + +func (l *FilesystemLoader) NormalizeForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) { var manifest NormalizeLaneManifest - if d := l.readJSON(laneManifestPath("normalize", laneID), &manifest); !d.Reused { + if d := l.readJSON(laneManifestPath("normalize", stepID, laneID), &manifest); !d.Reused { return pipeline.NormalizeCheckpoint{}, d } - if d := l.validateLaneManifest(manifest.StageManifest, StageNormalize, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused { + if d := l.validateLaneManifest(manifest.StageManifest, StageNormalize, stepID, laneID, moduleKey, dependencies, StatusSucceeded); !d.Reused { return pipeline.NormalizeCheckpoint{}, d } var payload artifactSingleEnvelope - if d := l.readJSON(lanePayloadPath("normalize", laneID, "output.json"), &payload); !d.Reused { + if d := l.readJSON(lanePayloadPath("normalize", stepID, laneID, "output.json"), &payload); !d.Reused { return pipeline.NormalizeCheckpoint{}, d } values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output}) @@ -149,7 +161,7 @@ func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline. func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDecision { if !l.Enabled() { - return pipeline.CheckpointDecision{Reason: "checkpoint loading disabled"} + return pipeline.CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} } target, err := fileio.SafePath(l.root, name) if err != nil { @@ -158,7 +170,7 @@ func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDec data, err := os.ReadFile(target) if err != nil { if os.IsNotExist(err) { - return pipeline.CheckpointDecision{Reason: "checkpoint artifact is missing"} + return pipeline.CheckpointDecision{Category: "executed", ReasonCode: "checkpoint_missing", Reason: "checkpoint artifact is missing"} } return invalidDecision("read checkpoint artifact: %v", err) } @@ -169,13 +181,16 @@ func (l *FilesystemLoader) readJSON(name string, out any) pipeline.CheckpointDec } func (l *FilesystemLoader) validateManifest(manifest StageManifest, stage StageName, laneID string, moduleKey string, status StageStatus, dependencies []pipeline.CheckpointFingerprint) pipeline.CheckpointDecision { - return l.validateLaneManifest(manifest, stage, laneID, moduleKey, dependencies, status) + return l.validateLaneManifest(manifest, stage, "", laneID, moduleKey, dependencies, status) } -func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage StageName, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, statuses ...StageStatus) pipeline.CheckpointDecision { +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) } + if manifest.WorkspaceSchemaVersion == WorkspaceSchemaVersionV2 { + return invalidDecision("checkpoint workspace schema version %q is incompatible with %q and must be recomputed", manifest.WorkspaceSchemaVersion, WorkspaceSchemaVersion) + } if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion { return invalidDecision("checkpoint workspace schema version %q is not supported", manifest.WorkspaceSchemaVersion) } @@ -185,6 +200,9 @@ func (l *FilesystemLoader) validateLaneManifest(manifest StageManifest, stage St if manifest.Stage != stage { return invalidDecision("checkpoint stage %q does not match %q", manifest.Stage, stage) } + if strings.TrimSpace(stepID) != "" && manifest.StepID != stepID { + return invalidDecision("checkpoint step does not match requested step") + } if strings.TrimSpace(laneID) != "" && manifest.LaneID != laneID { return invalidDecision("checkpoint lane %q does not match %q", manifest.LaneID, laneID) } @@ -244,9 +262,65 @@ func fingerprintsEqual(a []pipeline.CheckpointFingerprint, b []pipeline.Checkpoi } func reusedDecision() pipeline.CheckpointDecision { - return pipeline.CheckpointDecision{Reused: true, Reason: "checkpoint is valid"} + return pipeline.CheckpointDecision{Reused: true, Category: "reused", ReasonCode: "checkpoint_valid", Reason: "checkpoint is valid"} } func invalidDecision(format string, args ...any) pipeline.CheckpointDecision { - return pipeline.CheckpointDecision{Reason: fmt.Sprintf(format, args...)} + 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 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 + ")" + } } diff --git a/internal/framework/checkpoint/manifest.go b/internal/framework/checkpoint/manifest.go index 654f392..6b4d27a 100644 --- a/internal/framework/checkpoint/manifest.go +++ b/internal/framework/checkpoint/manifest.go @@ -3,9 +3,8 @@ package checkpoint import "time" const ( - // These names and values are frozen checkpoint wire-compatibility - // identifiers. They intentionally retain the former terminology. - WorkspaceSchemaVersion = "notarius.workspace.v2" + WorkspaceSchemaVersion = "notarius.workspace.v3" + WorkspaceSchemaVersionV2 = "notarius.workspace.v2" WorkspaceSchemaVersionV1 = "notarius.workspace.v1" ) @@ -32,6 +31,7 @@ const ( type StageManifest struct { WorkspaceSchemaVersion string `json:"workspace_schema_version"` Stage StageName `json:"stage"` + StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id,omitempty"` ModuleKey string `json:"module_key,omitempty"` DependencyFingerprints []Fingerprint `json:"dependency_fingerprints,omitempty"` diff --git a/internal/framework/checkpoint/recorder.go b/internal/framework/checkpoint/recorder.go index 0fe7c7f..d66a508 100644 --- a/internal/framework/checkpoint/recorder.go +++ b/internal/framework/checkpoint/recorder.go @@ -71,93 +71,137 @@ func (r *FilesystemRecorder) SourceFailed(moduleKey string, err error) error { } func (r *FilesystemRecorder) ExtractRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { - manifest := r.laneManifest(StageExtract, StatusRunning, laneID, moduleKey, dependencies) + return r.ExtractRunningForStep("", laneID, moduleKey, dependencies) +} + +func (r *FilesystemRecorder) ExtractRunningForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := r.laneManifest(StageExtract, StatusRunning, stepID, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("extract", laneID), ExtractLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("extract", stepID, laneID), ExtractLaneManifest{StageManifest: manifest}) } func (r *FilesystemRecorder) ExtractSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, outputs []pipeline.CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + return r.ExtractSucceededForStep("", laneID, moduleKey, dependencies, outputs, rejected, warnings) +} + +func (r *FilesystemRecorder) ExtractSucceededForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, outputs []pipeline.CheckpointArtifact, 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 { + if err := r.writePayload(lanePayloadPath("extract", stepID, laneID, "outputs.json"), payload); err != nil { return err } - manifest := r.laneManifest(StageExtract, statusForRejected(rejected), laneID, moduleKey, dependencies) + manifest := r.laneManifest(StageExtract, statusForRejected(rejected), stepID, laneID, moduleKey, dependencies) manifest.OutputDigests = checkpointFingerprints(artifactOutputDigests(outputs)) manifest.ValidationStatus = validationStatusString(warnings, rejected) manifest.Rejections = rejectionSummaries(rejected) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("extract", laneID), ExtractLaneManifest{StageManifest: manifest, ChunkCount: len(outputs) + len(rejected), OutputCount: len(outputs)}) + return r.writeManifest(laneManifestPath("extract", stepID, laneID), ExtractLaneManifest{StageManifest: manifest, ChunkCount: len(outputs) + len(rejected), OutputCount: len(outputs)}) } func (r *FilesystemRecorder) ExtractFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := r.laneManifest(StageExtract, StatusFailed, laneID, moduleKey, dependencies) + return r.ExtractFailedForStep("", laneID, moduleKey, dependencies, err) +} + +func (r *FilesystemRecorder) ExtractFailedForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := r.laneManifest(StageExtract, StatusFailed, stepID, laneID, moduleKey, dependencies) manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) - return r.writeManifest(laneManifestPath("extract", laneID), ExtractLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("extract", stepID, laneID), ExtractLaneManifest{StageManifest: manifest}) } func (r *FilesystemRecorder) MergeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { - manifest := r.laneManifest(StageMerge, StatusRunning, laneID, moduleKey, dependencies) + return r.MergeRunningForStep("", laneID, moduleKey, dependencies) +} + +func (r *FilesystemRecorder) MergeRunningForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := r.laneManifest(StageMerge, StatusRunning, stepID, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("merge", stepID, laneID), MergeLaneManifest{StageManifest: manifest}) } func (r *FilesystemRecorder) MergeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { - if err := r.writePayload(lanePayloadPath("merge", laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { + return r.MergeSucceededForStep("", laneID, moduleKey, dependencies, output, warnings) +} + +func (r *FilesystemRecorder) MergeSucceededForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { + if err := r.writePayload(lanePayloadPath("merge", stepID, laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { return err } - manifest := r.laneManifest(StageMerge, StatusSucceeded, laneID, moduleKey, dependencies) + manifest := r.laneManifest(StageMerge, StatusSucceeded, stepID, laneID, moduleKey, dependencies) manifest.OutputDigests = checkpointFingerprints(artifactOutputDigests([]pipeline.CheckpointArtifact{output})) manifest.ValidationStatus = validationStatusString(warnings, nil) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("merge", stepID, laneID), MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } func (r *FilesystemRecorder) MergeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { - manifest := r.laneManifest(StageMerge, StatusSucceededWithRejections, laneID, moduleKey, dependencies) + return r.MergeRejectedForStep("", laneID, moduleKey, dependencies, rejected) +} + +func (r *FilesystemRecorder) MergeRejectedForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { + manifest := r.laneManifest(StageMerge, StatusSucceededWithRejections, stepID, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("merge", stepID, laneID), MergeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } func (r *FilesystemRecorder) MergeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := r.laneManifest(StageMerge, StatusFailed, laneID, moduleKey, dependencies) + return r.MergeFailedForStep("", laneID, moduleKey, dependencies, err) +} + +func (r *FilesystemRecorder) MergeFailedForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := r.laneManifest(StageMerge, StatusFailed, stepID, laneID, moduleKey, dependencies) manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) - return r.writeManifest(laneManifestPath("merge", laneID), MergeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("merge", stepID, laneID), MergeLaneManifest{StageManifest: manifest}) } func (r *FilesystemRecorder) NormalizeRunning(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { - manifest := r.laneManifest(StageNormalize, StatusRunning, laneID, moduleKey, dependencies) + return r.NormalizeRunningForStep("", laneID, moduleKey, dependencies) +} + +func (r *FilesystemRecorder) NormalizeRunningForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) error { + manifest := r.laneManifest(StageNormalize, StatusRunning, stepID, laneID, moduleKey, dependencies) manifest.StartedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("normalize", stepID, laneID), NormalizeLaneManifest{StageManifest: manifest}) } func (r *FilesystemRecorder) NormalizeSucceeded(laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { - if err := r.writePayload(lanePayloadPath("normalize", laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { + return r.NormalizeSucceededForStep("", laneID, moduleKey, dependencies, output, warnings) +} + +func (r *FilesystemRecorder) NormalizeSucceededForStep(stepID, laneID, moduleKey string, dependencies []pipeline.CheckpointFingerprint, output pipeline.CheckpointArtifact, warnings []contracts.Warning) error { + if err := r.writePayload(lanePayloadPath("normalize", stepID, laneID, "output.json"), artifactSingleEnvelope{Output: artifactCheckpointEnvelopeFromOutput(output), Warnings: cloneWarnings(warnings)}); err != nil { return err } - manifest := r.laneManifest(StageNormalize, StatusSucceeded, laneID, moduleKey, dependencies) + manifest := r.laneManifest(StageNormalize, StatusSucceeded, stepID, laneID, moduleKey, dependencies) manifest.OutputDigests = checkpointFingerprints(artifactOutputDigests([]pipeline.CheckpointArtifact{output})) manifest.ValidationStatus = validationStatusString(warnings, nil) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("normalize", stepID, laneID), NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } func (r *FilesystemRecorder) NormalizeRejected(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { - manifest := r.laneManifest(StageNormalize, StatusSucceededWithRejections, laneID, moduleKey, dependencies) + return r.NormalizeRejectedForStep("", laneID, moduleKey, dependencies, rejected) +} + +func (r *FilesystemRecorder) NormalizeRejectedForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, rejected contracts.RejectedOutput) error { + manifest := r.laneManifest(StageNormalize, StatusSucceededWithRejections, stepID, laneID, moduleKey, dependencies) manifest.ValidationStatus = "rejected" manifest.Rejections = rejectionSummaries([]contracts.RejectedOutput{rejected}) manifest.CompletedAt = timePtr(r.timestamp()) - return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) + return r.writeManifest(laneManifestPath("normalize", stepID, laneID), NormalizeLaneManifest{StageManifest: manifest, InputCount: len(dependencies)}) } func (r *FilesystemRecorder) NormalizeFailed(laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { - manifest := r.laneManifest(StageNormalize, StatusFailed, laneID, moduleKey, dependencies) + return r.NormalizeFailedForStep("", laneID, moduleKey, dependencies, err) +} + +func (r *FilesystemRecorder) NormalizeFailedForStep(stepID, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint, err error) error { + manifest := r.laneManifest(StageNormalize, StatusFailed, stepID, laneID, moduleKey, dependencies) manifest.CompletedAt = timePtr(r.timestamp()) manifest.Metadata = errorMetadata(err) - return r.writeManifest(laneManifestPath("normalize", laneID), NormalizeLaneManifest{StageManifest: manifest}) + return r.writeManifest(laneManifestPath("normalize", stepID, laneID), NormalizeLaneManifest{StageManifest: manifest}) } func (r *FilesystemRecorder) writeManifest(name string, payload any) error { @@ -190,8 +234,9 @@ func (r *FilesystemRecorder) newStageManifest(stage StageName, status StageStatu return manifest } -func (r *FilesystemRecorder) laneManifest(stage StageName, status StageStatus, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) StageManifest { +func (r *FilesystemRecorder) laneManifest(stage StageName, status StageStatus, stepID string, laneID string, moduleKey string, dependencies []pipeline.CheckpointFingerprint) StageManifest { manifest := r.newStageManifest(stage, status) + manifest.StepID = stepID manifest.LaneID = laneID manifest.ModuleKey = moduleKey manifest.DependencyFingerprints = checkpointFingerprints(dependencies) @@ -424,12 +469,15 @@ func errorMetadata(err error) map[string]string { return map[string]string{"error": err.Error()} } -func laneManifestPath(stage string, laneID string) string { - return lanePayloadPath(stage, laneID, "manifest.json") +func laneManifestPath(stage string, stepID string, laneID string) string { + return lanePayloadPath(stage, stepID, laneID, "manifest.json") } -func lanePayloadPath(stage string, laneID string, file string) string { - return path.Join(stage, checkpointPathComponent(laneID), file) +func lanePayloadPath(stage string, stepID string, laneID string, file string) string { + if strings.TrimSpace(stepID) == "" { + return path.Join(stage, checkpointPathComponent(laneID), file) + } + return path.Join(stage, checkpointPathComponent(stepID), checkpointPathComponent(laneID), file) } func checkpointPathComponent(value string) string { diff --git a/internal/framework/checkpoint/recorder_test.go b/internal/framework/checkpoint/recorder_test.go index 5297e7e..7e7116e 100644 --- a/internal/framework/checkpoint/recorder_test.go +++ b/internal/framework/checkpoint/recorder_test.go @@ -35,9 +35,40 @@ func TestRootBasedRecorderOutputIsReusable(t *testing.T) { } } -func TestCheckpointSchemaCompatibilityIsUnchanged(t *testing.T) { - if WorkspaceSchemaVersion != "notarius.workspace.v2" || WorkspaceSchemaVersionV1 != "notarius.workspace.v1" { - t.Fatal("checkpoint schema identifiers changed") +func TestStepAwareRecorderAndLoaderIsolateLaneState(t *testing.T) { + root := t.TempDir() + identity := testIdentity(t) + recorder, err := NewFilesystemRecorder(root, identity) + if err != nil { + t.Fatal(err) + } + if err := recorder.(pipeline.StepCheckpointRecorder).ExtractSucceededForStep("step-a", "lane", "module", nil, nil, nil, nil); err != nil { + t.Fatal(err) + } + loader, err := NewFilesystemLoader(root, identity) + if err != nil { + t.Fatal(err) + } + stepLoader := loader.(pipeline.StepCheckpointLoader) + if _, decision := stepLoader.ExtractForStep("step-b", "lane", "module", nil); decision.Reused { + t.Fatal("checkpoint from another step was reused") + } + loaded, decision := stepLoader.ExtractForStep("step-a", "lane", "module", nil) + if !decision.Reused || len(loaded.Outputs) != 0 { + t.Fatalf("step-aware load = %#v, decision=%#v", loaded, decision) + } + relative, err := identity.RelativePath() + if err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(root, relative, "extract", "step-a", "lane", "manifest.json")); err != nil { + t.Fatal(err) + } +} + +func TestCheckpointSchemaCompatibilityIdentifiers(t *testing.T) { + if WorkspaceSchemaVersion != "notarius.workspace.v3" || WorkspaceSchemaVersionV2 != "notarius.workspace.v2" || WorkspaceSchemaVersionV1 != "notarius.workspace.v1" { + t.Fatal("checkpoint schema identifiers are incorrect") } } diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 7803848..a126ed5 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -40,17 +40,76 @@ type CheckpointRecorder interface { NormalizeFailed(laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error } +// StepCheckpointRecorder is implemented by checkpoint stores that isolate +// lane artifacts by their ordered pipeline step. The legacy recorder methods +// remain available for callers that do not have step context. +type StepCheckpointRecorder interface { + ExtractRunningForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint) error + ExtractSucceededForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error + ExtractFailedForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error + MergeRunningForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint) error + MergeSucceededForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error + MergeRejectedForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, rejected contracts.RejectedOutput) error + MergeFailedForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error + NormalizeRunningForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint) error + NormalizeSucceededForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error + NormalizeRejectedForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, rejected contracts.RejectedOutput) error + NormalizeFailedForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint, err error) error +} + type CheckpointDecision struct { - Reused bool `json:"reused"` + Reused bool `json:"reused"` + Category string `json:"category,omitempty"` + ReasonCode string `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"` } type CheckpointEvent struct { - Stage string `json:"stage"` - LaneID string `json:"lane_id,omitempty"` - ModuleKey string `json:"module_key,omitempty"` - Action string `json:"action"` - 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 string `json:"action"` + Category string `json:"category,omitempty"` + ReasonCode string `json:"reason_code,omitempty"` + Detail string `json:"detail,omitempty"` + Reason string `json:"reason,omitempty"` +} + +type CheckpointExecutionPolicy struct { + ForcedLanes map[string]struct{} + RequireReusableLanes map[string]struct{} +} + +func CheckpointLaneKey(stepID, laneID string) string { + return strings.TrimSpace(stepID) + "\x00" + strings.TrimSpace(laneID) +} + +func (policy CheckpointExecutionPolicy) forced(stepID, laneID string) bool { + _, ok := policy.ForcedLanes[CheckpointLaneKey(stepID, laneID)] + return ok +} + +func (policy CheckpointExecutionPolicy) requiresReusable(stepID, laneID string) bool { + _, ok := policy.RequireReusableLanes[CheckpointLaneKey(stepID, laneID)] + return ok +} + +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 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 nil } type SourceCheckpoint struct { @@ -93,6 +152,15 @@ type CheckpointLoader interface { Normalize(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) } +// StepCheckpointLoader is the step-aware counterpart used by the persistent +// checkpoint implementation. Loaders without this optional interface remain +// usable by framework callers and test doubles through the legacy methods. +type StepCheckpointLoader interface { + ExtractForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) + MergeForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) + NormalizeForStep(stepID, laneID string, moduleKey string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) +} + type noopCheckpointRecorder struct{} type noopCheckpointLoader struct{} @@ -136,16 +204,83 @@ func (noopCheckpointRecorder) NormalizeFailed(string, string, []CheckpointFinger func (noopCheckpointLoader) Enabled() bool { return false } func (noopCheckpointLoader) Source(string) (SourceCheckpoint, CheckpointDecision) { - return SourceCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} + return SourceCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} } func (noopCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { - return ExtractCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} + return ExtractCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} } func (noopCheckpointLoader) Merge(string, string, []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { - return MergeCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} + return MergeCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} } func (noopCheckpointLoader) Normalize(string, string, []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { - return NormalizeCheckpoint{}, CheckpointDecision{Reason: "checkpoint loading disabled"} + return NormalizeCheckpoint{}, CheckpointDecision{Category: "executed", ReasonCode: "loading_disabled", Reason: "checkpoint loading disabled"} +} + +func checkpointExtractRunning(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.ExtractRunningForStep(stepID, laneID, moduleKey, deps) + } + return recorder.ExtractRunning(laneID, moduleKey, deps) +} +func checkpointExtractSucceeded(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.ExtractSucceededForStep(stepID, laneID, moduleKey, deps, outputs, rejected, warnings) + } + return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings) +} +func checkpointExtractFailed(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, err error) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.ExtractFailedForStep(stepID, laneID, moduleKey, deps, err) + } + return recorder.ExtractFailed(laneID, moduleKey, deps, err) +} +func checkpointMergeRunning(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.MergeRunningForStep(stepID, laneID, moduleKey, deps) + } + return recorder.MergeRunning(laneID, moduleKey, deps) +} +func checkpointMergeSucceeded(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.MergeSucceededForStep(stepID, laneID, moduleKey, deps, output, warnings) + } + return recorder.MergeSucceeded(laneID, moduleKey, deps, output, warnings) +} +func checkpointMergeRejected(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, rejected contracts.RejectedOutput) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.MergeRejectedForStep(stepID, laneID, moduleKey, deps, rejected) + } + return recorder.MergeRejected(laneID, moduleKey, deps, rejected) +} +func checkpointMergeFailed(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, err error) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.MergeFailedForStep(stepID, laneID, moduleKey, deps, err) + } + return recorder.MergeFailed(laneID, moduleKey, deps, err) +} +func checkpointNormalizeRunning(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.NormalizeRunningForStep(stepID, laneID, moduleKey, deps) + } + return recorder.NormalizeRunning(laneID, moduleKey, deps) +} +func checkpointNormalizeSucceeded(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.NormalizeSucceededForStep(stepID, laneID, moduleKey, deps, output, warnings) + } + return recorder.NormalizeSucceeded(laneID, moduleKey, deps, output, warnings) +} +func checkpointNormalizeRejected(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, rejected contracts.RejectedOutput) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.NormalizeRejectedForStep(stepID, laneID, moduleKey, deps, rejected) + } + return recorder.NormalizeRejected(laneID, moduleKey, deps, rejected) +} +func checkpointNormalizeFailed(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, err error) error { + if stepAware, ok := recorder.(StepCheckpointRecorder); ok { + return stepAware.NormalizeFailedForStep(stepID, laneID, moduleKey, deps, err) + } + return recorder.NormalizeFailed(laneID, moduleKey, deps, err) } func digestFingerprints(name string, digest string) []CheckpointFingerprint { diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 28174ee..f5ad699 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -38,21 +38,22 @@ func New() *Runner { } type RunInput struct { - Prepared *PreparedPipeline - SourceID string - Path string - RawInput []byte - SessionID string - RunID string - StartedAt time.Time - LLMProfiles []artifacts.LLMProfileManifest - Metadata map[string]any - Warnings []contracts.Warning - ChunkCacheMode ChunkCacheMode - ChunkPlans ChunkPlanStore - Checkpoints CheckpointRecorder - Checkpoint CheckpointLoader - Debug DebugRecorder + Prepared *PreparedPipeline + SourceID string + Path string + RawInput []byte + SessionID string + RunID string + StartedAt time.Time + LLMProfiles []artifacts.LLMProfileManifest + Metadata map[string]any + Warnings []contracts.Warning + ChunkCacheMode ChunkCacheMode + ChunkPlans ChunkPlanStore + Checkpoints CheckpointRecorder + Checkpoint CheckpointLoader + CheckpointPolicy CheckpointExecutionPolicy + Debug DebugRecorder // ExtractWorkers bounds run-wide extract jobs. Values less than one use a // single worker so direct framework callers retain deterministic behavior. ExtractWorkers int @@ -131,7 +132,7 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err return failOutput(output), err } sourceCheckpoint, sourceDecision := checkpointLoader.Source(adapter.Key()) - recordCheckpointEvent(&output, checkpointLoader, "source", "", adapter.Key(), sourceDecision) + recordCheckpointEvent(&output, checkpointLoader, "source", "", "", adapter.Key(), sourceDecision) doc := sourceCheckpoint.Document sourceStarted := time.Now().UTC() if err := writeDebugTimed(debugRecorder, "source/input.json", debugTimedEnvelope{ @@ -597,29 +598,47 @@ func failOutput(output RunOutput) RunOutput { return output } -func recordCheckpointEvent(output *RunOutput, loader CheckpointLoader, stage string, laneID string, moduleKey string, decision CheckpointDecision) { +func recordCheckpointEvent(output *RunOutput, loader CheckpointLoader, stage string, stepID string, laneID string, moduleKey string, decision CheckpointDecision) { if output == nil || loader == nil || !loader.Enabled() { return } - action := "executed" - if decision.Reused { - action = "reused" - } + action := checkpointDecisionCategory(decision) output.CheckpointEvents = append(output.CheckpointEvents, CheckpointEvent{ - Stage: stage, - LaneID: laneID, - ModuleKey: moduleKey, - Action: action, - Reason: decision.Reason, + Stage: stage, + StepID: stepID, + LaneID: laneID, + ModuleKey: moduleKey, + Action: action, + Category: checkpointDecisionCategory(decision), + ReasonCode: decision.ReasonCode, + Detail: decision.Detail, + Reason: decision.Reason, }) } +func checkpointDecisionCategory(decision CheckpointDecision) string { + if decision.Category != "" { + return decision.Category + } + if decision.Reused { + return "reused" + } + return "executed" +} + func populateOutputManifest(output *RunOutput) { if output == nil { return } output.Manifest.NormalizedOutputs = normalizedOutputManifests(output.NormalizeOutputs) output.Manifest.RejectedOutputs = rejectedOutputManifests(output.Rejected) + 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}) + } + output.Manifest.CheckpointDecisions = decisions + } } func normalizedOutputManifests(outputs []contracts.SerializedOutput) []artifacts.NormalizedOutputManifest { diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 8eddc08..341e2f5 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -36,12 +36,15 @@ type finalizedExtractResults struct { decision CheckpointDecision } -func loadExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { +func loadExtract(loader CheckpointLoader, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + if stepAware, ok := loader.(StepCheckpointLoader); ok { + return stepAware.ExtractForStep(stepID, laneID, moduleKey, deps) + } return loader.Extract(laneID, moduleKey, deps) } -func recordExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { - return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings) +func recordExtract(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + return checkpointExtractSucceeded(recorder, stepID, laneID, moduleKey, deps, outputs, rejected, warnings) } type extractJob struct { @@ -87,10 +90,10 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, step PreparedP return output, err } if !state.decision.Reused { - if err := checkpoints.ExtractRunning(prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil { + if err := checkpointExtractRunning(checkpoints, input.stepID, prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil { return output, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err) } - } else if err := finalizeLaneExtract(checkpoints, state); err != nil { + } else if err := finalizeLaneExtract(checkpoints, input.stepID, state); err != nil { return output, err } states[i] = state @@ -188,13 +191,13 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, step PreparedP if result.err != nil { state.failed = true runErrors = append(runErrors, orderedRunError{stage: 0, lane: state.index, chunk: result.chunkIndex, err: result.err}) - _ = checkpoints.ExtractFailed(state.prepared.resolved.ID, state.prepared.resolved.Extract.Module, state.deps, result.err) + _ = checkpointExtractFailed(checkpoints, input.stepID, state.prepared.resolved.ID, state.prepared.resolved.Extract.Module, state.deps, result.err) cancel() } else { state.results[result.chunkIndex] = result } if state.remaining == 0 && !state.failed && ctx.Err() == nil { - if err := finalizeLaneExtract(checkpoints, state); err != nil { + if err := finalizeLaneExtract(checkpoints, input.stepID, state); err != nil { state.failed = true runErrors = append(runErrors, orderedRunError{stage: 0, lane: state.index, chunk: len(chunks), err: err}) cancel() @@ -233,27 +236,31 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou extractReferences := operationReferenceSet(input, lane.ExtractReferences) 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, lane.ID, lane.Extract.Module, state.deps) + cp, decision := loadExtract(loader, input.stepID, lane.ID, lane.Extract.Module, state.deps) + decision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, decision) + if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, decision); err != nil { + return nil, err + } 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()} + if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, stored); decodeErr != nil { + decision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored extract artifact failed canonical codec validation", Reason: "extract artifact checkpoint is not canonical"} break } } } + if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, decision); err != nil { + return nil, err + } state.decision = decision if decision.Reused { state.remaining = 0 for _, stored := range cp.Outputs { - value, decodeErr := decodeCheckpointArtifact(typed.codec, stored) + value, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, stored) if decodeErr != nil { return nil, fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr) } - stored, decodeErr = hydrateCheckpointArtifact(typed.codec, stored, value) - if decodeErr != nil { - return nil, fmt.Errorf("hydrate extract checkpoint for lane %q: %w", lane.ID, decodeErr) - } + stored = hydrated 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 @@ -330,7 +337,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. return result } -func finalizeLaneExtract(checkpoints CheckpointRecorder, state *laneExtractState) error { +func finalizeLaneExtract(checkpoints CheckpointRecorder, stepID string, state *laneExtractState) error { lane := state.prepared.resolved indexes := make([]int, 0, len(state.results)) for index := range state.results { @@ -351,7 +358,7 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, state *laneExtractState sort.SliceStable(state.serialized, func(i, j int) bool { return state.serialized[i].ChunkIndex < state.serialized[j].ChunkIndex }) sort.SliceStable(state.rejected, func(i, j int) bool { return state.rejected[i].ChunkIndex < state.rejected[j].ChunkIndex }) if !state.decision.Reused { - if err := recordExtract(checkpoints, lane.ID, lane.Extract.Module, state.deps, state.serialized, state.rejected, state.warnings); err != nil { + if err := recordExtract(checkpoints, stepID, lane.ID, lane.Extract.Module, state.deps, state.serialized, state.rejected, state.warnings); err != nil { return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err) } } @@ -370,7 +377,7 @@ 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), lane.ID, lane.Extract.Module, results.decision) + 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_typed.go b/internal/framework/pipeline/runner_typed.go index 15d8cc1..53c79eb 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -13,17 +13,23 @@ import ( "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" ) -func loadMerge(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { +func loadMerge(loader CheckpointLoader, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { + if stepAware, ok := loader.(StepCheckpointLoader); ok { + return stepAware.MergeForStep(stepID, laneID, moduleKey, deps) + } return loader.Merge(laneID, moduleKey, deps) } -func loadNormalize(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { +func loadNormalize(loader CheckpointLoader, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { + if stepAware, ok := loader.(StepCheckpointLoader); ok { + return stepAware.NormalizeForStep(stepID, laneID, moduleKey, deps) + } return loader.Normalize(laneID, moduleKey, deps) } -func recordMerge(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { - return recorder.MergeSucceeded(laneID, moduleKey, deps, output, warnings) +func recordMerge(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { + return checkpointMergeSucceeded(recorder, stepID, laneID, moduleKey, deps, output, warnings) } -func recordNormalize(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { - return recorder.NormalizeSucceeded(laneID, moduleKey, deps, output, warnings) +func recordNormalize(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { + return checkpointNormalizeSucceeded(recorder, stepID, laneID, moduleKey, deps, output, warnings) } func cloneCheckpointArtifact(output CheckpointArtifact) CheckpointArtifact { output.Artifact = contracts.CloneSerializedArtifact(output.Artifact) @@ -111,6 +117,25 @@ func decodeCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtif return codec.decode(append([]byte(nil), artifact.Artifact.Content...)) } +func decodeCanonicalCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtifact) (any, CheckpointArtifact, error) { + value, err := decodeCheckpointArtifact(codec, artifact) + if err != nil { + return nil, CheckpointArtifact{}, err + } + canonical, err := serializeArtifact(codec, value, false) + if err != nil { + return nil, CheckpointArtifact{}, 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") + } + hydrated, err := hydrateCheckpointArtifact(codec, cloneCheckpointArtifact(artifact), value) + if err != nil { + return nil, CheckpointArtifact{}, err + } + return value, hydrated, nil +} + func checkpointArtifact(codec artifactCodecEntry, laneID, moduleKey, sourceID string, value any) (CheckpointArtifact, error) { serialized, err := serializeArtifact(codec, value, false) if err != nil { @@ -157,13 +182,20 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi mergeReferences := operationReferenceSet(input, lane.MergeReferences) mergeDeps := append(artifactCheckpointDigests(extracts.serialized), generatedReferenceDependencies(mergeReferences)...) mergeDeps = normalizeCheckpointFingerprints(mergeDeps) - mergeCP, mergeDecision := loadMerge(loader, lane.ID, lane.Merge.Module, mergeDeps) + mergeCP, mergeDecision := loadMerge(loader, input.stepID, lane.ID, lane.Merge.Module, mergeDeps) + mergeDecision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision) + if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision); err != nil { + return err + } if mergeDecision.Reused { - if _, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil { - mergeDecision = CheckpointDecision{Reason: "merge artifact checkpoint codec is incompatible: " + decodeErr.Error()} + if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil { + mergeDecision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored merge artifact failed canonical codec validation", Reason: "merge artifact checkpoint is not canonical"} } } - recordCheckpointEvent(output, loader, string(StageMerge), lane.ID, lane.Merge.Module, mergeDecision) + if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision); err != nil { + return err + } + recordCheckpointEvent(output, loader, string(StageMerge), input.stepID, lane.ID, lane.Merge.Module, mergeDecision) if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, 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": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return err } @@ -171,19 +203,16 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi var serializedMerge CheckpointArtifact var mergeWarnings []contracts.Warning if mergeDecision.Reused { - value, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output) + value, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(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} - serializedMerge, decodeErr = hydrateCheckpointArtifact(typed.codec, cloneCheckpointArtifact(mergeCP.Output), value) - if decodeErr != nil { - return fmt.Errorf("hydrate merge checkpoint for lane %q: %w", lane.ID, decodeErr) - } + serializedMerge = hydrated mergeWarnings = cloneWarnings(mergeCP.Warnings) output.Warnings = append(output.Warnings, mergeWarnings...) } else { - if err := checkpoints.MergeRunning(lane.ID, lane.Merge.Module, mergeDeps); err != nil { + if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil { return err } ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { @@ -226,18 +255,18 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi return true, nil, nil }) if runErr != nil { - _ = checkpoints.MergeFailed(lane.ID, lane.Merge.Module, mergeDeps, runErr) + _ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, runErr) return runErr } if !ok { output.Rejected = append(output.Rejected, *rejection) - if err := checkpoints.MergeRejected(lane.ID, lane.Merge.Module, mergeDeps, *rejection); err != nil { + if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejection); err != nil { return err } return nil } output.Warnings = append(output.Warnings, mergeWarnings...) - if err := recordMerge(checkpoints, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil { + if err := recordMerge(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil { return err } } @@ -249,31 +278,35 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi normalizeReferences := operationReferenceSet(input, lane.NormalizeReferences) normalizeDeps := append(artifactCheckpointDigests([]CheckpointArtifact{serializedMerge}), generatedReferenceDependencies(normalizeReferences)...) normalizeDeps = normalizeCheckpointFingerprints(normalizeDeps) - normalizeCP, normalizeDecision := loadNormalize(loader, lane.ID, lane.Normalize.Module, normalizeDeps) + normalizeCP, normalizeDecision := loadNormalize(loader, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps) + normalizeDecision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision) + if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision); err != nil { + return err + } if normalizeDecision.Reused { - if _, decodeErr := decodeCheckpointArtifact(typed.codec, normalizeCP.Output); decodeErr != nil { - normalizeDecision = CheckpointDecision{Reason: "normalize artifact checkpoint codec is incompatible: " + decodeErr.Error()} + if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, normalizeCP.Output); decodeErr != nil { + normalizeDecision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored normalize artifact failed canonical codec validation", Reason: "normalize artifact checkpoint is not canonical"} } } - recordCheckpointEvent(output, loader, string(StageNormalize), lane.ID, lane.Normalize.Module, normalizeDecision) + if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision); err != nil { + return err + } + recordCheckpointEvent(output, loader, string(StageNormalize), input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision) if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, 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": debugCheckpointArtifact(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil { return err } var serializedNormalize CheckpointArtifact var normalizeWarnings []contracts.Warning if normalizeDecision.Reused { - value, decodeErr := decodeCheckpointArtifact(typed.codec, normalizeCP.Output) + _, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, normalizeCP.Output) if decodeErr != nil { return fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr) } - serializedNormalize, decodeErr = hydrateCheckpointArtifact(typed.codec, cloneCheckpointArtifact(normalizeCP.Output), value) - if decodeErr != nil { - return fmt.Errorf("hydrate normalize checkpoint for lane %q: %w", lane.ID, decodeErr) - } + serializedNormalize = hydrated normalizeWarnings = cloneWarnings(normalizeCP.Warnings) output.Warnings = append(output.Warnings, normalizeWarnings...) } else { - if err := checkpoints.NormalizeRunning(lane.ID, lane.Normalize.Module, normalizeDeps); err != nil { + if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil { return err } ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) { @@ -315,18 +348,18 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi return true, nil, nil }) if runErr != nil { - _ = checkpoints.NormalizeFailed(lane.ID, lane.Normalize.Module, normalizeDeps, runErr) + _ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, runErr) return runErr } if !ok { output.Rejected = append(output.Rejected, *rejection) - if err := checkpoints.NormalizeRejected(lane.ID, lane.Normalize.Module, normalizeDeps, *rejection); err != nil { + if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejection); err != nil { return err } return nil } output.Warnings = append(output.Warnings, normalizeWarnings...) - if err := recordNormalize(checkpoints, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil { + if err := recordNormalize(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil { return err } } diff --git a/internal/framework/pipeline/runner_typed_checkpoint_test.go b/internal/framework/pipeline/runner_typed_checkpoint_test.go index 70396f3..50d4e9c 100644 --- a/internal/framework/pipeline/runner_typed_checkpoint_test.go +++ b/internal/framework/pipeline/runner_typed_checkpoint_test.go @@ -43,3 +43,26 @@ func TestDecodeCheckpointArtifactRejectsIncompatibleCodecIdentityAndBytes(t *tes }) } } + +func TestDecodeCanonicalCheckpointArtifactRejectsNonCanonicalBytes(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) + } + stored := CheckpointArtifact{Artifact: artifact, SchemaDigest: contracts.DigestArtifactSchema(artifact.Schema)} + if _, _, err := decodeCanonicalCheckpointArtifact(codec, stored); err != nil { + t.Fatalf("canonical artifact rejected: %v", err) + } + stored.Artifact.Content = []byte(`{"items": ["one"]}`) + if _, _, err := decodeCanonicalCheckpointArtifact(codec, stored); err == nil { + t.Fatal("non-canonical artifact was accepted") + } +} diff --git a/internal/framework/pipeline/synchronized_collaborators.go b/internal/framework/pipeline/synchronized_collaborators.go index d818400..1b92f46 100644 --- a/internal/framework/pipeline/synchronized_collaborators.go +++ b/internal/framework/pipeline/synchronized_collaborators.go @@ -76,6 +76,30 @@ func (l *lockedCheckpointLoader) Normalize(lane, key string, deps []CheckpointFi defer l.mu.Unlock() return l.inner.Normalize(lane, key, deps) } +func (l *lockedCheckpointLoader) ExtractForStep(step, lane, key string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + l.mu.Lock() + defer l.mu.Unlock() + if stepAware, ok := l.inner.(StepCheckpointLoader); ok { + return stepAware.ExtractForStep(step, lane, key, deps) + } + return l.inner.Extract(lane, key, deps) +} +func (l *lockedCheckpointLoader) MergeForStep(step, lane, key string, deps []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { + l.mu.Lock() + defer l.mu.Unlock() + if stepAware, ok := l.inner.(StepCheckpointLoader); ok { + return stepAware.MergeForStep(step, lane, key, deps) + } + return l.inner.Merge(lane, key, deps) +} +func (l *lockedCheckpointLoader) NormalizeForStep(step, lane, key string, deps []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { + l.mu.Lock() + defer l.mu.Unlock() + if stepAware, ok := l.inner.(StepCheckpointLoader); ok { + return stepAware.NormalizeForStep(step, lane, key, deps) + } + return l.inner.Normalize(lane, key, deps) +} type lockedCheckpointRecorder struct { inner CheckpointRecorder @@ -135,3 +159,92 @@ func (r *lockedCheckpointRecorder) NormalizeRejected(lane, key string, deps []Ch func (r *lockedCheckpointRecorder) NormalizeFailed(lane, key string, deps []CheckpointFingerprint, err error) error { return r.call(func() error { return r.inner.NormalizeFailed(lane, key, deps, err) }) } + +func (r *lockedCheckpointRecorder) ExtractRunningForStep(step, lane, key string, deps []CheckpointFingerprint) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.ExtractRunningForStep(step, lane, key, deps) + } + return r.inner.ExtractRunning(lane, key, deps) + }) +} +func (r *lockedCheckpointRecorder) ExtractSucceededForStep(step, lane, key string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.ExtractSucceededForStep(step, lane, key, deps, outputs, rejected, warnings) + } + return r.inner.ExtractSucceeded(lane, key, deps, outputs, rejected, warnings) + }) +} +func (r *lockedCheckpointRecorder) ExtractFailedForStep(step, lane, key string, deps []CheckpointFingerprint, err error) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.ExtractFailedForStep(step, lane, key, deps, err) + } + return r.inner.ExtractFailed(lane, key, deps, err) + }) +} +func (r *lockedCheckpointRecorder) MergeRunningForStep(step, lane, key string, deps []CheckpointFingerprint) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.MergeRunningForStep(step, lane, key, deps) + } + return r.inner.MergeRunning(lane, key, deps) + }) +} +func (r *lockedCheckpointRecorder) MergeSucceededForStep(step, lane, key string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.MergeSucceededForStep(step, lane, key, deps, output, warnings) + } + return r.inner.MergeSucceeded(lane, key, deps, output, warnings) + }) +} +func (r *lockedCheckpointRecorder) MergeRejectedForStep(step, lane, key string, deps []CheckpointFingerprint, rejected contracts.RejectedOutput) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.MergeRejectedForStep(step, lane, key, deps, rejected) + } + return r.inner.MergeRejected(lane, key, deps, rejected) + }) +} +func (r *lockedCheckpointRecorder) MergeFailedForStep(step, lane, key string, deps []CheckpointFingerprint, err error) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.MergeFailedForStep(step, lane, key, deps, err) + } + return r.inner.MergeFailed(lane, key, deps, err) + }) +} +func (r *lockedCheckpointRecorder) NormalizeRunningForStep(step, lane, key string, deps []CheckpointFingerprint) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.NormalizeRunningForStep(step, lane, key, deps) + } + return r.inner.NormalizeRunning(lane, key, deps) + }) +} +func (r *lockedCheckpointRecorder) NormalizeSucceededForStep(step, lane, key string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.NormalizeSucceededForStep(step, lane, key, deps, output, warnings) + } + return r.inner.NormalizeSucceeded(lane, key, deps, output, warnings) + }) +} +func (r *lockedCheckpointRecorder) NormalizeRejectedForStep(step, lane, key string, deps []CheckpointFingerprint, rejected contracts.RejectedOutput) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.NormalizeRejectedForStep(step, lane, key, deps, rejected) + } + return r.inner.NormalizeRejected(lane, key, deps, rejected) + }) +} +func (r *lockedCheckpointRecorder) NormalizeFailedForStep(step, lane, key string, deps []CheckpointFingerprint, err error) error { + return r.call(func() error { + if v, ok := r.inner.(StepCheckpointRecorder); ok { + return v.NormalizeFailedForStep(step, lane, key, deps, err) + } + return r.inner.NormalizeFailed(lane, key, deps, err) + }) +}