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

Decompose pipeline runner orchestration

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-22 02:03:27 +00:00
parent 64ea23c21f
commit 5bdd56cfb1
4 changed files with 146 additions and 91 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -112,6 +112,30 @@ func requireReusableCheckpoint(policy CheckpointExecutionPolicy, stepID, laneID
return nil
}
// resolveCheckpointDecision applies runner policy and canonical payload
// validation at the single point where a stage's observable decision is made.
func resolveCheckpointDecision(output *RunOutput, loader CheckpointLoader, policy CheckpointExecutionPolicy, stage ModuleStage, stepID, laneID, moduleKey string, decision CheckpointDecision, codec artifactCodecEntry, artifacts []CheckpointArtifact) (CheckpointDecision, error) {
decision = forceCheckpointDecision(policy, stepID, laneID, decision)
if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil {
return decision, err
}
if decision.Reused {
for _, artifact := range artifacts {
if _, _, err := decodeCanonicalCheckpointArtifact(codec, artifact); err != nil {
decision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored " + string(stage) + " artifact failed canonical codec validation", Reason: string(stage) + " artifact checkpoint is not canonical"}
break
}
}
}
if err := requireReusableCheckpoint(policy, stepID, laneID, decision); err != nil {
return decision, err
}
if output != nil {
recordCheckpointEvent(output, loader, string(stage), stepID, laneID, moduleKey, decision)
}
return decision, nil
}
type SourceCheckpoint struct {
Document *source.SourceDocument
}

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

@@ -252,22 +252,8 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
}
if chunkResult.accepted {
for _, step := range input.Prepared.Steps {
stepInput := input
stepInput.stepID = step.ID
stepReferences, referenceProvenance, handoffErr := buildStepReferenceSets(input, step, output.NormalizeOutputs)
if handoffErr != nil {
return failOutput(output), fmt.Errorf("prepare generated references for pipeline step %q: %w", step.ID, handoffErr)
}
stepInput.references = stepReferences
output.Manifest.References = append(output.Manifest.References, referenceProvenance...)
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)
}
if err := r.runPreparedSteps(ctx, input, checkpoints, checkpointLoader, doc, sourceInput, sessionID, chunkResult.chunks, &output); err != nil {
return failOutput(output), err
}
}
@@ -335,6 +321,27 @@ func (r *Runner) Run(ctx context.Context, input RunInput) (output RunOutput, err
return output, nil
}
func (r *Runner) runPreparedSteps(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, output *RunOutput) error {
for _, step := range input.Prepared.Steps {
stepInput := input
stepInput.stepID = step.ID
stepReferences, referenceProvenance, err := buildStepReferenceSets(input, step, output.NormalizeOutputs)
if err != nil {
return fmt.Errorf("prepare generated references for pipeline step %q: %w", step.ID, err)
}
stepInput.references = stepReferences
output.Manifest.References = append(output.Manifest.References, referenceProvenance...)
laneOutput, laneErr := r.runLanes(ctx, stepInput, step, checkpoints, loader, doc, sourceInput, sessionID, chunks)
if err := mergeLaneOutput(output, laneOutput); err != nil {
return err
}
if laneErr != nil {
return fmt.Errorf("execute pipeline step %q: %w", step.ID, laneErr)
}
}
return nil
}
func runWithRetry(ctx context.Context, retries int, run func(attempt int) (bool, *contracts.RejectedOutput, error)) (bool, *contracts.RejectedOutput, error) {
attempts := retries + 1
var last *contracts.RejectedOutput

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

@@ -77,28 +77,46 @@ type orderedRunError struct {
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, err := initializeLaneStates(input, step, checkpoints, loader, doc, chunks, &output)
if err != nil {
return output, err
}
completedOutputs, runErrors := r.runLaneEngine(parent, input, checkpoints, loader, doc, sourceInput, sessionID, chunks, states)
if err := mergeCompletedLanes(&output, completedOutputs); err != nil {
return output, err
}
if err := selectRunError(parent, runErrors); err != nil {
return output, err
}
return output, nil
}
func initializeLaneStates(input RunInput, step PreparedPipelineStep, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, chunks []source.Chunk, output *RunOutput) ([]*laneExtractState, error) {
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)
return nil, fmt.Errorf("typed lane %q executor is not prepared", prepared.resolved.ID)
}
if err := setTypedLaneManifestMetadata(&output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil {
return output, err
if err := setTypedLaneManifestMetadata(output, prepared.resolved.ID, prepared.typed.extractor, prepared.typed.merger, prepared.typed.normalizer); err != nil {
return nil, err
}
state, err := prepareLaneExtract(input, loader, doc, chunks, i, prepared)
if err != nil {
return output, err
return nil, err
}
if !state.decision.Reused {
if err := checkpointExtractRunning(checkpoints, input.stepID, prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil {
return output, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err)
return nil, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err)
}
} else if err := finalizeLaneExtract(checkpoints, input.stepID, state); err != nil {
return output, err
return nil, err
}
states[i] = state
}
return states, nil
}
func (r *Runner) runLaneEngine(parent context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, states []*laneExtractState) ([]RunOutput, []orderedRunError) {
workerCount := input.ExtractWorkers
if workerCount < 1 {
workerCount = 1
@@ -216,15 +234,16 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, step PreparedP
}
close(continuations)
continuationWorkers.Wait()
return completedOutputs, runErrors
}
func mergeCompletedLanes(output *RunOutput, completedOutputs []RunOutput) error {
for i := range completedOutputs {
if err := mergeLaneOutput(&output, completedOutputs[i]); err != nil {
return output, err
if err := mergeLaneOutput(output, completedOutputs[i]); err != nil {
return err
}
}
if err := selectRunError(parent, runErrors); err != nil {
return output, err
}
return output, nil
return nil
}
func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.SourceDocument, chunks []source.Chunk, index int, prepared preparedLaneExecutor) (*laneExtractState, error) {
@@ -237,19 +256,8 @@ func prepareLaneExtract(input RunInput, loader CheckpointLoader, doc *source.Sou
deps := append(digestFingerprints("chunks", digest), generatedReferenceDependencies(extractReferences)...)
state := &laneExtractState{index: index, prepared: prepared, deps: normalizeCheckpointFingerprints(deps), remaining: len(chunks), results: make(map[int]extractJobResult, len(chunks))}
cp, decision := loadExtract(loader, input.stepID, lane.ID, lane.Extract.Module, state.deps)
decision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, decision)
if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, decision); err != nil {
return nil, err
}
if decision.Reused {
for _, stored := range cp.Outputs {
if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, stored); decodeErr != nil {
decision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored extract artifact failed canonical codec validation", Reason: "extract artifact checkpoint is not canonical"}
break
}
}
}
if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, decision); err != nil {
decision, err = resolveCheckpointDecision(nil, loader, input.CheckpointPolicy, StageExtract, input.stepID, lane.ID, lane.Extract.Module, decision, typed.codec, cp.Outputs)
if err != nil {
return nil, err
}
state.decision = decision

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

@@ -160,13 +160,13 @@ type laneRunError struct {
func (e *laneRunError) Error() string { return e.err.Error() }
func (e *laneRunError) Unwrap() error { return e.err }
func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) (err error) {
activeStage := StageMerge
defer func() {
if err != nil {
err = &laneRunError{stage: activeStage, err: err}
}
}()
type mergeStageResult struct {
artifact erasedMergeArtifact
serialized CheckpointArtifact
terminal bool
}
func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) error {
lane, typed := prepared.resolved, prepared.typed
if typed == nil {
return fmt.Errorf("typed lane %q executor is not prepared", lane.ID)
@@ -174,7 +174,26 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
if err := setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer); err != nil {
return err
}
merged, err := r.runMergeStage(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, prepared, extracts, output)
if err != nil {
return &laneRunError{stage: StageMerge, err: err}
}
if merged.terminal {
return nil
}
normalized, err := r.runNormalizeStage(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, prepared, merged.artifact, merged.serialized, output)
if err != nil {
return &laneRunError{stage: StageNormalize, err: err}
}
if normalized.accepted {
output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{StepID: input.stepID, LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(normalized.serialized.Artifact)})
}
return nil
}
func (r *Runner) runMergeStage(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, extracts finalizedExtractResults, output *RunOutput) (mergeStageResult, error) {
var stageResult mergeStageResult
lane, typed := prepared.resolved, prepared.typed
mergeInputs := make([]contracts.ExtractArtifact[any], len(extracts.accepted))
for i, value := range extracts.accepted {
mergeInputs[i] = contracts.ExtractArtifact[any]{LaneID: value.LaneID, ExtractorKey: value.ExtractorKey, SourceID: value.SourceID, ChunkID: value.ChunkID, ChunkIndex: value.ChunkIndex, ChunkRef: value.ChunkRef, Value: value.Value}
@@ -183,21 +202,12 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
mergeDeps := append(artifactCheckpointDigests(extracts.serialized), generatedReferenceDependencies(mergeReferences)...)
mergeDeps = normalizeCheckpointFingerprints(mergeDeps)
mergeCP, mergeDecision := loadMerge(loader, input.stepID, lane.ID, lane.Merge.Module, mergeDeps)
mergeDecision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision)
if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision); err != nil {
return err
mergeDecision, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageMerge, input.stepID, lane.ID, lane.Merge.Module, mergeDecision, typed.codec, []CheckpointArtifact{mergeCP.Output})
if err != nil {
return stageResult, err
}
if mergeDecision.Reused {
if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil {
mergeDecision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored merge artifact failed canonical codec validation", Reason: "merge artifact checkpoint is not canonical"}
}
}
if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, mergeDecision); err != nil {
return err
}
recordCheckpointEvent(output, loader, string(StageMerge), input.stepID, lane.ID, lane.Merge.Module, mergeDecision)
if err := writeDebugTimed(input.Debug, path.Join("merge", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageMerge), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Merge.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": mergeDecision.Reused, "decision": mergeDecision, "source": debugSourceDocumentEnvelope(doc), "extract_outputs": debugCheckpointArtifacts(extracts.serialized), "options": redactSensitiveMap(lane.Merge.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
return err
return stageResult, err
}
var merged erasedMergeArtifact
var serializedMerge CheckpointArtifact
@@ -205,7 +215,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
if mergeDecision.Reused {
value, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, mergeCP.Output)
if decodeErr != nil {
return fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr)
return stageResult, fmt.Errorf("decode merge checkpoint for lane %q: %w", lane.ID, decodeErr)
}
merged = erasedMergeArtifact{LaneID: lane.ID, MergerKey: lane.Merge.Module, SourceID: doc.ID, Value: value}
serializedMerge = hydrated
@@ -213,7 +223,7 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
output.Warnings = append(output.Warnings, mergeWarnings...)
} else {
if err := checkpointMergeRunning(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps); err != nil {
return err
return stageResult, err
}
ok, rejection, runErr := runWithRetry(ctx, lane.Merge.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
started := time.Now().UTC()
@@ -256,58 +266,62 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
})
if runErr != nil {
_ = checkpointMergeFailed(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, runErr)
return runErr
return stageResult, runErr
}
if !ok {
output.Rejected = append(output.Rejected, *rejection)
if err := checkpointMergeRejected(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, *rejection); err != nil {
return err
return stageResult, err
}
return nil
stageResult.terminal = true
return stageResult, nil
}
output.Warnings = append(output.Warnings, mergeWarnings...)
if err := recordMerge(checkpoints, input.stepID, lane.ID, lane.Merge.Module, mergeDeps, serializedMerge, mergeWarnings); err != nil {
return err
return stageResult, err
}
}
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
return stageResult, err
}
stageResult.artifact = merged
stageResult.serialized = serializedMerge
return stageResult, nil
}
activeStage = StageNormalize
type normalizeStageResult struct {
serialized CheckpointArtifact
warnings []contracts.Warning
accepted bool
}
func (r *Runner) runNormalizeStage(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, prepared preparedLaneExecutor, merged erasedMergeArtifact, serializedMerge CheckpointArtifact, output *RunOutput) (normalizeStageResult, error) {
var stageResult normalizeStageResult
lane, typed := prepared.resolved, prepared.typed
normalizeReferences := operationReferenceSet(input, lane.NormalizeReferences)
normalizeDeps := append(artifactCheckpointDigests([]CheckpointArtifact{serializedMerge}), generatedReferenceDependencies(normalizeReferences)...)
normalizeDeps = normalizeCheckpointFingerprints(normalizeDeps)
normalizeCP, normalizeDecision := loadNormalize(loader, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps)
normalizeDecision = forceCheckpointDecision(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision)
if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision); err != nil {
return err
normalizeDecision, err := resolveCheckpointDecision(output, loader, input.CheckpointPolicy, StageNormalize, input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision, typed.codec, []CheckpointArtifact{normalizeCP.Output})
if err != nil {
return stageResult, err
}
if normalizeDecision.Reused {
if _, _, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, normalizeCP.Output); decodeErr != nil {
normalizeDecision = CheckpointDecision{Category: "executed", ReasonCode: "artifact_not_canonical", Detail: "stored normalize artifact failed canonical codec validation", Reason: "normalize artifact checkpoint is not canonical"}
}
}
if err := requireReusableCheckpoint(input.CheckpointPolicy, input.stepID, lane.ID, normalizeDecision); err != nil {
return err
}
recordCheckpointEvent(output, loader, string(StageNormalize), input.stepID, lane.ID, lane.Normalize.Module, normalizeDecision)
if err := writeDebugTimed(input.Debug, path.Join("normalize", debugPathComponent(lane.ID), "input.json"), debugTimedEnvelope{Stage: string(StageNormalize), StepID: input.stepID, LaneID: lane.ID, ModuleKey: lane.Normalize.Module, StartedAt: time.Now().UTC(), Payload: map[string]any{"reused": normalizeDecision.Reused, "decision": normalizeDecision, "source": debugSourceDocumentEnvelope(doc), "merge_output": debugCheckpointArtifact(serializedMerge), "options": redactSensitiveMap(lane.Normalize.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
return err
return stageResult, err
}
var serializedNormalize CheckpointArtifact
var normalizeWarnings []contracts.Warning
if normalizeDecision.Reused {
_, hydrated, decodeErr := decodeCanonicalCheckpointArtifact(typed.codec, normalizeCP.Output)
if decodeErr != nil {
return fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr)
return stageResult, fmt.Errorf("decode normalize checkpoint for lane %q: %w", lane.ID, decodeErr)
}
serializedNormalize = hydrated
normalizeWarnings = cloneWarnings(normalizeCP.Warnings)
output.Warnings = append(output.Warnings, normalizeWarnings...)
} else {
if err := checkpointNormalizeRunning(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps); err != nil {
return err
return stageResult, err
}
ok, rejection, runErr := runWithRetry(ctx, lane.Normalize.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
started := time.Now().UTC()
@@ -349,25 +363,27 @@ func (r *Runner) continueTypedLane(ctx context.Context, input RunInput, checkpoi
})
if runErr != nil {
_ = checkpointNormalizeFailed(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, runErr)
return runErr
return stageResult, runErr
}
if !ok {
output.Rejected = append(output.Rejected, *rejection)
if err := checkpointNormalizeRejected(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, *rejection); err != nil {
return err
return stageResult, err
}
return nil
return stageResult, nil
}
output.Warnings = append(output.Warnings, normalizeWarnings...)
if err := recordNormalize(checkpoints, input.stepID, lane.ID, lane.Normalize.Module, normalizeDeps, serializedNormalize, normalizeWarnings); err != nil {
return err
return stageResult, err
}
}
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
return stageResult, err
}
output.NormalizeOutputs = append(output.NormalizeOutputs, contracts.SerializedOutput{StepID: input.stepID, LaneID: lane.ID, NormalizerKey: lane.Normalize.Module, SourceID: doc.ID, Artifact: contracts.CloneSerializedArtifact(serializedNormalize.Artifact)})
return nil
stageResult.serialized = serializedNormalize
stageResult.warnings = normalizeWarnings
stageResult.accepted = true
return stageResult, nil
}
func setTypedLaneManifestMetadata(output *RunOutput, laneID string, extractor, merger, normalizer any) error {