diff --git a/docs/internal/pipeline.md b/docs/internal/pipeline.md index fd99e07..b6b5d18 100644 --- a/docs/internal/pipeline.md +++ b/docs/internal/pipeline.md @@ -31,13 +31,19 @@ calls `pipeline.ResolvePipeline`. 7. validates each selected module and validator option set through its registry entry; and 8. calculates a digest over the resolved structure, including typed artifact - kind and schema identity. + kind and schema identity and the effective validator policy in its resolved + execution order. Resolution returns a `ResolvedPipeline` containing ordered lanes, concrete bindings, validator chains, reference targets, and the digest. It does not read reference bytes or construct runtime modules. CLI lane and reference selector syntax is defined in the [CLI reference](../cli.md#run). +The digest includes each resolved validator chain's stage, lane, owning module, +ordered validator bindings, execution classes, targets, and artifact kinds. +Changing a default chain or an explicit override therefore changes pipeline +identity whenever it changes the effective validator policy. + ## Reference Materialization The CLI calls `MaterializeReferences` after resolution and before constructing diff --git a/docs/operations.md b/docs/operations.md index 3c2130d..e5340f1 100644 --- a/docs/operations.md +++ b/docs/operations.md @@ -89,7 +89,10 @@ Treat checkpoint directories as sensitive local state. A checkpoint is reused only when its stored status, dependencies, payloads, and digests match the current invocation. Changes to input bytes, the resolved pipeline, selected lanes, the runtime LLM profile override, or bound reference -content invalidate reuse. +content invalidate reuse. The resolved pipeline identity includes effective +default and explicitly overridden validator chains, so adding, removing, +reordering, or reconfiguring a validator invalidates checkpoints even when the +pipeline profile itself is unchanged. Typed artifact checkpoints additionally record codec-owned bytes, artifact kind, schema ID and version, exact schema digest, and media type. A missing or diff --git a/internal/framework/pipeline/profile.go b/internal/framework/pipeline/profile.go index b431447..7204599 100644 --- a/internal/framework/pipeline/profile.go +++ b/internal/framework/pipeline/profile.go @@ -1010,6 +1010,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) { Chunk ModuleBinding ChunkReferences ResolvedReferenceTarget ArtifactLanes []ResolvedArtifactLane + ValidatorChains []ResolvedValidatorChain Output ModuleBinding }{ ID: resolved.ID, @@ -1017,6 +1018,7 @@ func resolvedPipelineDigest(resolved ResolvedPipeline) (string, error) { Chunk: resolved.Chunk, ChunkReferences: resolved.ChunkReferences, ArtifactLanes: resolved.ArtifactLanes, + ValidatorChains: resolved.ValidatorChains, Output: resolved.Output, } encoded, err := json.Marshal(withoutDigest) diff --git a/internal/framework/pipeline/profile_test.go b/internal/framework/pipeline/profile_test.go index 7fc0572..82b5532 100644 --- a/internal/framework/pipeline/profile_test.go +++ b/internal/framework/pipeline/profile_test.go @@ -1062,6 +1062,186 @@ func TestResolvePipelineDigestChangesWhenBindingChanges(t *testing.T) { } } +func TestResolvePipelineDigestIncludesEffectiveValidatorChain(t *testing.T) { + resolve := func(t *testing.T, validators []ModuleBinding, explicitEmpty bool) ResolvedPipeline { + t.Helper() + catalog := newProfileCatalog(t) + registerProfileValidatorSpec(t, catalog, ValidatorSpec{Key: "second-validator", ExecutionClass: contracts.ExecutionClassDeterministic}) + if len(validators) > 0 { + if err := catalog.ValidatorChains.Register(ValidatorChainMapping{ + Stage: StageExtract, + Module: "event-extractor", + Validators: validators, + }); err != nil { + t.Fatalf("register validator chain: %v", err) + } + } + + profile := baselineProfile() + if explicitEmpty { + lane := profile.Artifacts["events"] + lane.Extract.Validators = ValidatorOverride{Set: true} + profile.Artifacts["events"] = lane + } + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + return resolved + } + + inherited := resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, false) + repeated := resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, false) + if inherited.Digest != repeated.Digest { + t.Fatalf("same resolved validator policy produced digests %q and %q", inherited.Digest, repeated.Digest) + } + + tests := []struct { + name string + left ResolvedPipeline + right ResolvedPipeline + }{ + { + name: "different default", + left: resolve(t, []ModuleBinding{Binding("grounded")}, false), + right: resolve(t, []ModuleBinding{Binding("second-validator")}, false), + }, + { + name: "added validator", + left: resolve(t, nil, false), + right: resolve(t, []ModuleBinding{Binding("grounded")}, false), + }, + { + name: "removed validator", + left: resolve(t, []ModuleBinding{Binding("grounded")}, false), + right: resolve(t, nil, false), + }, + { + name: "reordered chain", + left: inherited, + right: resolve(t, []ModuleBinding{Binding("second-validator"), Binding("grounded")}, false), + }, + { + name: "explicit empty", + left: inherited, + right: resolve(t, []ModuleBinding{Binding("grounded"), Binding("second-validator")}, true), + }, + } + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + if tc.left.Digest == tc.right.Digest { + t.Fatalf("digest = %q for both validator policies, want changed digest", tc.left.Digest) + } + }) + } +} + +func TestResolvedPipelineDigestIncludesCompleteValidatorPolicy(t *testing.T) { + base := validatorDigestFixture() + baseDigest, err := resolvedPipelineDigest(base) + if err != nil { + t.Fatalf("resolvedPipelineDigest(base) error = %v, want nil", err) + } + + tests := []struct { + name string + mutate func(*ResolvedPipeline) + }{ + {name: "stage", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Stage = StageMerge }}, + {name: "lane", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].LaneID = "other" }}, + {name: "owner module", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].ModuleKey = "other-extractor" }}, + {name: "validator order", mutate: func(value *ResolvedPipeline) { + value.ValidatorChains[0].Validators[0], value.ValidatorChains[0].Validators[1] = value.ValidatorChains[0].Validators[1], value.ValidatorChains[0].Validators[0] + }}, + {name: "validator module", mutate: func(value *ResolvedPipeline) { + value.ValidatorChains[0].Validators[0].Binding.Module = "other-validator" + }}, + {name: "llm profile", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.LLMProfile = "fast" }}, + {name: "retries", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].Binding.Retries++ }}, + {name: "options", mutate: func(value *ResolvedPipeline) { + value.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.75 + }}, + {name: "references", mutate: func(value *ResolvedPipeline) { + value.ValidatorChains[0].Validators[0].Binding.References["rules"] = "other.md" + }}, + {name: "execution class", mutate: func(value *ResolvedPipeline) { + value.ValidatorChains[0].Validators[0].ExecutionClass = contracts.ExecutionClassDeterministic + }}, + {name: "target", mutate: func(value *ResolvedPipeline) { + value.ValidatorChains[0].Validators[0].Target = ValidatorTargetSerialized + }}, + {name: "artifact kind", mutate: func(value *ResolvedPipeline) { value.ValidatorChains[0].Validators[0].ArtifactKind = "test/other" }}, + } + + for _, tc := range tests { + t.Run(tc.name, func(t *testing.T) { + changed := validatorDigestFixture() + tc.mutate(&changed) + digest, err := resolvedPipelineDigest(changed) + if err != nil { + t.Fatalf("resolvedPipelineDigest(changed) error = %v, want nil", err) + } + if digest == baseDigest { + t.Fatalf("digest = %q after %s change, want different digest", digest, tc.name) + } + }) + } +} + +func TestResolvedPipelineDigestCanonicalizesValidatorBindingMaps(t *testing.T) { + left := validatorDigestFixture() + right := validatorDigestFixture() + right.ValidatorChains[0].Validators[0].Binding.Options = map[string]any{} + right.ValidatorChains[0].Validators[0].Binding.Options["threshold"] = 0.5 + right.ValidatorChains[0].Validators[0].Binding.Options["mode"] = "strict" + right.ValidatorChains[0].Validators[0].Binding.References = map[string]string{} + right.ValidatorChains[0].Validators[0].Binding.References["examples"] = "examples.md" + right.ValidatorChains[0].Validators[0].Binding.References["rules"] = "rules.md" + + leftDigest, err := resolvedPipelineDigest(left) + if err != nil { + t.Fatalf("resolvedPipelineDigest(left) error = %v, want nil", err) + } + rightDigest, err := resolvedPipelineDigest(right) + if err != nil { + t.Fatalf("resolvedPipelineDigest(right) error = %v, want nil", err) + } + if leftDigest != rightDigest { + t.Fatalf("equivalent validator maps produced digests %q and %q", leftDigest, rightDigest) + } +} + +func validatorDigestFixture() ResolvedPipeline { + return ResolvedPipeline{ + ID: "validator-policy", + ValidatorChains: []ResolvedValidatorChain{{ + Stage: StageExtract, + LaneID: "events", + ModuleKey: "event-extractor", + Validators: []ResolvedValidator{ + { + Binding: ModuleBinding{ + Module: "semantic-validator", + LLMProfile: "careful", + Retries: 2, + Options: map[string]any{"mode": "strict", "threshold": 0.5}, + References: map[string]string{"rules": "rules.md", "examples": "examples.md"}, + }, + ExecutionClass: contracts.ExecutionClassLLMBacked, + Target: ValidatorTargetTyped, + ArtifactKind: "test/notes", + }, + { + Binding: Binding("grounded"), + ExecutionClass: contracts.ExecutionClassDeterministic, + Target: ValidatorTargetTyped, + ArtifactKind: "test/notes", + }, + }, + }}, + } +} + func TestBindingTrimsModuleAndLeavesResolutionFieldsEmpty(t *testing.T) { binding := Binding(" module ") if binding.Module != "module" {