From 7aadb088a6853b66fda2a484cf7e28858b78dbe5 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Wed, 22 Jul 2026 02:27:19 +0000 Subject: [PATCH] Hydrate accepted producer checkpoints --- docs/internal/pipeline.md | 27 +- docs/internal/state.md | 11 + docs/operations.md | 28 +- .../framework/checkpoint/filesystem_test.go | 101 ++++++ internal/framework/checkpoint/loader.go | 51 +++ internal/framework/pipeline/checkpoint.go | 6 +- .../runner_accepted_checkpoint_test.go | 290 ++++++++++++++++++ .../framework/pipeline/runner_concurrent.go | 48 ++- internal/framework/pipeline/runner_typed.go | 5 +- .../pipeline/runner_typed_checkpoint_test.go | 14 +- .../pipeline/synchronized_collaborators.go | 5 + 11 files changed, 565 insertions(+), 21 deletions(-) create mode 100644 internal/framework/pipeline/runner_accepted_checkpoint_test.go diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index 4804ba5..b6344c7 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -313,12 +313,31 @@ bytes, and decode failures become explicit reuse misses and execute the lane normally. Dependency fingerprints and debug content digests use the same stable codec bytes that cross those boundaries. +That progressive extract, merge, and normalize reuse is the ordinary resume +path. A lane marked as a required predecessor for selective recomputation takes +a separate accepted-output path before extract scheduling. The loader reads the +existing successful normalize manifest and payload by step, lane, and +normalizer, without consulting extract or merge dependencies. It requires the +current non-empty checkpoint identity to match, so the invocation identity +still binds the input, resolved topology and configuration, references, runtime +overrides, profiles, and component fingerprints. + +The runner decodes that accepted normalized artifact with the prepared codec, +re-encodes it, and requires exact kind, schema identity and digest, media type, +canonical bytes, content digest, and producer provenance. A valid result becomes +a runner-owned cloned normalized output, restores only normalize-checkpoint +warnings, and records one `accepted_artifact_reused` normalize decision. It does +not invoke or record extract, merge, normalize, or their validators. Invalid or +unavailable accepted state records its decision and fails the producer step; +the dependent step never starts and the producer is not implicitly rerun. + Generated references add downstream dependencies containing the producer's artifact kind, complete schema identity, media type, canonical content digest, -and size. Compatible producer checkpoints may therefore feed a later step -without re-executing the producer. A missing, rejected, corrupt, incompatible, -or changed producer invalidates every transitive dependent lane while leaving -independent work eligible for reuse. The runner records bounded decision +and size. Compatible accepted producer outputs may therefore feed a later step +without re-executing the producer. Forced lanes bypass accepted-output +hydration and execute normally. A missing, rejected, corrupt, incompatible, or +changed producer blocks its dependent while leaving independent work eligible +for reuse. The runner records bounded decision categories: `reused`, `executed`, `forced_recompute`, and `dependency_invalidated`. diff --git a/docs/internal/state.md b/docs/internal/state.md index 5333c80..4153fb7 100644 --- a/docs/internal/state.md +++ b/docs/internal/state.md @@ -43,6 +43,17 @@ not permit stale downstream reuse. The CLI's selective recomputation policy records `forced_recompute` for the selected step and its dependents while requiring compatible predecessor checkpoints. +Ordinary resume loads extract, merge, and normalize checkpoints progressively +and may execute later lane stages after an earlier cache miss. Selective +recomputation instead asks the loader for the required producer's accepted +normalize artifact. That lookup reuses the existing normalize files, requires +workspace schema v3 plus an exact non-empty invocation identity, and deliberately +does not require extract or merge checkpoint files or dependency fingerprints. +The runner performs canonical codec and producer-provenance validation before +cloning the artifact into normal step output. Success restores only stored +normalize warnings and emits one normalize decision; failure retains the files, +records the decision, and stops without executing the producer or consumer. + The loader assigns a typed category and reason code at each validation site; diagnostic prose is not classified after the fact. The runner then applies forced-execution policy, validates reusable artifact bytes through the prepared diff --git a/docs/operations.md b/docs/operations.md index 1fa02ee..a5c3271 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -151,7 +151,8 @@ recomputation is acceptable. `--resume` loads compatible accepted work only when checkpoint recording is enabled. A normal resumed run may reuse source, extract, merge, and normalize -checkpoints independently. Generated references add a dependency fingerprint +checkpoints independently and may recompute a stage after a cache miss. +Generated references add a dependency fingerprint for the producer's artifact kind, schema identity, media type, canonical content digest, and size. If that fingerprint changes or the producer is missing, dependent checkpoints are invalidated; unrelated work remains eligible @@ -165,14 +166,22 @@ persistent identity of otherwise identical checkpoints. Decisions are bounded and categorized as `reused`, `executed`, `forced_recompute`, or `dependency_invalidated`. -If a required predecessor cannot be reused during selective recomputation, the -run fails before the dependent step starts. The failure manifest retains the -completed upstream outcomes and dependency context but not generated reference -content. For diagnosis, first check the failed step and lane in the manifest, -then inspect checkpoint decision categories and reason codes. A -`dependency_invalidated` decision means the stored producer, codec identity, -schema, or canonical content no longer matches; a missing or rejected producer -requires rerunning it rather than copying an artifact into the checkpoint root. +For an unselected producer required by a recomputed step, Notarius loads the +accepted normalized artifact directly. Valid normalize state is sufficient even +when that producer's extract or merge checkpoint is missing or corrupt. The +normalize manifest must be successful and match workspace schema v3, the exact +current invocation identity, step, lane, and normalizer; its payload digest and +canonical codec representation must also validate. A forced producer bypasses +this lookup and executes. + +If a required predecessor's accepted normalized artifact is missing, rejected, +corrupt, non-canonical, or incompatible, the run fails before the dependent +step starts. It does not fall back to rerunning that predecessor. The failure +manifest retains completed upstream outcomes and dependency context but not +generated reference content. For diagnosis, first check the producer step and +lane in the manifest, then inspect checkpoint decision categories and reason +codes. Rerun the producer explicitly rather than copying an artifact into the +checkpoint root. The decision that caused a required-predecessor failure is retained before the run returns, and the CLI error identifies its step, lane, and reason code. @@ -195,6 +204,7 @@ Checkpoint reason codes are stable diagnostic identifiers: | `artifact_codec_incompatible` | Stored artifact identity is incomplete or incompatible with the codec contract. | | `artifact_not_canonical` | The codec can decode the artifact, but its bytes are not canonical. | | `checkpoint_reused` | The stored checkpoint passed validation and was reused. | +| `accepted_artifact_reused` | A required producer's accepted normalized artifact was canonically validated and hydrated. | | `recompute_step` | Selective recomputation forced execution of this lane. | Decision detail is bounded explanatory text, not a data-recovery channel. It diff --git a/internal/framework/checkpoint/filesystem_test.go b/internal/framework/checkpoint/filesystem_test.go index fef4ae0..82fcff8 100644 --- a/internal/framework/checkpoint/filesystem_test.go +++ b/internal/framework/checkpoint/filesystem_test.go @@ -322,6 +322,107 @@ func TestFilesystemCheckpointDecisionFamiliesAreStableAndSafe(t *testing.T) { } } +func TestFilesystemLoaderReadsAcceptedNormalizeWithoutStageDependencies(t *testing.T) { + fixture := seedAcceptedNormalizeCheckpoint(t) + identityRoot := filepath.Join(fixture.root, mustRelativePath(t, fixture.identity)) + for _, stage := range []string{"extract", "merge"} { + if _, err := os.Stat(filepath.Join(identityRoot, laneManifestPath(stage, "step-1", "lane-a"))); !os.IsNotExist(err) { + t.Fatalf("%s checkpoint stat error = %v, want absent prerequisite", stage, err) + } + } + checkpoint, decision := fixture.loader.AcceptedNormalize("step-1", "lane-a", "normalize-module") + if !decision.Reused || decision.Category != pipeline.CheckpointDecisionReused || decision.ReasonCode != pipeline.CheckpointReasonAcceptedArtifactReused { + t.Fatalf("accepted normalize decision = %#v", decision) + } + if checkpoint.Output.Artifact.Content == nil || string(checkpoint.Output.Artifact.Content) != string(fixture.normalize.Artifact.Content) { + t.Fatalf("accepted normalize output = %#v, want recorded artifact", checkpoint.Output) + } + if len(checkpoint.Warnings) != 1 || checkpoint.Warnings[0].ReasonCode != "normalized" { + t.Fatalf("accepted normalize warnings = %#v", checkpoint.Warnings) + } +} + +func TestFilesystemLoaderRejectsInvalidAcceptedNormalize(t *testing.T) { + tests := []struct { + name string + mutate func(*testing.T, *filesystemCheckpointFixture) + code pipeline.CheckpointReasonCode + }{ + {"missing", func(t *testing.T, fixture *filesystemCheckpointFixture) { + if err := os.Remove(acceptedNormalizeManifest(t, *fixture)); err != nil { + t.Fatal(err) + } + }, pipeline.CheckpointReasonMissing}, + {"rejected status", func(t *testing.T, fixture *filesystemCheckpointFixture) { + editManifest(t, acceptedNormalizeManifest(t, *fixture), func(m map[string]any) { m["status"] = string(StatusSucceededWithRejections) }) + }, pipeline.CheckpointReasonStatusNotReusable}, + {"corrupt payload", func(t *testing.T, fixture *filesystemCheckpointFixture) { + if err := os.WriteFile(acceptedNormalizePayload(t, *fixture), []byte("{"), 0o600); err != nil { + t.Fatal(err) + } + }, pipeline.CheckpointReasonDecodeFailed}, + {"wrong codec identity", func(t *testing.T, fixture *filesystemCheckpointFixture) { + editJSON(t, acceptedNormalizePayload(t, *fixture), func(m map[string]any) { m["output"].(map[string]any)["artifact_kind"] = "" }) + }, pipeline.CheckpointReasonArtifactCodecIncompatible}, + {"wrong content digest", func(t *testing.T, fixture *filesystemCheckpointFixture) { + editJSON(t, acceptedNormalizePayload(t, *fixture), func(m map[string]any) { + m["output"].(map[string]any)["content"].(map[string]any)["content_digest"] = "sha256:wrong" + }) + }, pipeline.CheckpointReasonArtifactDigestMismatch}, + {"unverifiable identity", func(t *testing.T, fixture *filesystemCheckpointFixture) { + loader := fixture.loader.(*FilesystemLoader) + fixture.loader = &FilesystemLoader{root: loader.root} + }, pipeline.CheckpointReasonIdentityMismatch}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + fixture := seedAcceptedNormalizeCheckpoint(t) + test.mutate(t, &fixture) + _, decision := fixture.loader.AcceptedNormalize("step-1", "lane-a", "normalize-module") + if decision.Reused || decision.ReasonCode != test.code { + t.Fatalf("accepted normalize decision = %#v, want %q", decision, test.code) + } + if strings.Contains(decision.Detail, fixture.root) || strings.Contains(decision.Detail, string(fixture.normalize.Artifact.Content)) { + t.Fatalf("accepted normalize decision leaked path or content: %#v", decision) + } + if _, err := os.Stat(acceptedNormalizePayload(t, fixture)); err != nil { + t.Fatalf("accepted normalize payload was removed after rejection: %v", err) + } + }) + } +} + +func seedAcceptedNormalizeCheckpoint(t *testing.T) filesystemCheckpointFixture { + t.Helper() + root := t.TempDir() + identity := testIdentity(t) + artifact := checkpointArtifact("normalize-module", `{"accepted":true}`) + recorder, err := NewFilesystemRecorder(root, identity) + if err != nil { + t.Fatal(err) + } + stepRecorder := recorder.(pipeline.StepCheckpointRecorder) + warnings := []contracts.Warning{{Scope: "normalize", ReasonCode: "normalized", Message: "normalized warning"}} + if err := stepRecorder.NormalizeSucceededForStep("step-1", "lane-a", "normalize-module", []pipeline.CheckpointFingerprint{{Name: "merge", Value: "sha256:unavailable"}}, artifact, warnings); err != nil { + t.Fatal(err) + } + loader, err := NewFilesystemLoader(root, identity) + if err != nil { + t.Fatal(err) + } + return filesystemCheckpointFixture{root: root, identity: identity, loader: loader, normalize: artifact} +} + +func acceptedNormalizeManifest(t *testing.T, fixture filesystemCheckpointFixture) string { + t.Helper() + return filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), laneManifestPath("normalize", "step-1", "lane-a")) +} + +func acceptedNormalizePayload(t *testing.T, fixture filesystemCheckpointFixture) string { + t.Helper() + return filepath.Join(fixture.root, mustRelativePath(t, fixture.identity), lanePayloadPath("normalize", "step-1", "lane-a", "output.json")) +} + type checkpointStage struct { name string manifest func(filesystemCheckpointFixture) string diff --git a/internal/framework/checkpoint/loader.go b/internal/framework/checkpoint/loader.go index c7a52b4..5c9670b 100644 --- a/internal/framework/checkpoint/loader.go +++ b/internal/framework/checkpoint/loader.go @@ -141,6 +141,57 @@ func (l *FilesystemLoader) NormalizeForStep(stepID, laneID, moduleKey string, de return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, reusedDecision() } +func (l *FilesystemLoader) AcceptedNormalize(stepID, laneID, moduleKey string) (pipeline.NormalizeCheckpoint, pipeline.CheckpointDecision) { + var manifest NormalizeLaneManifest + if d := l.readJSON(laneManifestPath("normalize", stepID, laneID), &manifest); !d.Reused { + return pipeline.NormalizeCheckpoint{}, d + } + if d := l.validateAcceptedNormalizeManifest(manifest.StageManifest, stepID, laneID, moduleKey); !d.Reused { + return pipeline.NormalizeCheckpoint{}, d + } + var payload artifactSingleEnvelope + if d := l.readJSON(lanePayloadPath("normalize", stepID, laneID, "output.json"), &payload); !d.Reused { + return pipeline.NormalizeCheckpoint{}, d + } + values, err := artifactCheckpointOutputs([]artifactCheckpointEnvelope{payload.Output}) + if err != nil { + return pipeline.NormalizeCheckpoint{}, artifactDecision(err, "accepted normalize checkpoint artifact is invalid") + } + if len(values) != 1 { + return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactPayloadInvalid, "accepted normalize checkpoint payload is invalid") + } + if !fingerprintsEqual(checkpointToPipelineFingerprints(manifest.OutputDigests), artifactOutputDigests(values)) { + return pipeline.NormalizeCheckpoint{}, decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonArtifactDigestMismatch, "accepted normalize checkpoint digest does not match its payload") + } + return pipeline.NormalizeCheckpoint{Output: values[0], Warnings: cloneWarnings(payload.Warnings)}, decision(pipeline.CheckpointDecisionReused, pipeline.CheckpointReasonAcceptedArtifactReused, "accepted normalized artifact is reusable") +} + +func (l *FilesystemLoader) validateAcceptedNormalizeManifest(manifest StageManifest, stepID, laneID, moduleKey string) pipeline.CheckpointDecision { + if manifest.WorkspaceSchemaVersion != WorkspaceSchemaVersion { + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonWorkspaceSchemaIncompatible, "checkpoint workspace schema is incompatible") + } + identity := strings.TrimSpace(l.identityDigest) + if identity == "" || strings.TrimSpace(manifest.Metadata["checkpoint_identity_digest"]) == "" || manifest.Metadata["checkpoint_identity_digest"] != identity { + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonIdentityMismatch, "checkpoint identity is unavailable or does not match the current invocation") + } + if manifest.Stage != StageNormalize { + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStageMismatch, "checkpoint stage does not match normalize") + } + if strings.TrimSpace(stepID) == "" || manifest.StepID != stepID { + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStepMismatch, "checkpoint step does not match the requested step") + } + if strings.TrimSpace(laneID) == "" || manifest.LaneID != laneID { + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonLaneMismatch, "checkpoint lane does not match the requested lane") + } + if strings.TrimSpace(moduleKey) == "" || manifest.ModuleKey != moduleKey { + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonModuleMismatch, "checkpoint module does not match the requested normalizer") + } + if manifest.Status != StatusSucceeded { + return decision(pipeline.CheckpointDecisionExecuted, pipeline.CheckpointReasonStatusNotReusable, "checkpoint status cannot provide an accepted normalized artifact") + } + return reusedDecision() +} + func artifactCheckpointOutputs(values []artifactCheckpointEnvelope) ([]pipeline.CheckpointArtifact, error) { if len(values) == 0 { return nil, nil diff --git a/internal/framework/pipeline/checkpoint.go b/internal/framework/pipeline/checkpoint.go index 2aa1bb3..6d0fe82 100644 --- a/internal/framework/pipeline/checkpoint.go +++ b/internal/framework/pipeline/checkpoint.go @@ -176,7 +176,7 @@ func forceCheckpointDecision(policy CheckpointExecutionPolicy, stepID, laneID st } func requireReusableCheckpoint(policy CheckpointExecutionPolicy, stepID, laneID string, decision CheckpointDecision) error { - if policy.requiresReusable(stepID, laneID) && !decision.Reused { + if policy.requiresReusable(stepID, laneID) && !policy.forced(stepID, laneID) && !decision.Reused { return fmt.Errorf("required reusable checkpoint unavailable for step %q lane %q (%s)", strings.TrimSpace(stepID), strings.TrimSpace(laneID), decision.ReasonCode) } return nil @@ -241,6 +241,7 @@ type CheckpointLoader interface { Extract(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) Merge(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) Normalize(laneID string, moduleKey string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) + AcceptedNormalize(stepID, laneID, moduleKey string) (NormalizeCheckpoint, CheckpointDecision) } // StepCheckpointLoader is the step-aware counterpart used by the persistent @@ -306,6 +307,9 @@ func (noopCheckpointLoader) Merge(string, string, []CheckpointFingerprint) (Merg func (noopCheckpointLoader) Normalize(string, string, []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { return NormalizeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") } +func (noopCheckpointLoader) AcceptedNormalize(string, string, string) (NormalizeCheckpoint, CheckpointDecision) { + return NormalizeCheckpoint{}, checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonLoadingDisabled, "checkpoint loading disabled") +} func checkpointExtractRunning(recorder CheckpointRecorder, stepID, laneID, moduleKey string, deps []CheckpointFingerprint) error { if stepAware, ok := recorder.(StepCheckpointRecorder); ok { diff --git a/internal/framework/pipeline/runner_accepted_checkpoint_test.go b/internal/framework/pipeline/runner_accepted_checkpoint_test.go new file mode 100644 index 0000000..f30ddc4 --- /dev/null +++ b/internal/framework/pipeline/runner_accepted_checkpoint_test.go @@ -0,0 +1,290 @@ +package pipeline + +import ( + "context" + "encoding/json" + "reflect" + "strings" + "testing" + + "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" +) + +type acceptedCheckpointLoader struct { + CheckpointLoader + accepted map[string]NormalizeCheckpoint + acceptedDecision map[string]CheckpointDecision + acceptedCalls map[string]int + extractDeps [][]CheckpointFingerprint + mergeDeps [][]CheckpointFingerprint + normalizeDeps [][]CheckpointFingerprint +} + +func newAcceptedCheckpointLoader() *acceptedCheckpointLoader { + return &acceptedCheckpointLoader{ + CheckpointLoader: NoopCheckpointLoader(), + accepted: make(map[string]NormalizeCheckpoint), + acceptedDecision: make(map[string]CheckpointDecision), + acceptedCalls: make(map[string]int), + } +} + +func (l *acceptedCheckpointLoader) Enabled() bool { return true } +func (l *acceptedCheckpointLoader) AcceptedNormalize(stepID, laneID, _ string) (NormalizeCheckpoint, CheckpointDecision) { + key := CheckpointLaneKey(stepID, laneID) + l.acceptedCalls[key]++ + return l.accepted[key], l.acceptedDecision[key] +} +func (l *acceptedCheckpointLoader) Extract(_ string, _ string, dependencies []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { + l.extractDeps = append(l.extractDeps, append([]CheckpointFingerprint(nil), dependencies...)) + return ExtractCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") +} +func (l *acceptedCheckpointLoader) Merge(_ string, _ string, dependencies []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) { + l.mergeDeps = append(l.mergeDeps, append([]CheckpointFingerprint(nil), dependencies...)) + return MergeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") +} +func (l *acceptedCheckpointLoader) Normalize(_ string, _ string, dependencies []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) { + l.normalizeDeps = append(l.normalizeDeps, append([]CheckpointFingerprint(nil), dependencies...)) + return NormalizeCheckpoint{}, NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing") +} + +func TestRunnerHydratesRequiredNormalizedArtifact(t *testing.T) { + value := codecNotes{Items: []string{"canonical producer value"}} + input, _, _ := handoffFixture(t, value) + prepared := input.Prepared + producer := &prepared.Steps[0].lanes[0] + consumer := &prepared.Steps[1].lanes[0] + doc := prepared.input.(*typedTestInput).doc + stored, err := checkpointArtifact(producer.typed.codec, producer.resolved.ID, producer.resolved.Normalize.Module, doc.ID, value) + if err != nil { + t.Fatal(err) + } + + operationCalls := 0 + producer.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + operationCalls++ + return erasedTypedResult{Value: value}, nil + } + producer.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + operationCalls++ + return erasedTypedResult{Value: value}, nil + } + producer.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + operationCalls++ + return erasedTypedResult{Value: value}, nil + } + validatorCalls := 0 + validator := preparedValidator{typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + validatorCalls++ + return contracts.ValidationResult{Approved: true}, nil + }} + producer.extractValidators.validators = []preparedValidator{validator} + producer.mergeValidators.validators = []preparedValidator{validator} + producer.normalizeValidators.validators = []preparedValidator{validator} + + var received contracts.ReferenceSet + consumer.typed.extract = func(_ context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + received = CloneReferenceSet(request.References) + return erasedTypedResult{Value: codecScore{Value: 3}}, nil + } + loader := newAcceptedCheckpointLoader() + producerKey := CheckpointLaneKey(producer.resolved.StepID, producer.resolved.ID) + consumerKey := CheckpointLaneKey(consumer.resolved.StepID, consumer.resolved.ID) + loader.accepted[producerKey] = NormalizeCheckpoint{Output: stored, Warnings: []contracts.Warning{{Scope: "normalize", ReasonCode: "stored-warning", Message: "stored normalize warning"}}} + loader.acceptedDecision[producerKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable") + policy := CheckpointExecutionPolicy{ + RequireReusableLanes: map[string]struct{}{producerKey: {}}, + ForcedLanes: map[string]struct{}{consumerKey: {}}, + } + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + if operationCalls != 0 || validatorCalls != 0 { + t.Fatalf("hydrated producer calls = operations %d validators %d, want zero", operationCalls, validatorCalls) + } + if loader.acceptedCalls[producerKey] != 1 || len(loader.extractDeps) != 1 { + t.Fatalf("loader calls = accepted %#v extract %d, want producer hydration and consumer execution only", loader.acceptedCalls, len(loader.extractDeps)) + } + item := received.Slots["producer-output"].Items[0] + if string(item.Content) != string(stored.Artifact.Content) || item.Producer.StepID != producer.resolved.StepID || item.Producer.LaneID != producer.resolved.ID { + t.Fatalf("consumer generated reference = %#v, want exact hydrated producer bytes and identity", item) + } + if len(output.Warnings) != 1 || output.Warnings[0].ReasonCode != "stored-warning" { + t.Fatalf("hydrated warnings = %#v, want normalize checkpoint warnings only", output.Warnings) + } + assertAcceptedNormalizeEvent(t, output.CheckpointEvents, producer.resolved.StepID, producer.resolved.ID, CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused) + for _, event := range output.CheckpointEvents { + if event.StepID == producer.resolved.StepID && event.LaneID == producer.resolved.ID && event.Stage != string(StageNormalize) { + t.Fatalf("hydrated producer synthesized checkpoint event: %#v", event) + } + } + + freshInput, _, _ := handoffFixture(t, value) + freshPrepared := freshInput.Prepared + freshPrepared.Steps[0].lanes[0].typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: value}, nil + } + freshPrepared.Steps[0].lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{Value: value}, nil + } + freshPrepared.Steps[1].lanes[0].typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecScore{Value: 3}}, nil + } + freshLoader := &handoffDependencyLoader{CheckpointLoader: NoopCheckpointLoader()} + freshOutput, err := New().Run(context.Background(), RunInput{Prepared: freshPrepared, RawInput: []byte("input"), Checkpoint: freshLoader}) + if err != nil { + t.Fatalf("fresh Run() error = %v", err) + } + if len(freshLoader.extract) != 2 || len(loader.extractDeps) != 1 || !reflect.DeepEqual(freshLoader.extract[1], loader.extractDeps[0]) { + t.Fatalf("consumer dependencies differ: fresh %#v hydrated %#v", freshLoader.extract, loader.extractDeps) + } + if !reflect.DeepEqual(freshOutput.Manifest.References, output.Manifest.References) { + t.Fatalf("generated provenance differs: fresh %#v hydrated %#v", freshOutput.Manifest.References, output.Manifest.References) + } +} + +func TestRunnerRejectsInvalidRequiredNormalizedArtifactBeforeConsumer(t *testing.T) { + const contentSentinel = "sensitive-campaign-payload-74291" + tests := []struct { + name string + decision CheckpointDecision + mutate func(*CheckpointArtifact) + wantCode CheckpointReasonCode + }{ + {"missing", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonMissing, "checkpoint missing"), nil, CheckpointReasonMissing}, + {"rejected status", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonStatusNotReusable, "status rejected"), nil, CheckpointReasonStatusNotReusable}, + {"corrupt payload", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable"), func(v *CheckpointArtifact) { v.Artifact.Content = []byte(`{"items":[`) }, CheckpointReasonArtifactPayloadInvalid}, + {"non canonical", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable"), func(v *CheckpointArtifact) { v.Artifact.Content = []byte(`{"items": ["stored"]}`) }, CheckpointReasonArtifactNotCanonical}, + {"wrong codec identity", NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable"), func(v *CheckpointArtifact) { v.Artifact.Kind = "test/score" }, CheckpointReasonArtifactCodecIncompatible}, + {"wrong content digest", NewCheckpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactDigestMismatch, "content digest mismatch"), nil, CheckpointReasonArtifactDigestMismatch}, + } + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + input, _, _ := handoffFixture(t, codecNotes{Items: []string{contentSentinel}}) + prepared := input.Prepared + producer := &prepared.Steps[0].lanes[0] + consumer := &prepared.Steps[1].lanes[0] + doc := prepared.input.(*typedTestInput).doc + stored, err := checkpointArtifact(producer.typed.codec, producer.resolved.ID, producer.resolved.Normalize.Module, doc.ID, codecNotes{Items: []string{contentSentinel}}) + if err != nil { + t.Fatal(err) + } + if test.mutate != nil { + test.mutate(&stored) + } + consumerCalls := 0 + consumer.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + consumerCalls++ + return erasedTypedResult{Value: codecScore{Value: 1}}, nil + } + loader := newAcceptedCheckpointLoader() + producerKey := CheckpointLaneKey(producer.resolved.StepID, producer.resolved.ID) + loader.accepted[producerKey] = NormalizeCheckpoint{Output: stored} + loader.acceptedDecision[producerKey] = test.decision + policy := CheckpointExecutionPolicy{RequireReusableLanes: map[string]struct{}{producerKey: {}}} + output, runErr := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}) + if runErr == nil || !strings.Contains(runErr.Error(), string(test.wantCode)) || consumerCalls != 0 { + t.Fatalf("Run() error = %v consumer calls = %d, want %q before consumer", runErr, consumerCalls, test.wantCode) + } + assertAcceptedNormalizeEvent(t, output.CheckpointEvents, producer.resolved.StepID, producer.resolved.ID, CheckpointDecisionExecuted, test.wantCode) + encoded, err := json.Marshal(struct { + Manifest any + Events any + Error string + }{output.Manifest, output.CheckpointEvents, runErr.Error()}) + if err != nil { + t.Fatal(err) + } + if strings.Contains(string(encoded), contentSentinel) || strings.Contains(string(encoded), string(stored.Artifact.Content)) { + t.Fatalf("failed hydration diagnostics leaked artifact content: %s", encoded) + } + }) + } +} + +func TestForcedRequiredLaneExecutesInsteadOfHydrating(t *testing.T) { + prepared := preparedOrderedPipeline(t, 1, + orderedLaneSpec{id: "unrelated", profile: "score"}, + orderedLaneSpec{id: "producer", profile: "notes"}, + orderedLaneSpec{id: "consumer", profile: "score"}, + ) + unrelated := &prepared.Steps[0].lanes[0] + producer := &prepared.Steps[1].lanes[0] + consumer := &prepared.Steps[2].lanes[0] + installGeneratedReferenceTarget(&consumer.resolved.ExtractReferences, StageExtract, consumer, "step-2", "producer") + + doc := prepared.input.(*typedTestInput).doc + unrelatedArtifact, err := checkpointArtifact(unrelated.typed.codec, unrelated.resolved.ID, unrelated.resolved.Normalize.Module, doc.ID, codecScore{Value: 9}) + if err != nil { + t.Fatal(err) + } + producerArtifact, err := checkpointArtifact(producer.typed.codec, producer.resolved.ID, producer.resolved.Normalize.Module, doc.ID, codecNotes{Items: []string{"stale"}}) + if err != nil { + t.Fatal(err) + } + unrelatedCalls, producerCalls := 0, 0 + unrelated.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + unrelatedCalls++ + return erasedTypedResult{Value: codecScore{Value: 9}}, nil + } + producer.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + producerCalls++ + return erasedTypedResult{Value: codecNotes{Items: []string{"fresh"}}}, nil + } + producer.typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecNotes{Items: []string{"fresh"}}}, nil + } + consumer.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecScore{Value: 1}}, nil + } + loader := newAcceptedCheckpointLoader() + unrelatedKey := CheckpointLaneKey(unrelated.resolved.StepID, unrelated.resolved.ID) + producerKey := CheckpointLaneKey(producer.resolved.StepID, producer.resolved.ID) + consumerKey := CheckpointLaneKey(consumer.resolved.StepID, consumer.resolved.ID) + loader.accepted[unrelatedKey] = NormalizeCheckpoint{Output: unrelatedArtifact} + loader.acceptedDecision[unrelatedKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable") + loader.accepted[producerKey] = NormalizeCheckpoint{Output: producerArtifact} + loader.acceptedDecision[producerKey] = NewCheckpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted artifact reusable") + policy := CheckpointExecutionPolicy{ + ForcedLanes: map[string]struct{}{producerKey: {}, consumerKey: {}}, + RequireReusableLanes: map[string]struct{}{unrelatedKey: {}, producerKey: {}}, + } + if _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Checkpoint: loader, CheckpointPolicy: policy}); err != nil { + t.Fatalf("Run() error = %v", err) + } + if producerCalls == 0 || unrelatedCalls != 0 || loader.acceptedCalls[producerKey] != 0 || loader.acceptedCalls[unrelatedKey] != 1 { + t.Fatalf("calls producer=%d unrelated=%d accepted=%#v, want forced producer execution and unrelated hydration", producerCalls, unrelatedCalls, loader.acceptedCalls) + } +} + +func installGeneratedReferenceTarget(target *ResolvedReferenceTarget, stage ModuleStage, consumer *preparedLaneExecutor, producerStep, producerLane string) { + *target = ResolvedReferenceTarget{ + Stage: stage, + StepID: consumer.resolved.StepID, + LaneID: consumer.resolved.ID, + Module: consumer.resolved.Extract.Module, + Bindings: []ReferenceBinding{{ + Stage: stage, + LaneID: consumer.resolved.ID, + SlotName: "producer-output", + Artifact: &ArtifactReference{Step: producerStep, Lane: producerLane}, + }}, + ReferenceSet: contracts.ReferenceSet{Slots: map[string]contracts.ResolvedReferenceSlot{ + "producer-output": {Slot: contracts.ReferenceSlot{Name: "producer-output", AcceptedArtifactKinds: []contracts.ArtifactKind{"test/notes"}, AcceptedMediaTypes: []string{"application/json"}}}, + }}, + } +} + +func assertAcceptedNormalizeEvent(t *testing.T, events []CheckpointEvent, stepID, laneID string, category CheckpointDecisionCategory, code CheckpointReasonCode) { + t.Helper() + for _, event := range events { + if event.Stage == string(StageNormalize) && event.StepID == stepID && event.LaneID == laneID { + if event.Category != category || event.ReasonCode != code { + t.Fatalf("accepted normalize event = %#v, want %q/%q", event, category, code) + } + return + } + } + t.Fatalf("accepted normalize event missing from %#v", events) +} diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index d815923..30c6cc7 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -26,6 +26,8 @@ type laneExtractState struct { results map[int]extractJobResult remaining int failed bool + terminal bool + output RunOutput } type finalizedExtractResults struct { @@ -100,6 +102,14 @@ func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints if err := setTypedLaneManifestMetadata(output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil { return nil, err } + if input.CheckpointPolicy.requiresReusable(input.stepID, prepared.resolved.ID) && !input.CheckpointPolicy.forced(input.stepID, prepared.resolved.ID) { + state, err := hydrateRequiredLane(input, loader, doc, i, prepared, output) + if err != nil { + return nil, err + } + states[i] = state + continue + } state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared, output) if err != nil { return nil, err @@ -147,7 +157,7 @@ func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoin for chunkIndex := range chunks { for laneIndex := range states { state := states[laneIndex] - if state.decision.Reused { + if state.terminal || state.decision.Reused { continue } select { @@ -187,7 +197,9 @@ func collectLaneResults(ctx context.Context, cancel context.CancelFunc, input Ru var pendingContinuations []*laneExtractState launched, completed := 0, 0 for _, state := range states { - if state.decision.Reused { + if state.terminal { + completedOutputs[state.index] = state.output + } else if state.decision.Reused { pendingContinuations = append(pendingContinuations, state) } } @@ -241,6 +253,38 @@ func collectLaneResults(ctx context.Context, cancel context.CancelFunc, input Ru return completedOutputs, runErrors } +func hydrateRequiredLane(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, index int, prepared preparedLaneExecutor, output *RunOutput) (*laneExtractState, error) { + lane, typed := prepared.resolved, prepared.typed + local := RunOutput{Manifest: manifestFromPipeline(input)} + checkpoint, decision := loader.AcceptedNormalize(input.stepID, lane.ID, lane.Normalize.Module) + if decision.Reused { + decision = checkpointDecision(CheckpointDecisionReused, CheckpointReasonAcceptedArtifactReused, "accepted normalized artifact is reusable") + if checkpoint.Output.LaneID != lane.ID || checkpoint.Output.ModuleKey != lane.Normalize.Module || checkpoint.Output.SourceID != doc.ID { + decision = checkpointDecision(CheckpointDecisionExecuted, CheckpointReasonArtifactPayloadInvalid, "accepted normalized artifact provenance does not match the producer lane") + } + } + decision, err := resolveCheckpointDecision(&local, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, decision, typed.codec, []CheckpointArtifact{checkpoint.Output}) + if err != nil { + if mergeErr := mergeLaneOutput(output, local); mergeErr != nil { + return nil, mergeErr + } + return nil, err + } + _, hydrated, err := decodeCanonicalCheckpointArtifact(typed.codec, checkpoint.Output) + if err != nil { + return nil, fmt.Errorf("hydrate accepted normalized artifact for step %q lane %q: %w", input.stepID, lane.ID, err) + } + local.Warnings = append(local.Warnings, cloneWarnings(checkpoint.Warnings)...) + local.NormalizeOutputs = append(local.NormalizeOutputs, contracts.SerializedOutput{ + StepID: input.stepID, + LaneID: lane.ID, + NormalizerKey: lane.Normalize.Module, + SourceID: doc.ID, + Artifact: contracts.CloneSerializedArtifact(hydrated.Artifact), + }) + return &laneExtractState{index: index, prepared: prepared, decision: decision, terminal: true, output: local}, nil +} + func mergeCompletedLanes(output *RunOutput, completedOutputs []RunOutput) error { for i := range completedOutputs { if err := mergeLaneOutput(output, completedOutputs[i]); err != nil { diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 475ad67..ed41725 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -1,6 +1,7 @@ package pipeline import ( + "bytes" "context" "crypto/sha256" "encoding/hex" @@ -125,7 +126,7 @@ func decodeCheckpointArtifact(codec artifactCodecEntry, artifact CheckpointArtif if artifact.Artifact.Kind != codec.spec.Kind { return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "artifact kind %q does not match codec %q", artifact.Artifact.Kind, codec.spec.Kind) } - if artifact.Artifact.Schema.ID != codec.spec.Schema.ID || artifact.Artifact.Schema.Version != codec.spec.Schema.Version { + if artifact.Artifact.Schema.ID != codec.spec.Schema.ID || artifact.Artifact.Schema.Name != codec.spec.Schema.Name || artifact.Artifact.Schema.Version != codec.spec.Schema.Version { return nil, checkpointArtifactValidationFailure(CheckpointReasonArtifactCodecIncompatible, "artifact schema %q version %q does not match codec schema %q version %q", artifact.Artifact.Schema.ID, artifact.Artifact.Schema.Version, codec.spec.Schema.ID, codec.spec.Schema.Version) } if artifact.SchemaDigest != expectedDigest { @@ -150,7 +151,7 @@ func decodeCanonicalCheckpointArtifact(codec artifactCodecEntry, artifact Checkp if err != nil { return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactPayloadInvalid, "encode canonical artifact: %w", err) } - if canonical.Kind != artifact.Artifact.Kind || canonical.MediaType != artifact.Artifact.MediaType || checkpointContentDigest(canonical.Content) != checkpointContentDigest(artifact.Artifact.Content) { + if canonical.Kind != artifact.Artifact.Kind || canonical.Schema.ID != artifact.Artifact.Schema.ID || canonical.Schema.Name != artifact.Artifact.Schema.Name || canonical.Schema.Version != artifact.Artifact.Schema.Version || canonical.MediaType != artifact.Artifact.MediaType || !bytes.Equal(canonical.Content, artifact.Artifact.Content) || checkpointContentDigest(canonical.Content) != checkpointContentDigest(artifact.Artifact.Content) { return nil, CheckpointArtifact{}, checkpointArtifactValidationFailure(CheckpointReasonArtifactNotCanonical, "stored artifact is not canonical") } hydrated, err := hydrateCheckpointArtifact(codec, cloneCheckpointArtifact(artifact), value) diff --git a/internal/framework/pipeline/runner_typed_checkpoint_test.go b/internal/framework/pipeline/runner_typed_checkpoint_test.go index e6dbdaa..5bd6cd4 100644 --- a/internal/framework/pipeline/runner_typed_checkpoint_test.go +++ b/internal/framework/pipeline/runner_typed_checkpoint_test.go @@ -19,6 +19,13 @@ func (l requiredCheckpointLoader) Enabled() bool { return true } func (l requiredCheckpointLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { return l.checkpoint, l.decision } +func (l requiredCheckpointLoader) AcceptedNormalize(string, string, string) (NormalizeCheckpoint, CheckpointDecision) { + output := CheckpointArtifact{} + if len(l.checkpoint.Outputs) > 0 { + output = l.checkpoint.Outputs[0] + } + return NormalizeCheckpoint{Output: output}, l.decision +} func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { const unsafeDetail = "unsafe-loader-detail-/private/checkpoint/path" @@ -39,9 +46,10 @@ func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { prepared := preparedConcurrentPipeline(t, 1) step := prepared.Steps[0] lane := step.lanes[0] + doc := prepared.input.(*typedTestInput).doc checkpoint := ExtractCheckpoint{} if test.corrupt { - stored, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Extract.Module, "source", codecNotes{Items: []string{"stored"}}) + stored, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Normalize.Module, doc.ID, codecNotes{Items: []string{"stored"}}) if err != nil { t.Fatal(err) } @@ -59,7 +67,7 @@ func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { } var found bool for _, event := range output.CheckpointEvents { - if event.Stage == string(StageExtract) && event.StepID == step.ID && event.LaneID == lane.resolved.ID { + if event.Stage == string(StageNormalize) && event.StepID == step.ID && event.LaneID == lane.resolved.ID { found = true if event.Action != test.wantAction || event.ReasonCode != test.wantCode { t.Fatalf("checkpoint event = %#v, want action %q and reason %q", event, test.wantAction, test.wantCode) @@ -74,7 +82,7 @@ func TestRequiredCheckpointFailureRetainsDecision(t *testing.T) { } var manifestFound bool for _, decision := range output.Manifest.CheckpointDecisions { - if decision.Stage == string(StageExtract) && decision.StepID == step.ID && decision.LaneID == lane.resolved.ID { + if decision.Stage == string(StageNormalize) && decision.StepID == step.ID && decision.LaneID == lane.resolved.ID { manifestFound = decision.Category == string(test.wantAction) && decision.ReasonCode == string(test.wantCode) } } diff --git a/internal/framework/pipeline/synchronized_collaborators.go b/internal/framework/pipeline/synchronized_collaborators.go index 1b92f46..23121b5 100644 --- a/internal/framework/pipeline/synchronized_collaborators.go +++ b/internal/framework/pipeline/synchronized_collaborators.go @@ -76,6 +76,11 @@ func (l *lockedCheckpointLoader) Normalize(lane, key string, deps []CheckpointFi defer l.mu.Unlock() return l.inner.Normalize(lane, key, deps) } +func (l *lockedCheckpointLoader) AcceptedNormalize(step, lane, key string) (NormalizeCheckpoint, CheckpointDecision) { + l.mu.Lock() + defer l.mu.Unlock() + return l.inner.AcceptedNormalize(step, lane, key) +} func (l *lockedCheckpointLoader) ExtractForStep(step, lane, key string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) { l.mu.Lock() defer l.mu.Unlock()