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

Remove obsolete sequential extraction path

Browse Source
This commit is contained in:
Eric Rakestraw
2026-07-17 14:05:21 +00:00
parent 2df7084d5d
commit ef6926322d
4 changed files with 226 additions and 138 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -54,9 +54,8 @@ type RunInput struct {
// single worker so direct framework callers retain deterministic behavior.
ExtractWorkers int
pipeline ResolvedPipeline
llmClient contracts.StructuredLLMClient
extractDecision *CheckpointDecision
pipeline ResolvedPipeline
llmClient contracts.StructuredLLMClient
}
type RunOutput struct {

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

@@ -28,6 +28,22 @@ type laneExtractState struct {
failed bool
}
type finalizedExtractResults struct {
accepted []erasedExtractArtifact
serialized []CheckpointArtifact
warnings []contracts.Warning
rejected []contracts.RejectedOutput
decision CheckpointDecision
}
func loadExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
return loader.Extract(laneID, moduleKey, deps)
}
func recordExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings)
}
type extractJob struct {
lane *laneExtractState
chunk source.Chunk
@@ -56,19 +72,6 @@ type orderedRunError struct {
err error
}
type completedExtractLoader struct {
CheckpointLoader
laneID string
checkpoint ExtractCheckpoint
}
func (l completedExtractLoader) Extract(laneID, _ string, _ []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
if laneID == l.laneID {
return l.checkpoint, CheckpointDecision{Reused: true, Reason: "coordinated extract result"}
}
return ExtractCheckpoint{}, CheckpointDecision{Reason: "extract result unavailable"}
}
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) {
output := RunOutput{Manifest: manifestFromPipeline(input)}
states := make([]*laneExtractState, len(input.Prepared.lanes))
@@ -85,6 +88,8 @@ func (r *Runner) runLanes(parent context.Context, input RunInput, checkpoints Ch
if err := checkpoints.ExtractRunning(prepared.resolved.ID, prepared.resolved.Extract.Module, state.deps); err != nil {
return output, fmt.Errorf("write extract checkpoint for lane %q: %w", prepared.resolved.ID, err)
}
} else if err := finalizeLaneExtract(checkpoints, state); err != nil {
return output, err
}
states[i] = state
}
@@ -325,10 +330,26 @@ func finalizeLaneExtract(checkpoints CheckpointRecorder, state *laneExtractState
func (r *Runner) continueLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, state *laneExtractState) (RunOutput, error) {
lane := state.prepared.resolved
local := RunOutput{Manifest: manifestFromPipeline(input)}
checkpoint := ExtractCheckpoint{Outputs: state.serialized, Rejected: state.rejected, Warnings: state.warnings}
coordinatedLoader := completedExtractLoader{CheckpointLoader: loader, laneID: lane.ID, checkpoint: checkpoint}
input.extractDecision = &state.decision
err := r.runTypedLane(ctx, input, checkpoints, coordinatedLoader, doc, sourceInput, sessionID, chunks, state.prepared, &local)
results := finalizedExtractResults{
accepted: state.values,
serialized: state.serialized,
warnings: state.warnings,
rejected: state.rejected,
decision: state.decision,
}
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 {
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 {
return local, &laneRunError{stage: StageExtract, err: err}
}
if len(results.accepted) == 0 {
return local, nil
}
err := r.continueTypedLane(ctx, input, checkpoints, loader, doc, sourceInput, sessionID, state.prepared, results, &local)
return local, err
}

180
internal/framework/pipeline/runner_extract_handoff_test.go Normal file
View File

@@ -0,0 +1,180 @@
package pipeline
import (
"context"
"reflect"
"strings"
"testing"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
type extractCaptureRecorder struct {
CheckpointRecorder
checkpoint ExtractCheckpoint
}
func (r *extractCaptureRecorder) ExtractSucceeded(_ string, _ string, _ []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
r.checkpoint = ExtractCheckpoint{
Outputs: cloneCheckpointArtifacts(outputs),
Rejected: cloneRejectedOutputs(rejected),
Warnings: cloneWarnings(warnings),
}
return nil
}
type extractResultLoader struct {
CheckpointLoader
checkpoint ExtractCheckpoint
decision CheckpointDecision
}
func (l *extractResultLoader) Enabled() bool { return true }
func (l *extractResultLoader) Extract(string, string, []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
return ExtractCheckpoint{
Outputs: cloneCheckpointArtifacts(l.checkpoint.Outputs),
Rejected: cloneRejectedOutputs(l.checkpoint.Rejected),
Warnings: cloneWarnings(l.checkpoint.Warnings),
}, l.decision
}
func cloneCheckpointArtifacts(values []CheckpointArtifact) []CheckpointArtifact {
if len(values) == 0 {
return nil
}
cloned := make([]CheckpointArtifact, len(values))
for i := range values {
cloned[i] = cloneCheckpointArtifact(values[i])
}
return cloned
}
func TestRunnerContinuesFromFreshAndReusedExtractResults(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
extractCalls := 0
installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
extractCalls++
return erasedTypedResult{
Value: typedValueForLane(0, request.Chunk.Index),
Warnings: []contracts.Warning{{Scope: "extract", ReasonCode: "observed", Message: "accepted extract"}},
}, nil
})
freshDebug := newCapturedDebugRecorder()
recorder := &extractCaptureRecorder{CheckpointRecorder: NoopCheckpointRecorder()}
freshLoader := &extractResultLoader{
CheckpointLoader: NoopCheckpointLoader(),
decision: CheckpointDecision{Reason: "extract checkpoint not found"},
}
fresh, err := New().Run(context.Background(), RunInput{
Prepared: prepared,
RawInput: []byte("input"),
Checkpoints: recorder,
Checkpoint: freshLoader,
Debug: freshDebug,
})
if err != nil {
t.Fatalf("fresh Run() error = %v, want nil", err)
}
if extractCalls != 1 {
t.Fatalf("fresh extract calls = %d, want 1", extractCalls)
}
assertExtractDecision(t, fresh.CheckpointEvents, "executed", "extract checkpoint not found")
assertExtractDebugPaths(t, freshDebug, true)
reusedDebug := newCapturedDebugRecorder()
reusedLoader := &extractResultLoader{
CheckpointLoader: NoopCheckpointLoader(),
checkpoint: recorder.checkpoint,
decision: CheckpointDecision{Reused: true, Reason: "extract checkpoint matched"},
}
reused, err := New().Run(context.Background(), RunInput{
Prepared: prepared,
RawInput: []byte("input"),
Checkpoint: reusedLoader,
Debug: reusedDebug,
})
if err != nil {
t.Fatalf("reused Run() error = %v, want nil", err)
}
if extractCalls != 1 {
t.Fatalf("extract calls after reuse = %d, want 1", extractCalls)
}
assertExtractDecision(t, reused.CheckpointEvents, "reused", "extract checkpoint matched")
assertExtractDebugPaths(t, reusedDebug, false)
if !reflect.DeepEqual(reused.NormalizeOutputs, fresh.NormalizeOutputs) {
t.Fatalf("reused normalize outputs = %#v, want fresh outputs %#v", reused.NormalizeOutputs, fresh.NormalizeOutputs)
}
if !reflect.DeepEqual(reused.Warnings, fresh.Warnings) {
t.Fatalf("reused warnings = %#v, want fresh warnings %#v", reused.Warnings, fresh.Warnings)
}
}
func TestRunnerPromotesOnlyAcceptedExtractRetryWarnings(t *testing.T) {
prepared := preparedAttemptDebugPipeline(t)
prepared.lanes[0].resolved.Extract.Retries = 1
attempts := 0
installExtractOperation(prepared, 0, func(_ context.Context, request contracts.TypedExtractionRequest) (erasedTypedResult, error) {
attempts++
scope := "discarded"
if attempts == 2 {
scope = "accepted"
}
return erasedTypedResult{
Value: typedValueForLane(0, request.Chunk.Index),
Warnings: []contracts.Warning{{Scope: scope, ReasonCode: "observed", Message: scope}},
}, nil
})
validatorCalls := 0
prepared.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
}
debug := newCapturedDebugRecorder()
output, err := New().Run(context.Background(), RunInput{Prepared: prepared, RawInput: []byte("input"), Debug: debug})
if err != nil {
t.Fatalf("Run() error = %v, want nil", err)
}
if attempts != 2 {
t.Fatalf("extract attempts = %d, want 2", attempts)
}
if len(output.Warnings) != 1 || output.Warnings[0].Scope != "accepted" {
t.Fatalf("promoted warnings = %#v, want accepted attempt only", output.Warnings)
}
name := "extract/notes/chunk-000001/attempt-02.json"
if !debug.has(name) {
t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names())
}
}
func assertExtractDecision(t *testing.T, events []CheckpointEvent, action string, reason string) {
t.Helper()
for _, event := range events {
if event.Stage == string(StageExtract) {
if event.Action != action || event.Reason != reason {
t.Fatalf("extract checkpoint event = %#v, want action %q reason %q", event, action, reason)
}
return
}
}
t.Fatalf("extract checkpoint event missing from %#v", events)
}
func assertExtractDebugPaths(t *testing.T, debug *capturedDebugRecorder, wantAttempt bool) {
t.Helper()
for _, name := range []string{"extract/notes/input.json", "extract/notes/output.json"} {
if !debug.has(name) {
t.Fatalf("debug artifact %q is missing; names = %#v", name, debug.names())
}
}
attemptPath := "extract/notes/chunk-000001/attempt-01.json"
if debug.has(attemptPath) != wantAttempt {
t.Fatalf("attempt debug path present = %t, want %t; names = %#v", debug.has(attemptPath), wantAttempt, debug.names())
}
input := debug.envelope(t, "extract/notes/input.json")
wantReuse := !wantAttempt
if !strings.Contains(string(debug.json["extract/notes/input.json"]), `"reused":`+map[bool]string{true: "true", false: "false"}[wantReuse]) {
t.Fatalf("extract input payload = %#v, want reused %t", input.Payload, wantReuse)
}
}

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

@@ -6,25 +6,18 @@ import (
"encoding/hex"
"fmt"
"path"
"sort"
"time"
"gitea.maximumdirect.net/eric/notarius/internal/core/source"
"gitea.maximumdirect.net/eric/notarius/internal/framework/contracts"
)
func loadExtract(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (ExtractCheckpoint, CheckpointDecision) {
return loader.Extract(laneID, moduleKey, deps)
}
func loadMerge(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (MergeCheckpoint, CheckpointDecision) {
return loader.Merge(laneID, moduleKey, deps)
}
func loadNormalize(loader CheckpointLoader, laneID, moduleKey string, deps []CheckpointFingerprint) (NormalizeCheckpoint, CheckpointDecision) {
return loader.Normalize(laneID, moduleKey, deps)
}
func recordExtract(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, outputs []CheckpointArtifact, rejected []contracts.RejectedOutput, warnings []contracts.Warning) error {
return recorder.ExtractSucceeded(laneID, moduleKey, deps, outputs, rejected, warnings)
}
func recordMerge(recorder CheckpointRecorder, laneID, moduleKey string, deps []CheckpointFingerprint, output CheckpointArtifact, warnings []contracts.Warning) error {
return recorder.MergeSucceeded(laneID, moduleKey, deps, output, warnings)
}
@@ -126,8 +119,8 @@ type laneRunError struct {
func (e *laneRunError) Error() string { return e.err.Error() }
func (e *laneRunError) Unwrap() error { return e.err }
func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints CheckpointRecorder, loader CheckpointLoader, doc *source.SourceDocument, sourceInput contracts.LLMInputMaterial, sessionID string, chunks []source.Chunk, prepared preparedLaneExecutor, output *RunOutput) (err error) {
activeStage := StageExtract
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}
@@ -139,116 +132,11 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C
}
setTypedLaneManifestMetadata(output, lane.ID, typed.extractor, typed.merger, typed.normalizer)
values := make([]erasedExtractArtifact, 0, len(chunks))
serializedExtracts := make([]CheckpointArtifact, 0, len(chunks))
extractWarnings := []contracts.Warning{}
rejectedStart := len(output.Rejected)
chunksDigest, err := joinedChunkDigest(chunks)
if err != nil {
return fmt.Errorf("digest chunks for lane %q: %w", lane.ID, err)
}
extractDeps := digestFingerprints("chunks", chunksDigest)
cp, decision := loadExtract(loader, lane.ID, lane.Extract.Module, extractDeps)
if decision.Reused {
for _, stored := range cp.Outputs {
if _, decodeErr := decodeCheckpointArtifact(typed.codec, stored); decodeErr != nil {
decision = CheckpointDecision{Reason: "extract artifact checkpoint codec is incompatible: " + decodeErr.Error()}
break
}
}
}
reportedDecision := decision
if input.extractDecision != nil {
reportedDecision = *input.extractDecision
}
recordCheckpointEvent(output, loader, string(StageExtract), lane.ID, lane.Extract.Module, reportedDecision)
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": reportedDecision.Reused, "decision": reportedDecision, "source": debugSourceDocumentEnvelope(doc), "chunks": debugSourceChunkEnvelopes(chunks), "options": redactSensitiveMap(lane.Extract.Options), "metadata": redactSensitiveMap(input.Metadata)}}); err != nil {
return err
}
if decision.Reused {
for _, stored := range cp.Outputs {
value, decodeErr := decodeCheckpointArtifact(typed.codec, stored)
if decodeErr != nil {
return fmt.Errorf("decode extract checkpoint for lane %q: %w", lane.ID, decodeErr)
}
stored = hydrateCheckpointArtifact(typed.codec, stored, value)
artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: stored.ChunkID, ChunkIndex: stored.ChunkIndex, ChunkRef: stored.ChunkRef, Value: value}
if stored.ChunkIndex >= 0 && stored.ChunkIndex < len(chunks) && artifact.ChunkRef == (source.SourceRef{}) {
artifact.ChunkRef = chunks[stored.ChunkIndex].Ref
}
values = append(values, artifact)
serializedExtracts = append(serializedExtracts, cloneCheckpointArtifact(stored))
}
extractWarnings = cloneWarnings(cp.Warnings)
output.Warnings = append(output.Warnings, extractWarnings...)
output.Rejected = append(output.Rejected, cloneRejectedOutputs(cp.Rejected)...)
} else {
if err := checkpoints.ExtractRunning(lane.ID, lane.Extract.Module, extractDeps); err != nil {
return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err)
}
for i := range chunks {
chunk := chunks[i]
var accepted erasedExtractArtifact
var serializedAccepted CheckpointArtifact
var acceptedWarnings []contracts.Warning
ok, rejection, runErr := runWithRetry(ctx, lane.Extract.Retries, func(attempt int) (bool, *contracts.RejectedOutput, error) {
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)
result, callErr := typed.extract(attemptCtx, typed.extractor, contracts.TypedExtractionRequest{Source: doc, Chunk: &chunk, SourceInput: chunkInputMaterial(sourceInput, chunk), SessionID: sessionID, References: CloneReferenceSet(lane.ExtractReferences.ReferenceSet), LLMProfile: lane.Extract.LLMProfile, Metadata: cloneMetadata(input.Metadata)})
if callErr != nil {
_ = writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Error: callErr.Error()}, llmScope)
return false, nil, fmt.Errorf("extract lane %q chunk %q with extractor %q: %w", lane.ID, chunk.ID, lane.Extract.Module, callErr)
}
artifact := erasedExtractArtifact{LaneID: lane.ID, ExtractorKey: lane.Extract.Module, SourceID: doc.ID, ChunkID: chunk.ID, ChunkIndex: chunk.Index, ChunkRef: chunk.Ref, Value: result.Value}
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: result.Value}, prepared.extractValidators, attempt, input.Debug)
if validateErr != nil || rejected != nil {
return false, rejected, validateErr
}
stored, encodeErr := checkpointArtifact(typed.codec, artifact.LaneID, artifact.ExtractorKey, artifact.SourceID, artifact.Value)
if encodeErr != nil {
return false, nil, encodeErr
}
stored.ChunkID, stored.ChunkIndex, stored.ChunkRef = artifact.ChunkID, artifact.ChunkIndex, artifact.ChunkRef
accepted, serializedAccepted = artifact, stored
acceptedWarnings = append(cloneWarnings(result.Warnings), warnings...)
if debugErr := writeDebugAttempt(input.Debug, attemptPath, debugTimedEnvelope{Stage: string(StageExtract), LaneID: lane.ID, ModuleKey: lane.Extract.Module, Attempt: attempt, StartedAt: started, Payload: map[string]any{"output": debugCheckpointArtifact(stored), "warnings": debugWarningEnvelopes(acceptedWarnings)}}, llmScope); debugErr != nil {
return false, nil, debugErr
}
return true, nil, nil
})
if runErr != nil {
_ = checkpoints.ExtractFailed(lane.ID, lane.Extract.Module, extractDeps, runErr)
return runErr
}
if !ok {
output.Rejected = append(output.Rejected, *rejection)
continue
}
values = append(values, accepted)
serializedExtracts = append(serializedExtracts, serializedAccepted)
extractWarnings = append(extractWarnings, acceptedWarnings...)
output.Warnings = append(output.Warnings, acceptedWarnings...)
}
if err := recordExtract(checkpoints, lane.ID, lane.Extract.Module, extractDeps, serializedExtracts, cloneRejectedOutputs(output.Rejected[rejectedStart:]), extractWarnings); err != nil {
return fmt.Errorf("write extract checkpoint for lane %q: %w", lane.ID, err)
}
}
sort.SliceStable(values, func(i, j int) bool { return values[i].ChunkIndex < values[j].ChunkIndex })
sort.SliceStable(serializedExtracts, func(i, j int) bool { return serializedExtracts[i].ChunkIndex < serializedExtracts[j].ChunkIndex })
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": reportedDecision.Reused, "outputs": debugCheckpointArtifacts(serializedExtracts), "rejected": debugRejectedOutputEnvelopes(output.Rejected[rejectedStart:]), "warnings": debugWarningEnvelopes(extractWarnings)}}); err != nil {
return err
}
if len(values) == 0 {
return nil
}
activeStage = StageMerge
mergeInputs := make([]contracts.ExtractArtifact[any], len(values))
for i, value := range values {
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}
}
mergeDeps := artifactCheckpointDigests(serializedExtracts)
mergeDeps := artifactCheckpointDigests(extracts.serialized)
mergeCP, mergeDecision := loadMerge(loader, lane.ID, lane.Merge.Module, mergeDeps)
if mergeDecision.Reused {
if _, decodeErr := decodeCheckpointArtifact(typed.codec, mergeCP.Output); decodeErr != nil {
@@ -256,7 +144,7 @@ func (r *Runner) runTypedLane(ctx context.Context, input RunInput, checkpoints C
}
}
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(serializedExtracts), "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), 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