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

Implement selective checkpoint recomputation

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-21 21:59:48 +00:00
parent 22d4f29670
commit c437682407
17 changed files with 902 additions and 180 deletions
Download Patch File Download Diff File Expand all files Collapse all files

56
internal/cli/recompute_policy_test.go Normal file
View File

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

171
internal/cli/run.go
View File

@@ -28,7 +28,7 @@ import (
const defaultConfigPath = "/usr/local/etc/notarius/config.yml"
const usage = `Usage:
notarius help
notarius run <pipeline-id> --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 <pipeline-id> --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 ""

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

@@ -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"`
}

1
internal/core/debugbundle/summary.go
View File

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

12
internal/framework/checkpoint/filesystem_test.go
View File

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

18
internal/framework/checkpoint/identity.go
View File

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

10
internal/framework/checkpoint/identity_test.go
View File

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

104
internal/framework/checkpoint/loader.go
View File

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

6
internal/framework/checkpoint/manifest.go
View File

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

108
internal/framework/checkpoint/recorder.go
View File

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

37
internal/framework/checkpoint/recorder_test.go
View File

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

155
internal/framework/pipeline/checkpoint.go
View File

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

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

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

43
internal/framework/pipeline/runner_concurrent.go
View File

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

97
internal/framework/pipeline/runner_typed.go
View File

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

23
internal/framework/pipeline/runner_typed_checkpoint_test.go
View File

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

113
internal/framework/pipeline/synchronized_collaborators.go
View File

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