From afb7ed3cf1229239464dd9653d591f6bc1de863c Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 21 Jul 2026 21:19:00 +0000 Subject: [PATCH] Execute ordered pipeline steps with barriers --- internal/framework/pipeline/debug.go | 1 + .../framework/pipeline/preparation_test.go | 67 +++++- internal/framework/pipeline/prepare.go | 67 ++++-- .../pipeline/prepared_fingerprints.go | 20 +- internal/framework/pipeline/runner.go | 22 +- .../pipeline/runner_attempt_debug_test.go | 28 +-- .../runner_candidate_encoding_test.go | 8 +- .../pipeline/runner_chunk_plan_test.go | 4 +- .../pipeline/runner_concurrency_test.go | 198 ++++++++++++++++-- .../framework/pipeline/runner_concurrent.go | 14 +- .../pipeline/runner_extract_handoff_test.go | 4 +- .../pipeline/runner_terminal_debug_test.go | 4 +- internal/framework/pipeline/runner_typed.go | 20 +- .../framework/pipeline/typed_execution.go | 1 + .../pipeline/typed_resolution_test.go | 4 +- 15 files changed, 365 insertions(+), 97 deletions(-) diff --git a/internal/framework/pipeline/debug.go b/internal/framework/pipeline/debug.go index a2e913f..eb9bfb9 100644 --- a/internal/framework/pipeline/debug.go +++ b/internal/framework/pipeline/debug.go @@ -63,6 +63,7 @@ func debugPathComponent(value string) string { type debugTimedEnvelope struct { Stage string `json:"stage,omitempty"` + StepID string `json:"step_id,omitempty"` LaneID string `json:"lane_id,omitempty"` ModuleKey string `json:"module_key,omitempty"` Attempt int `json:"attempt,omitempty"` diff --git a/internal/framework/pipeline/preparation_test.go b/internal/framework/pipeline/preparation_test.go index 074e8a4..11cad59 100644 --- a/internal/framework/pipeline/preparation_test.go +++ b/internal/framework/pipeline/preparation_test.go @@ -67,11 +67,71 @@ func TestPrepareConstructsEverythingInStableOrder(t *testing.T) { if !reflect.DeepEqual(built, want) { t.Fatalf("construction order = %#v, want %#v", built, want) } - if prepared.Input.Module != "input" || prepared.Chunk.Module != "chunk" || prepared.Output.Module != "output" || len(prepared.ArtifactLanes) != 1 { + if prepared.Input.Module != "input" || prepared.Chunk.Module != "chunk" || prepared.Output.Module != "output" || len(prepared.Steps) != 1 || len(prepared.Steps[0].ArtifactLanes) != 1 { t.Fatalf("PreparedPipeline = %#v, want explicit resolved components", prepared) } } +func TestPrepareBuildsEveryOrderedStepBeforeExecution(t *testing.T) { + var built []string + registries, input := constructionRegistries(t, &built, nil) + profile := constructionProfile() + lane := profile.Artifacts["artifact"] + profile.Artifacts = nil + profile.Steps = []PipelineStepProfile{ + {ID: "first", Artifacts: map[string]ArtifactLaneProfile{"first-artifact": lane}}, + {ID: "second", Artifacts: map[string]ArtifactLaneProfile{"second-artifact": lane}}, + } + resolved, err := ResolvePipeline(profile, ResolveOptions{}, registries.catalog()) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + prepared, err := Prepare(resolved, registries, ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + want := []string{ + "input", "chunk", "validator", + "extract", "validator", "merge", "validator", "normalize", "validator", + "extract", "validator", "merge", "validator", "normalize", "validator", + "output", + } + if !reflect.DeepEqual(built, want) { + t.Fatalf("construction order = %#v, want %#v", built, want) + } + if len(prepared.Steps) != 2 || len(prepared.Steps[0].lanes) != 1 || len(prepared.Steps[1].lanes) != 1 { + t.Fatalf("prepared steps = %#v, want two constructed steps", prepared.Steps) + } + if len(input.requests) != 0 { + t.Fatalf("input Parse calls = %d, want zero during preparation", len(input.requests)) + } +} + +func TestPrepareRetainsGeneratedSelectorsWithoutReferenceBytes(t *testing.T) { + var built []string + registries, _ := constructionRegistries(t, &built, nil) + resolved, err := ResolvePipeline(constructionProfile(), ResolveOptions{}, registries.catalog()) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v, want nil", err) + } + resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings = []ReferenceBinding{{ + Stage: StageExtract, + LaneID: "artifact", + SlotName: "generated", + Artifact: &ArtifactReference{Step: "producer", Lane: "artifact"}, + }} + prepared, err := Prepare(resolved, registries, ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v, want nil", err) + } + if len(prepared.Steps[0].ArtifactLanes[0].Resolved.ExtractReferences.ReferenceSet.Slots) != 0 { + t.Fatalf("prepared generated reference set = %#v, want no reference bytes", prepared.Steps[0].ArtifactLanes[0].Resolved.ExtractReferences.ReferenceSet) + } + if len(built) == 0 { + t.Fatal("constructed modules = empty, want preparation to construct the selected components") + } +} + func TestPreparedCheckpointFingerprintsCollectEveryComponentInStableScopeOrder(t *testing.T) { provider := func(value string) checkpointFingerprintTestProvider { return checkpointFingerprintTestProvider{{Name: "identity", Value: value}} @@ -90,6 +150,7 @@ func TestPreparedCheckpointFingerprintsCollectEveryComponentInStableScopeOrder(t chunk: provider("chunk-validator"), }}}, output: provider("output"), + Steps: []PreparedPipelineStep{{ID: "default"}}, } lane := preparedLaneExecutor{ resolved: ResolvedArtifactLane{ @@ -107,7 +168,7 @@ func TestPreparedCheckpointFingerprintsCollectEveryComponentInStableScopeOrder(t mergeValidators: fingerprintTestValidatorChain("merge-validator", provider("merge-validator")), normalizeValidators: fingerprintTestValidatorChain("normalize-validator", provider("normalize-validator")), } - prepared.lanes = []preparedLaneExecutor{lane} + prepared.Steps[0].lanes = []preparedLaneExecutor{lane} first, err := collectPreparedCheckpointFingerprints(prepared) if err != nil { @@ -279,7 +340,7 @@ func TestPrepareDeliversTargetReferencesAsIndependentBuildInputs(t *testing.T) { t.Fatalf("resolved extract references = %q, want original content", got) } - _, err = prepared.lanes[0].typed.extract(context.Background(), prepared.lanes[0].typed.extractor, contracts.TypedExtractionRequest{ + _, err = prepared.Steps[0].lanes[0].typed.extract(context.Background(), prepared.Steps[0].lanes[0].typed.extractor, contracts.TypedExtractionRequest{ References: CloneReferenceSet(resolved.Steps[0].ArtifactLanes[0].ExtractReferences.ReferenceSet), }) if err != nil { diff --git a/internal/framework/pipeline/prepare.go b/internal/framework/pipeline/prepare.go index 7659e78..9e02c3d 100644 --- a/internal/framework/pipeline/prepare.go +++ b/internal/framework/pipeline/prepare.go @@ -12,21 +12,26 @@ import ( // resolved pipeline. Its implementation values are private so execution cannot // replace or reconfigure them after preparation. type PreparedPipeline struct { - Input ModuleBinding - Chunk ModuleBinding - ArtifactLanes []PreparedArtifactLane - Output ModuleBinding + Input ModuleBinding + Chunk ModuleBinding + Steps []PreparedPipelineStep + Output ModuleBinding resolved ResolvedPipeline dependencies ModuleDependencies input contracts.InputAdapter chunker contracts.Chunker chunkValidators preparedValidatorChain - lanes []preparedLaneExecutor output contracts.OutputEncoder checkpointFingerprints []CheckpointFingerprint } +type PreparedPipelineStep struct { + ID string + ArtifactLanes []PreparedArtifactLane + lanes []preparedLaneExecutor +} + type PreparedArtifactLane struct { Resolved ResolvedArtifactLane } @@ -68,9 +73,6 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend if err := validateResolvedPipeline(resolved); err != nil { return nil, err } - if len(resolved.Steps) > 1 { - return nil, fmt.Errorf("pipeline %q contains multiple ordered steps; execution support is not available yet", resolved.ID) - } if err := validateRegistrySet(resolved, registries); err != nil { return nil, err } @@ -102,16 +104,22 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend return nil, err } - lanes := stable.AllArtifactLanes() - prepared.ArtifactLanes = make([]PreparedArtifactLane, 0, len(lanes)) - prepared.lanes = make([]preparedLaneExecutor, 0, len(lanes)) - for _, lane := range lanes { - executor, err := prepareLane(stable, lane, registries, deps) - if err != nil { - return nil, err + prepared.Steps = make([]PreparedPipelineStep, len(stable.Steps)) + for stepIndex, step := range stable.Steps { + preparedStep := PreparedPipelineStep{ + ID: step.ID, + ArtifactLanes: make([]PreparedArtifactLane, 0, len(step.ArtifactLanes)), + lanes: make([]preparedLaneExecutor, 0, len(step.ArtifactLanes)), } - prepared.ArtifactLanes = append(prepared.ArtifactLanes, PreparedArtifactLane{Resolved: cloneResolvedArtifactLane(lane)}) - prepared.lanes = append(prepared.lanes, executor) + for _, lane := range step.ArtifactLanes { + executor, err := prepareLane(stable, lane, registries, deps) + if err != nil { + return nil, err + } + preparedStep.ArtifactLanes = append(preparedStep.ArtifactLanes, PreparedArtifactLane{Resolved: cloneResolvedArtifactLane(lane)}) + preparedStep.lanes = append(preparedStep.lanes, executor) + } + prepared.Steps[stepIndex] = preparedStep } output, err := registries.Outputs.BuildWithRequest(stable.Output.Module, request(stable.Output, contracts.ReferenceSet{})) @@ -126,6 +134,31 @@ func Prepare(resolved ResolvedPipeline, registries Registries, deps ModuleDepend return prepared, nil } +func rejectGeneratedBindings(resolved ResolvedPipeline) error { + check := func(stepID, laneID string, target ResolvedReferenceTarget) error { + for _, binding := range target.Bindings { + if binding.Artifact == nil { + continue + } + return fmt.Errorf("pipeline %q step %q lane %q %s reference slot %q uses a generated artifact; generated reference execution is not supported yet", resolved.ID, stepID, laneID, target.Stage, binding.SlotName) + } + return nil + } + if err := check("", "", resolved.ChunkReferences); err != nil { + return err + } + for _, step := range resolved.Steps { + for _, lane := range step.ArtifactLanes { + for _, target := range []ResolvedReferenceTarget{lane.ExtractReferences, lane.MergeReferences, lane.NormalizeReferences} { + if err := check(step.ID, lane.ID, target); err != nil { + return err + } + } + } + } + return nil +} + func prepareLane(pipeline ResolvedPipeline, lane ResolvedArtifactLane, registries Registries, deps ModuleDependencies) (preparedLaneExecutor, error) { executor := preparedLaneExecutor{resolved: cloneResolvedArtifactLane(lane)} request := func(binding ModuleBinding, references contracts.ReferenceSet) BuildRequest { diff --git a/internal/framework/pipeline/prepared_fingerprints.go b/internal/framework/pipeline/prepared_fingerprints.go index 238abea..d63daf3 100644 --- a/internal/framework/pipeline/prepared_fingerprints.go +++ b/internal/framework/pipeline/prepared_fingerprints.go @@ -26,16 +26,18 @@ func collectPreparedCheckpointFingerprints(prepared *PreparedPipeline) ([]Checkp {scope: "chunk:" + prepared.resolved.Chunk.Module, module: prepared.chunker}, } components = appendValidatorFingerprintComponents(components, "chunk:"+prepared.resolved.Chunk.Module, prepared.chunkValidators) - for _, lane := range prepared.lanes { - laneScope := func(stage ModuleStage, moduleKey string) string { - return string(stage) + ":" + lane.resolved.ID + ":" + moduleKey + for _, step := range prepared.Steps { + for _, lane := range step.lanes { + laneScope := func(stage ModuleStage, moduleKey string) string { + return string(stage) + ":" + lane.resolved.ID + ":" + moduleKey + } + components = append(components, checkpointFingerprintComponent{scope: laneScope(StageExtract, lane.resolved.Extract.Module), module: lane.typed.extractor}) + components = appendValidatorFingerprintComponents(components, laneScope(StageExtract, lane.resolved.Extract.Module), lane.extractValidators) + components = append(components, checkpointFingerprintComponent{scope: laneScope(StageMerge, lane.resolved.Merge.Module), module: lane.typed.merger}) + components = appendValidatorFingerprintComponents(components, laneScope(StageMerge, lane.resolved.Merge.Module), lane.mergeValidators) + components = append(components, checkpointFingerprintComponent{scope: laneScope(StageNormalize, lane.resolved.Normalize.Module), module: lane.typed.normalizer}) + components = appendValidatorFingerprintComponents(components, laneScope(StageNormalize, lane.resolved.Normalize.Module), lane.normalizeValidators) } - components = append(components, checkpointFingerprintComponent{scope: laneScope(StageExtract, lane.resolved.Extract.Module), module: lane.typed.extractor}) - components = appendValidatorFingerprintComponents(components, laneScope(StageExtract, lane.resolved.Extract.Module), lane.extractValidators) - components = append(components, checkpointFingerprintComponent{scope: laneScope(StageMerge, lane.resolved.Merge.Module), module: lane.typed.merger}) - components = appendValidatorFingerprintComponents(components, laneScope(StageMerge, lane.resolved.Merge.Module), lane.mergeValidators) - components = append(components, checkpointFingerprintComponent{scope: laneScope(StageNormalize, lane.resolved.Normalize.Module), module: lane.typed.normalizer}) - components = appendValidatorFingerprintComponents(components, laneScope(StageNormalize, lane.resolved.Normalize.Module), lane.normalizeValidators) } components = append(components, checkpointFingerprintComponent{scope: "output:" + prepared.resolved.Output.Module, module: prepared.output}) diff --git a/internal/framework/pipeline/runner.go b/internal/framework/pipeline/runner.go index 7c14587..93aebc4 100644 --- a/internal/framework/pipeline/runner.go +++ b/internal/framework/pipeline/runner.go @@ -59,6 +59,7 @@ type RunInput struct { pipeline ResolvedPipeline llmClient contracts.StructuredLLMClient + stepID string } type RunOutput struct { @@ -249,12 +250,16 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err } if chunkResult.accepted { - laneOutput, laneErr := r.runLanes(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks) - if err := mergeLaneOutput(&output, laneOutput); err != nil { - return failOutput(output), err - } - if laneErr != nil { - return failOutput(output), laneErr + for _, step := range input.Prepared.Steps { + stepInput := input + stepInput.stepID = step.ID + laneOutput, laneErr := r.runLanes(ctx, stepInput, step, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks) + if err := mergeLaneOutput(&output, laneOutput); err != nil { + return failOutput(output), err + } + if laneErr != nil { + return failOutput(output), fmt.Errorf("execute pipeline step %q: %w", step.ID, laneErr) + } } } @@ -460,7 +465,10 @@ func validateRunInput(input RunInput) error { if mode != ChunkCacheBypass && input.ChunkPlans == nil { return fmt.Errorf("chunk plan store is required for %q mode", mode) } - return validateResolvedPipeline(input.Prepared.resolved) + if err := validateResolvedPipeline(input.Prepared.resolved); err != nil { + return err + } + return rejectGeneratedBindings(input.Prepared.resolved) } func validateResolvedPipeline(pipeline ResolvedPipeline) error { diff --git a/internal/framework/pipeline/runner_attempt_debug_test.go b/internal/framework/pipeline/runner_attempt_debug_test.go index 33afe8c..c70e44e 100644 --- a/internal/framework/pipeline/runner_attempt_debug_test.go +++ b/internal/framework/pipeline/runner_attempt_debug_test.go @@ -109,11 +109,11 @@ func (attemptDebugLLM) CompleteStructured(_ context.Context, request contracts.S func preparedAttemptDebugPipeline(t *testing.T) *PreparedPipeline { t.Helper() prepared := preparedConcurrentPipeline(t, 1) - prepared.lanes = prepared.lanes[:1] + prepared.Steps[0].lanes = prepared.Steps[0].lanes[:1] prepared.resolved.Steps[0].ArtifactLanes = prepared.resolved.Steps[0].ArtifactLanes[:1] - prepared.ArtifactLanes = prepared.ArtifactLanes[:1] - prepared.lanes[0].mergeValidators = preparedValidatorChain{} - prepared.lanes[0].normalizeValidators = preparedValidatorChain{} + prepared.Steps[0].ArtifactLanes = prepared.Steps[0].ArtifactLanes[:1] + prepared.Steps[0].lanes[0].mergeValidators = preparedValidatorChain{} + prepared.Steps[0].lanes[0].normalizeValidators = preparedValidatorChain{} return prepared } @@ -126,13 +126,13 @@ func TestRunnerWritesAttemptScopedMergeAndNormalizeDebug(t *testing.T) { prepared := preparedAttemptDebugPipeline(t) debug := newCapturedDebugRecorder() client := WithDebugLLMRecording(attemptDebugLLM{}, debug) - prepared.lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + prepared.Steps[0].lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { if err := callAttemptDebugLLM(ctx, client, "merge"); err != nil { return erasedTypedResult{}, err } return erasedTypedResult{Value: codecNotes{Items: []string{"merged"}}, Warnings: []contracts.Warning{{Scope: "merge", ReasonCode: "observed", Message: "merge warning"}}}, nil } - prepared.lanes[0].typed.normalize = func(ctx context.Context, _ any, _ contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + prepared.Steps[0].lanes[0].typed.normalize = func(ctx context.Context, _ any, _ contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { if err := callAttemptDebugLLM(ctx, client, "normalize"); err != nil { return erasedTypedResult{}, err } @@ -187,7 +187,7 @@ func TestRunnerRecordsDistinctRetryAttemptsAndPromotesAcceptedWarningsOnly(t *te prepared := preparedAttemptDebugPipeline(t) debug := newCapturedDebugRecorder() client := WithDebugLLMRecording(attemptDebugLLM{}, debug) - lane := &prepared.lanes[0] + lane := &prepared.Steps[0].lanes[0] attempts := 0 operation := func(ctx context.Context) (erasedTypedResult, error) { attempts++ @@ -257,7 +257,7 @@ func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) { { name: "merge module error", configure: func(prepared *PreparedPipeline) { - prepared.lanes[0].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + prepared.Steps[0].lanes[0].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { return erasedTypedResult{}, errors.New("merge exploded") } }, @@ -267,7 +267,7 @@ func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) { { name: "normalize validator error", configure: func(prepared *PreparedPipeline) { - prepared.lanes[0].normalizeValidators.validators = []preparedValidator{{ + prepared.Steps[0].lanes[0].normalizeValidators.validators = []preparedValidator{{ resolved: ResolvedValidator{Binding: Binding("error-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { return contracts.ValidationResult{}, errors.New("validator exploded") @@ -281,7 +281,7 @@ func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) { { name: "merge final rejection", configure: func(prepared *PreparedPipeline) { - prepared.lanes[0].mergeValidators.validators = []preparedValidator{{ + prepared.Steps[0].lanes[0].mergeValidators.validators = []preparedValidator{{ resolved: ResolvedValidator{Binding: Binding("reject-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { return contracts.ValidationResult{Approved: false, ReasonCode: "rejected", Message: "not accepted"}, nil @@ -294,7 +294,7 @@ func TestRunnerAttemptDebugRepresentsFailuresAndFinalRejection(t *testing.T) { { name: "normalize serialization error", configure: func(prepared *PreparedPipeline) { - prepared.lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + prepared.Steps[0].lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { return erasedTypedResult{Value: "wrong artifact type"}, nil } }, @@ -331,13 +331,13 @@ func TestRunnerKeepsValidatorLLMCallsOutOfModuleAttempt(t *testing.T) { prepared := preparedAttemptDebugPipeline(t) debug := newCapturedDebugRecorder() client := WithDebugLLMRecording(attemptDebugLLM{}, debug) - prepared.lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + prepared.Steps[0].lanes[0].typed.merge = func(ctx context.Context, _ any, _ contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { if err := callAttemptDebugLLM(ctx, client, "merge-module"); err != nil { return erasedTypedResult{}, err } return erasedTypedResult{Value: codecNotes{Items: []string{"merged"}}}, nil } - prepared.lanes[0].mergeValidators.validators = []preparedValidator{{ + prepared.Steps[0].lanes[0].mergeValidators.validators = []preparedValidator{{ resolved: ResolvedValidator{Binding: Binding("llm-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(ctx context.Context, _ any, _ typedValidationTarget) (contracts.ValidationResult, error) { if err := callAttemptDebugLLM(ctx, client, "merge-validator"); err != nil { @@ -389,7 +389,7 @@ func (l attemptReuseLoader) Normalize(laneID, _ string, _ []CheckpointFingerprin func TestRunnerCheckpointReuseDoesNotSynthesizeModuleAttempts(t *testing.T) { prepared := preparedAttemptDebugPipeline(t) - lane := prepared.lanes[0] + lane := prepared.Steps[0].lanes[0] merge, err := checkpointArtifact(lane.typed.codec, lane.resolved.ID, lane.resolved.Merge.Module, "source", codecNotes{Items: []string{"merged"}}) if err != nil { t.Fatalf("checkpointArtifact(merge): %v", err) diff --git a/internal/framework/pipeline/runner_candidate_encoding_test.go b/internal/framework/pipeline/runner_candidate_encoding_test.go index 58ad3f8..f8bebd8 100644 --- a/internal/framework/pipeline/runner_candidate_encoding_test.go +++ b/internal/framework/pipeline/runner_candidate_encoding_test.go @@ -104,11 +104,11 @@ func installObservedNotesCodec(t *testing.T, prepared *PreparedPipeline, codec * if err != nil { t.Fatalf("codec entry error = %v", err) } - prepared.lanes[0].typed.codec = entry + prepared.Steps[0].lanes[0].typed.codec = entry } func configureCandidateOperation(prepared *PreparedPipeline, target ModuleStage, value codecNotes) { - lane := &prepared.lanes[0] + lane := &prepared.Steps[0].lanes[0] switch target { case StageMerge: lane.typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { @@ -136,9 +136,9 @@ func setCandidateValidator(prepared *PreparedPipeline, target ModuleStage, appro } switch target { case StageMerge: - prepared.lanes[0].mergeValidators.validators = []preparedValidator{validator} + prepared.Steps[0].lanes[0].mergeValidators.validators = []preparedValidator{validator} case StageNormalize: - prepared.lanes[0].normalizeValidators.validators = []preparedValidator{validator} + prepared.Steps[0].lanes[0].normalizeValidators.validators = []preparedValidator{validator} } } diff --git a/internal/framework/pipeline/runner_chunk_plan_test.go b/internal/framework/pipeline/runner_chunk_plan_test.go index 1a051de..539cf8b 100644 --- a/internal/framework/pipeline/runner_chunk_plan_test.go +++ b/internal/framework/pipeline/runner_chunk_plan_test.go @@ -469,9 +469,9 @@ func TestRunnerOmitsProducerProfileForDeterministicChunker(t *testing.T) { func TestRunnerRefreshChangesDownstreamChunkFingerprint(t *testing.T) { doc := typedTestDocumentWithUnits(2) prepared := preparedConcurrentPipeline(t, 1) - prepared.lanes = prepared.lanes[:1] + prepared.Steps[0].lanes = prepared.Steps[0].lanes[:1] prepared.resolved.Steps[0].ArtifactLanes = prepared.resolved.Steps[0].ArtifactLanes[:1] - prepared.ArtifactLanes = prepared.ArtifactLanes[:1] + prepared.Steps[0].ArtifactLanes = prepared.Steps[0].ArtifactLanes[:1] prepared.input = &typedTestInput{key: prepared.resolved.Input.Module, doc: doc} run := func(plan source.ChunkPlan) []CheckpointFingerprint { diff --git a/internal/framework/pipeline/runner_concurrency_test.go b/internal/framework/pipeline/runner_concurrency_test.go index 1faebbd..769ba89 100644 --- a/internal/framework/pipeline/runner_concurrency_test.go +++ b/internal/framework/pipeline/runner_concurrency_test.go @@ -11,6 +11,7 @@ import ( "testing" "time" + "gitea.maximumdirect.net/eric/notarius/internal/core/source" "gitea.maximumdirect.net/eric/notarius/internal/framework/contracts" frameworkllm "gitea.maximumdirect.net/eric/notarius/internal/framework/llm" ) @@ -60,8 +61,169 @@ func preparedConcurrentPipeline(t *testing.T, chunkCount int) *PreparedPipeline return prepared } +type orderedLaneSpec struct { + id string + profile string +} + +func preparedOrderedPipeline(t *testing.T, chunkCount int, specs ...orderedLaneSpec) *PreparedPipeline { + t.Helper() + catalog := typedResolutionCatalog(t, completeTypedCatalogOptions()) + profile := typedResolutionProfile() + profile.Artifacts = nil + for index, spec := range specs { + lane, ok := typedResolutionProfile().Artifacts[spec.profile] + if !ok { + t.Fatalf("typed lane profile %q is not defined", spec.profile) + } + profile.Steps = append(profile.Steps, PipelineStepProfile{ + ID: fmt.Sprintf("step-%d", index+1), + Artifacts: map[string]ArtifactLaneProfile{spec.id: lane}, + }) + } + resolved, err := ResolvePipeline(profile, ResolveOptions{}, catalog) + if err != nil { + t.Fatalf("ResolvePipeline() error = %v", err) + } + prepared, err := Prepare(resolved, registriesFromModuleCatalog(catalog), ModuleDependencies{}) + if err != nil { + t.Fatalf("Prepare() error = %v", err) + } + doc := typedTestDocumentWithUnits(chunkCount) + if adapter, ok := prepared.input.(*typedTestInput); ok { + adapter.doc = doc + } else { + t.Fatalf("prepared input = %T, want *typedTestInput", prepared.input) + } + prepared.chunker = &typedTestChunker{key: "typed/chunk", plan: typedTestPlan(doc)} + return prepared +} + +type countingOrderedOutput struct { + calls atomic.Int32 +} + +func (o *countingOrderedOutput) Key() string { return "typed/output" } +func (o *countingOrderedOutput) Encode(context.Context, contracts.OutputRequest) (contracts.OutputResult, error) { + o.calls.Add(1) + return contracts.OutputResult{}, nil +} + +type countingInputAdapter struct { + contracts.InputAdapter + calls atomic.Int32 +} + +func (a *countingInputAdapter) Parse(ctx context.Context, request contracts.ParseRequest) (*source.SourceDocument, error) { + a.calls.Add(1) + return a.InputAdapter.Parse(ctx, request) +} + +func TestRunnerRejectsGeneratedBindingsBeforeSourceParsing(t *testing.T) { + prepared := preparedConcurrentPipeline(t, 1) + input := &countingInputAdapter{InputAdapter: prepared.input} + prepared.input = input + prepared.resolved.Steps[0].ArtifactLanes[0].ExtractReferences.Bindings = []ReferenceBinding{{ + Stage: StageExtract, + LaneID: prepared.resolved.Steps[0].ArtifactLanes[0].ID, + SlotName: "generated", + Artifact: &ArtifactReference{Step: "producer", Lane: "source"}, + }} + + _, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input")}) + if err == nil || !strings.Contains(err.Error(), "generated reference execution is not supported yet") { + t.Fatalf("Run() error = %v, want generated binding rejection", err) + } + if got := input.calls.Load(); got != 0 { + t.Fatalf("input Parse calls = %d, want zero", got) + } +} + +func TestRunnerExecutesOrderedStepsWithHardBarriers(t *testing.T) { + prepared := preparedOrderedPipeline(t, 1, + orderedLaneSpec{id: "notes", profile: "notes"}, + orderedLaneSpec{id: "score", profile: "score"}, + ) + var mu sync.Mutex + var events []string + record := func(event string) { + mu.Lock() + events = append(events, event) + mu.Unlock() + } + first := &prepared.Steps[0].lanes[0] + second := &prepared.Steps[1].lanes[0] + first.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecNotes{Items: []string{"first"}}}, nil + } + originalFirstNormalize := first.typed.normalize + first.typed.normalize = func(ctx context.Context, implementation any, request contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + result, err := originalFirstNormalize(ctx, implementation, request) + if err == nil { + record("first-normalize-done") + } + return result, err + } + second.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + record("second-extract-started") + return erasedTypedResult{Value: codecScore{Value: 2}}, nil + } + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2}) + if err != nil { + t.Fatalf("Run() error = %v", err) + } + mu.Lock() + gotEvents := append([]string(nil), events...) + mu.Unlock() + if want := []string{"first-normalize-done", "second-extract-started"}; !reflect.DeepEqual(gotEvents, want) { + t.Fatalf("ordered events = %#v, want %#v", gotEvents, want) + } + if got := []string{output.NormalizeOutputs[0].LaneID, output.NormalizeOutputs[1].LaneID}; !reflect.DeepEqual(got, []string{"notes", "score"}) { + t.Fatalf("normalized lane order = %#v, want notes then score", got) + } +} + +func TestRunnerStopsLaterOrderedStepsAfterFailure(t *testing.T) { + prepared := preparedOrderedPipeline(t, 1, + orderedLaneSpec{id: "first", profile: "notes"}, + orderedLaneSpec{id: "failure", profile: "score"}, + orderedLaneSpec{id: "later", profile: "notes"}, + ) + first := &prepared.Steps[0].lanes[0] + second := &prepared.Steps[1].lanes[0] + third := &prepared.Steps[2].lanes[0] + first.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{Value: codecNotes{Items: []string{"first"}}}, nil + } + second.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + return erasedTypedResult{}, errors.New("ordered step extraction failed") + } + var thirdCalls atomic.Int32 + third.typed.extract = func(context.Context, any, contracts.TypedExtractionRequest) (erasedTypedResult, error) { + thirdCalls.Add(1) + return erasedTypedResult{Value: codecNotes{Items: []string{"later"}}}, nil + } + encoder := &countingOrderedOutput{} + prepared.output = encoder + + output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2}) + if err == nil || !strings.Contains(err.Error(), `execute pipeline step "step-2"`) { + t.Fatalf("Run() error = %v, want step-scoped failure", err) + } + if len(output.NormalizeOutputs) != 1 || output.NormalizeOutputs[0].LaneID != "first" { + t.Fatalf("completed outputs = %#v, want only the first step output", output.NormalizeOutputs) + } + if got := thirdCalls.Load(); got != 0 { + t.Fatalf("later step extract calls = %d, want zero", got) + } + if got := encoder.calls.Load(); got != 0 { + t.Fatalf("output encoder calls = %d, want zero after failure", got) + } +} + func installExtractOperation(prepared *PreparedPipeline, laneIndex int, operation func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error)) { - prepared.lanes[laneIndex].typed.extract = func(ctx context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { + prepared.Steps[0].lanes[laneIndex].typed.extract = func(ctx context.Context, _ any, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { return operation(ctx, request) } } @@ -82,7 +244,7 @@ func TestRunnerBoundsExtractJobsAndStabilizesReverseCompletion(t *testing.T) { } var active atomic.Int32 var maximum atomic.Int32 - for laneIndex := range prepared.lanes { + for laneIndex := range prepared.Steps[0].lanes { lane := laneIndex installExtractOperation(prepared, lane, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { jobIndex := request.Chunk.Index*2 + lane @@ -154,8 +316,8 @@ func TestRunnerStartsLaneContinuationWhileOtherLaneExtractsRemain(t *testing.T) return erasedTypedResult{}, ctx.Err() } }) - originalMerge := prepared.lanes[0].typed.merge - prepared.lanes[0].typed.merge = func(ctx context.Context, implementation any, request contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + originalMerge := prepared.Steps[0].lanes[0].typed.merge + prepared.Steps[0].lanes[0].typed.merge = func(ctx context.Context, implementation any, request contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { select { case mergeStarted <- struct{}{}: default: @@ -180,7 +342,7 @@ func TestRunnerStartsLaneContinuationWhileOtherLaneExtractsRemain(t *testing.T) func TestRunnerBoundsConcurrentLaneContinuations(t *testing.T) { prepared := preparedConcurrentPipeline(t, 1) - base := prepared.lanes[0] + base := prepared.Steps[0].lanes[0] lanes := make([]preparedLaneExecutor, 8) resolvedLanes := make([]ResolvedArtifactLane, len(lanes)) publicLanes := make([]PreparedArtifactLane, len(lanes)) @@ -215,9 +377,9 @@ func TestRunnerBoundsConcurrentLaneContinuations(t *testing.T) { resolvedLanes[i] = lane.resolved publicLanes[i] = PreparedArtifactLane{Resolved: lane.resolved} } - prepared.lanes = lanes + prepared.Steps[0].lanes = lanes prepared.resolved.Steps[0].ArtifactLanes = resolvedLanes - prepared.ArtifactLanes = publicLanes + prepared.Steps[0].ArtifactLanes = publicLanes output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), ExtractWorkers: 2}) if err != nil { @@ -235,7 +397,7 @@ func TestRunnerSelectsFrameworkErrorByStableLaneOrder(t *testing.T) { prepared := preparedConcurrentPipeline(t, 1) ready := make(chan struct{}, 2) release := make(chan struct{}) - for laneIndex := range prepared.lanes { + for laneIndex := range prepared.Steps[0].lanes { lane := laneIndex installExtractOperation(prepared, lane, func(context.Context, contracts.TypedExtractionRequest) (erasedTypedResult, error) { ready <- struct{}{} @@ -258,7 +420,7 @@ func TestRunnerSelectsFrameworkErrorByStableLaneOrder(t *testing.T) { func TestRunnerSelectsMergeErrorBeforeEarlierLaneNormalizeError(t *testing.T) { prepared := preparedConcurrentPipeline(t, 1) - for laneIndex := range prepared.lanes { + for laneIndex := range prepared.Steps[0].lanes { lane := laneIndex installExtractOperation(prepared, lane, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index)}, nil @@ -266,12 +428,12 @@ func TestRunnerSelectsMergeErrorBeforeEarlierLaneNormalizeError(t *testing.T) { } ready := make(chan struct{}, 2) release := make(chan struct{}) - prepared.lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { + prepared.Steps[0].lanes[0].typed.normalize = func(context.Context, any, contracts.TypedNormalizeRequest[any]) (erasedTypedResult, error) { ready <- struct{}{} <-release return erasedTypedResult{}, errors.New("earlier lane normalize failure") } - prepared.lanes[1].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { + prepared.Steps[0].lanes[1].typed.merge = func(context.Context, any, contracts.TypedMergeRequest[any]) (erasedTypedResult, error) { ready <- struct{}{} <-release return erasedTypedResult{}, errors.New("later lane merge failure") @@ -291,13 +453,13 @@ func TestRunnerSelectsMergeErrorBeforeEarlierLaneNormalizeError(t *testing.T) { func TestRunnerKeepsSuccessfulChunksWhenAnotherExtractIsRejected(t *testing.T) { prepared := preparedConcurrentPipeline(t, 2) - for laneIndex := range prepared.lanes { + for laneIndex := range prepared.Steps[0].lanes { lane := laneIndex installExtractOperation(prepared, lane, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index)}, nil }) } - validator := &prepared.lanes[0].extractValidators.validators[0] + validator := &prepared.Steps[0].lanes[0].extractValidators.validators[0] validator.typedValidate = func(_ context.Context, _ any, target typedValidationTarget) (contracts.ValidationResult, error) { if target.chunk != nil && target.chunk.Index == 0 { return contracts.ValidationResult{Approved: false, ReasonCode: "expected_rejection", Message: "rejected by test"}, nil @@ -308,7 +470,7 @@ func TestRunnerKeepsSuccessfulChunksWhenAnotherExtractIsRejected(t *testing.T) { if err != nil { t.Fatalf("Run() error = %v", err) } - if len(output.Rejected) != 1 || output.Rejected[0].LaneID != prepared.lanes[0].resolved.ID || output.Rejected[0].ChunkIndex != 0 { + if len(output.Rejected) != 1 || output.Rejected[0].LaneID != prepared.Steps[0].lanes[0].resolved.ID || output.Rejected[0].ChunkIndex != 0 { t.Fatalf("rejections = %#v, want the first lane's first chunk", output.Rejected) } if len(output.NormalizeOutputs) != 2 { @@ -330,9 +492,9 @@ func TestRunnerSeparatelyBoundsExtractJobsAndSharedProviderCalls(t *testing.T) { var jobActive atomic.Int32 var jobMaximum atomic.Int32 var attempts sync.Map - for laneIndex := range prepared.lanes { + for laneIndex := range prepared.Steps[0].lanes { lane := laneIndex - prepared.lanes[lane].resolved.Extract.Retries = 1 + prepared.Steps[0].lanes[lane].resolved.Extract.Retries = 1 installExtractOperation(prepared, lane, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { current := jobActive.Add(1) defer jobActive.Add(-1) @@ -353,7 +515,7 @@ func TestRunnerSeparatelyBoundsExtractJobsAndSharedProviderCalls(t *testing.T) { } return erasedTypedResult{Value: typedValueForLane(lane, request.Chunk.Index)}, nil }) - validator := &prepared.lanes[lane].extractValidators.validators[0] + validator := &prepared.Steps[0].lanes[lane].extractValidators.validators[0] validator.typedValidate = func(ctx context.Context, _ any, _ typedValidationTarget) (contracts.ValidationResult, error) { if _, callErr := client.CompleteStructured(ctx, contracts.StructuredCompletionRequest{StageName: "validate"}, nil); callErr != nil { return contracts.ValidationResult{}, callErr @@ -382,7 +544,7 @@ func TestRunnerSeparatelyBoundsExtractJobsAndSharedProviderCalls(t *testing.T) { func TestRunnerReturnsParentCancellationAndStopsQueuedExtracts(t *testing.T) { prepared := preparedConcurrentPipeline(t, 4) started := make(chan struct{}, 8) - for laneIndex := range prepared.lanes { + for laneIndex := range prepared.Steps[0].lanes { lane := laneIndex installExtractOperation(prepared, lane, func(ctx context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { started <- struct{}{} diff --git a/internal/framework/pipeline/runner_concurrent.go b/internal/framework/pipeline/runner_concurrent.go index 9498da8..b7664c4 100644 --- a/internal/framework/pipeline/runner_concurrent.go +++ b/internal/framework/pipeline/runner_concurrent.go @@ -72,10 +72,10 @@ type orderedRunError struct { err error } -func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk) (RunOutput, error) { +func (r *Runner) runLanes(parent context.Context, input RunInput, step PreparedPipelineStep, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk) (RunOutput, error) { output := RunOutput{Manifest: manifestFromPipeline(input)} - states := make([]*laneExtractState, len(input.Prepared.lanes)) - for i, prepared := range input.Prepared.lanes { + states := make([]*laneExtractState, len(step.lanes)) + for i, prepared := range step.lanes { if prepared.typed == nil { return output, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID) } @@ -280,7 +280,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. started := time.Now().UTC() attemptPath := path.Join("extract", debugPathComponent(lane.ID), fmt.Sprintf("chunk-%06d", chunk.Index+1), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) - terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started}) + terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "extract", llmScope, debugTimedEnvelope{Stage: string(StageExtract), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started}) requestMetadata, metadataErr := cloneMetadata(input.Metadata) if metadataErr != nil { return false, nil, terminal.record(nil, fmt.Errorf("clone extract request metadata: %w", metadataErr)) @@ -299,7 +299,7 @@ func (r *Runner) runExtractJob(ctx context.Context, input RunInput, doc *source. return false, nil, terminal.record(payload, attemptErr) } serializedCandidate.ChunkID, serializedCandidate.ChunkIndex, serializedCandidate.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value, candidate: &serializedCandidate}, state.prepared.extractValidators, attempt, input.Debug) + warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageExtract, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Extract.Module, source: doc, sourceID: doc.ID, sourceInput: chunkInputMaterial(sourceInput, chunk), sessionID: sessionID, references: lane.ExtractReferences.ReferenceSet, metadata: input.Metadata, chunk: &chunk, ref: chunk.Ref, value: extracted.Value, candidate: &serializedCandidate}, state.prepared.extractValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} if validateErr != nil || rejected != nil { @@ -368,10 +368,10 @@ 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) - if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageExtract), 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 { + 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} } - if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": results.decision.Reused, "outputs": debugCheckpointArtifacts(results.serialized), "rejected": debugRejectedOutputEnvelopes(results.rejected), "warnings": debugWarningEnvelopes(results.warnings)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("extract", debugPathComponent(lane.ID), "output.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, "outputs": debugCheckpointArtifacts(results.serialized), "rejected": debugRejectedOutputEnvelopes(results.rejected), "warnings": debugWarningEnvelopes(results.warnings)}}); err != nil { return local, &laneRunError{stage: StageExtract, err: err} } if len(results.accepted) == 0 { diff --git a/internal/framework/pipeline/runner_extract_handoff_test.go b/internal/framework/pipeline/runner_extract_handoff_test.go index 531ea14..2ba4fd6 100644 --- a/internal/framework/pipeline/runner_extract_handoff_test.go +++ b/internal/framework/pipeline/runner_extract_handoff_test.go @@ -212,7 +212,7 @@ func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) { func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) { prepared := preparedAttemptDebugPipeline(t) - prepared.lanes[0].resolved.Extract.Retries = 1 + prepared.Steps[0].lanes[0].resolved.Extract.Retries = 1 attempts := 0 installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) { attempts++ @@ -226,7 +226,7 @@ func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) { }, nil }) validatorCalls := 0 - prepared.lanes[0].extractValidators.validators[0].typedValidate = func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { + prepared.Steps[0].lanes[0].extractValidators.validators[0].typedValidate = func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { validatorCalls++ return contracts.ValidationResult{Approved: validatorCalls == 2, ReasonCode: "retry", Message: "retry extract"}, nil } diff --git a/internal/framework/pipeline/runner_terminal_debug_test.go b/internal/framework/pipeline/runner_terminal_debug_test.go index 7afa14b..a89b4af 100644 --- a/internal/framework/pipeline/runner_terminal_debug_test.go +++ b/internal/framework/pipeline/runner_terminal_debug_test.go @@ -220,7 +220,7 @@ func TestRunnerRecordsExtractTerminalOutcomes(t *testing.T) { } return erasedTypedResult{Value: codecNotes{Items: []string{value}}, Warnings: []contracts.Warning{{Scope: "extract", ReasonCode: "observed", Message: "extract warning"}}}, nil }) - prepared.lanes[0].extractValidators.validators = []preparedValidator{{ + prepared.Steps[0].lanes[0].extractValidators.validators = []preparedValidator{{ resolved: ResolvedValidator{Binding: Binding("terminal/extract-validator"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(context.Context, any, typedValidationTarget) (contracts.ValidationResult, error) { return contracts.ValidationResult{Approved: !tc.reject, ReasonCode: "extract_rejected", Message: "extract rejected"}, tc.validatorErr @@ -315,7 +315,7 @@ func TestRunnerKeepsExtractModuleAndValidatorLLMCallsIsolated(t *testing.T) { } return erasedTypedResult{Value: typedValueForLane(0, request.Chunk.Index)}, nil }) - prepared.lanes[0].extractValidators.validators = []preparedValidator{{ + prepared.Steps[0].lanes[0].extractValidators.validators = []preparedValidator{{ resolved: ResolvedValidator{Binding: Binding("llm-check"), Target: ValidatorTargetTyped, ArtifactKind: "test/notes"}, typedValidate: func(ctx context.Context, _ any, _ typedValidationTarget) (contracts.ValidationResult, error) { if err := callAttemptDebugLLM(ctx, client, "extract-validator"); err != nil { diff --git a/internal/framework/pipeline/runner_typed.go b/internal/framework/pipeline/runner_typed.go index 476bbba..5c42c5a 100644 --- a/internal/framework/pipeline/runner_typed.go +++ b/internal/framework/pipeline/runner_typed.go @@ -162,7 +162,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi } } recordCheckpointEvent(output, loader, string(StageMerge), lane.ID, lane.Merge.Module, mergeDecision) - if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), 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 { + 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 } var merged erasedMergeArtifact @@ -188,7 +188,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi started := time.Now().UTC() attemptPath := path.Join("merge", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) - terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started}) + terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "merge", llmScope, debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, Attempt: attempt, StartedAt: started}) requestMetadata, metadataErr := cloneMetadata(input.Metadata) if metadataErr != nil { return false, nil, terminal.record(nil, fmt.Errorf("clone merge request metadata: %w", metadataErr)) @@ -205,7 +205,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi attemptErr := fmt.Errorf("serialize merge candidate for lane %q: %w", lane.ID, encodeErr) return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug) + warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageMerge, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Merge.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.MergeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.mergeValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} if validateErr != nil || rejected != nil { @@ -239,7 +239,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi return err } } - if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageMerge), LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "output.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, "accepted": true, "output": debugCheckpointArtifact(serializedMerge), "warnings": debugWarningEnvelopes(mergeWarnings)}}); err != nil { return err } @@ -252,7 +252,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi } } recordCheckpointEvent(output, loader, string(StageNormalize), lane.ID, lane.Normalize.Module, normalizeDecision) - if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), 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 { + 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 @@ -276,7 +276,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi started := time.Now().UTC() attemptPath := path.Join("normalize", debugPathComponent(lane.ID), fmt.Sprintf("attempt-%02d", attempt)) attemptCtx, llmScope := withDebugLLMScope(ctx, attemptPath) - terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started}) + terminal := newAttemptTerminalRecorder(input.Debug, attemptPath, "normalize", llmScope, debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, Attempt: attempt, StartedAt: started}) requestMetadata, metadataErr := cloneMetadata(input.Metadata) if metadataErr != nil { return false, nil, terminal.record(nil, fmt.Errorf("clone normalize request metadata: %w", metadataErr)) @@ -292,7 +292,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi attemptErr := fmt.Errorf("serialize normalize candidate for lane %q: %w", lane.ID, encodeErr) return false, nil, terminal.record(map[string]any{"warnings": debugWarningEnvelopes(attemptWarnings)}, attemptErr) } - warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, attempt, input.Debug) + warnings, rejected, validateErr := r.validateTypedArtifact(attemptCtx, typed.codec, typedValidationTarget{stage: StageNormalize, stepID: input.stepID, laneID: lane.ID, moduleKey: lane.Normalize.Module, source: doc, sourceID: doc.ID, sourceInput: sourceInput.Clone(), sessionID: sessionID, references: lane.NormalizeReferences.ReferenceSet, metadata: input.Metadata, value: result.Value, candidate: &serializedCandidate}, prepared.normalizeValidators, attempt, input.Debug) attemptWarnings = append(attemptWarnings, warnings...) payload := map[string]any{"output": debugCheckpointArtifact(serializedCandidate), "warnings": debugWarningEnvelopes(attemptWarnings), "rejection": debugRejectedOutputPtr(rejected)} if validateErr != nil || rejected != nil { @@ -326,7 +326,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi return err } } - if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.json"), debugTimedEnvelope{Stage: string(StageNormalize), LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil { + if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "output.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, "accepted": true, "output": debugCheckpointArtifact(serializedNormalize), "warnings": debugWarningEnvelopes(normalizeWarnings)}}); err != nil { return err } output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(serializedNormalize.Artifact)}) @@ -414,12 +414,12 @@ func (r *Runner) validateTypedArtifact(ctx context.Context, codec artifactCodecE } if err != nil { validationErr := fmt.Errorf("validate typed %s output with validator %q: %w", target.stage, binding.Module, err) - if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { + if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall, Error: debugCall.Error}, llmScope); debugErr != nil { return warnings, nil, errors.Join(validationErr, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr)) } return warnings, nil, validationErr } - if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil { + if debugErr := writeDebugAttempt(debug, attemptPath, debugTimedEnvelope{Stage: string(target.stage), StepID: target.stepID, LaneID: target.laneID, ModuleKey: target.moduleKey, Attempt: attempt, StartedAt: started, Payload: debugCall}, llmScope); debugErr != nil { return warnings, nil, fmt.Errorf("write typed validator attempt debug artifact: %w", debugErr) } if !result.Approved { diff --git a/internal/framework/pipeline/typed_execution.go b/internal/framework/pipeline/typed_execution.go index cbcda4c..99f41f5 100644 --- a/internal/framework/pipeline/typed_execution.go +++ b/internal/framework/pipeline/typed_execution.go @@ -28,6 +28,7 @@ type erasedTypedResult struct { type typedValidationTarget struct { stage ModuleStage + stepID string laneID string moduleKey string source *source.SourceDocument diff --git a/internal/framework/pipeline/typed_resolution_test.go b/internal/framework/pipeline/typed_resolution_test.go index 2cad546..231bd2b 100644 --- a/internal/framework/pipeline/typed_resolution_test.go +++ b/internal/framework/pipeline/typed_resolution_test.go @@ -176,8 +176,8 @@ func TestPrepareConstructsHeterogeneousTypedLanes(t *testing.T) { if err != nil { t.Fatalf("Prepare() error = %v, want nil", err) } - if len(prepared.ArtifactLanes) != 2 || prepared.lanes[0].typed == nil || prepared.lanes[1].typed == nil { - t.Fatalf("PreparedPipeline lanes = %#v, want two typed executors", prepared.ArtifactLanes) + if len(prepared.Steps[0].ArtifactLanes) != 2 || prepared.Steps[0].lanes[0].typed == nil || prepared.Steps[0].lanes[1].typed == nil { + t.Fatalf("PreparedPipeline lanes = %#v, want two typed executors", prepared.Steps[0].ArtifactLanes) } }