From bb21a818050a57806a9971f0d270de6930b28cf9 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Thu, 24 Sep 2026 17:17:14 +0000 Subject: [PATCH] Keep transcript sections fixed throughout processing --- docs/roadmap/implementation.md | 9 ++ internal/framework/runner/runner.go | 112 ++++++++++++++++------- internal/framework/runner/runner_test.go | 68 ++++++++++++++ 3 files changed, 157 insertions(+), 32 deletions(-) diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 0601292..8b4adf8 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -525,6 +525,15 @@ required validation. ## Stage 5 — Fix Chunk Boundaries for the Entire Run +**Completed 2026-09-24.** The runner now partitions once before the module loop, +returns the initial section inventory, applies approved corrections only inside +their identified sections, carries current text forward, and recombines the +transcript after processing. Tests cover substantial text growth, stable segment +membership, empty work, and impossible chunk constraints. `go test +./internal/core/chunking` and `git diff --check` passed. Runner tests remain +unavailable because warning-field consumers still fail compilation; Stage 6 +replaces them with operation outcomes. + **Work:** Move the runner's chunking call before its stage loop. Own fixed chunk states and initial inventory in runner output. Execute each module using the current text of those same chunks, and apply fully approved candidates only within their diff --git a/internal/framework/runner/runner.go b/internal/framework/runner/runner.go index 50a2cc4..76b00bf 100644 --- a/internal/framework/runner/runner.go +++ b/internal/framework/runner/runner.go @@ -131,8 +131,32 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { working := cloneTranscript(input.Transcript) results := make([]ModuleResult, 0, len(input.ModuleSpecs)) + sections, err := chunkWorkingTranscript(input.Config, working) + if err != nil { + return RunOutput{FinalTranscript: working}, fmt.Errorf("chunk normalized transcript: %w", err) + } + inventory := make([]contracts.SectionMetadata, len(sections)) + workUnits := make([]WorkUnitOutcome, 0, len(input.ModuleSpecs)*len(sections)) + for i, section := range sections { + inventory[i] = contracts.SectionMetadataFromSection(section) + } + for moduleIndex, spec := range input.ModuleSpecs { + for _, section := range inventory { + workUnits = append(workUnits, WorkUnitOutcome{ + ModuleIndex: moduleIndex, ModuleKey: spec.ModuleKey, + ModuleInstance: spec.InstanceName, Section: section, + State: outcomes.WorkNotStarted, + }) + } + } + snapshot := func() RunOutput { + return RunOutput{ + FinalTranscript: transcriptFromSections(sections), ModuleResults: results, + Sections: inventory, WorkUnits: workUnits, + } + } - for _, spec := range input.ModuleSpecs { + for moduleIndex, spec := range input.ModuleSpecs { startedAt := time.Now().UTC() module, err := r.factory.ModuleForSpec(spec) if err != nil { @@ -145,31 +169,10 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { CompletedAt: time.Now().UTC(), } results = append(results, failed) - return RunOutput{ - FinalTranscript: working, - ModuleResults: results, - }, fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err) + return snapshot(), fmt.Errorf("module %q setup failed: %w", spec.InstanceName, err) } policy := module.ReplacementPolicy() - sections, err := chunkWorkingTranscript(input.Config, working) - if err != nil { - failed := ModuleResult{ - ModuleKey: spec.ModuleKey, - ModuleInstance: spec.InstanceName, - ReplacementPolicy: policy, - Status: ModuleStatusFailed, - ErrorMessage: err.Error(), - StartedAt: startedAt, - CompletedAt: time.Now().UTC(), - } - results = append(results, failed) - return RunOutput{ - FinalTranscript: working, - ModuleResults: results, - }, fmt.Errorf("module %q chunking failed: %w", spec.InstanceName, err) - } - pipelineResult, pipelineErr := runModulePipeline(ctx, collectSectionProposalsInput{ Module: module, Spec: spec, @@ -198,14 +201,19 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { CompletedAt: time.Now().UTC(), } results = append(results, failed) - return RunOutput{ - FinalTranscript: working, - ModuleResults: results, - }, fmt.Errorf("module %q failed: %w", spec.InstanceName, pipelineErr) + return snapshot(), fmt.Errorf("module %q failed: %w", spec.InstanceName, pipelineErr) } - applyResult := proposals.ApplyProposals(working, pipelineResult.Approved, policy) - working = applyResult.Transcript + updated, applied, skipped, appliedBySection, applyErr := applyWithinSections(sections, pipelineResult.Approved, policy) + if applyErr != nil { + return snapshot(), fmt.Errorf("module %q application failed: %w", spec.InstanceName, applyErr) + } + sections = updated + for sectionPos := range sections { + unit := &workUnits[moduleIndex*len(sections)+sectionPos] + unit.State = outcomes.WorkCompleted + unit.CorrectionsApplied = appliedBySection[sectionPos] + } results = append(results, ModuleResult{ ModuleKey: spec.ModuleKey, @@ -216,17 +224,57 @@ func (r *Runner) run(ctx context.Context, input RunInput) (RunOutput, error) { Warnings: pipelineResult.Warnings, ValidatorDecisions: pipelineResult.ValidatorDecisions, ValidatorRejected: pipelineResult.ValidatorRejected, - AppliedChanges: applyResult.Applied, - SkippedChanges: applyResult.Skipped, + AppliedChanges: applied, + SkippedChanges: skipped, StartedAt: startedAt, CompletedAt: time.Now().UTC(), }) } - output := RunOutput{FinalTranscript: working, ModuleResults: results} + output := snapshot() return output, nil } +func transcriptFromSections(sections []chunking.Section) *schema.Transcript { + segments := make([]schema.Segment, 0) + for _, section := range sections { + segments = append(segments, section.Segments...) + } + return &schema.Transcript{Segments: segments} +} + +// applyWithinSections leaves section membership and initial metadata unchanged. +// Approved proposals are applied only to their identified section. +func applyWithinSections(sections []chunking.Section, approved []proposals.EnrichedCorrectionProposal, policy proposals.ReplacementPolicy) ([]chunking.Section, []proposals.AppliedChange, []proposals.SkippedChange, []int, error) { + bySection := make(map[int][]proposals.EnrichedCorrectionProposal, len(sections)) + known := make(map[int]struct{}, len(sections)) + for _, section := range sections { + known[section.Index] = struct{}{} + } + for _, candidate := range approved { + if candidate.SectionIndex == nil { + return nil, nil, nil, nil, fmt.Errorf("approved proposal %d has no section identity", candidate.ProposalIndex) + } + if _, ok := known[*candidate.SectionIndex]; !ok { + return nil, nil, nil, nil, fmt.Errorf("approved proposal %d has unknown section identity", candidate.ProposalIndex) + } + bySection[*candidate.SectionIndex] = append(bySection[*candidate.SectionIndex], candidate) + } + updated := make([]chunking.Section, len(sections)) + copy(updated, sections) + applied := make([]proposals.AppliedChange, 0) + skipped := make([]proposals.SkippedChange, 0) + appliedBySection := make([]int, len(sections)) + for i, section := range sections { + result := proposals.ApplyProposals(transcriptFromSection(section), bySection[section.Index], policy) + updated[i].Segments = result.Transcript.Segments + applied = append(applied, result.Applied...) + skipped = append(skipped, result.Skipped...) + appliedBySection[i] = len(result.Applied) + } + return updated, applied, skipped, appliedBySection, nil +} + type collectSectionProposalsInput struct { Module contracts.TranscriptModule Spec contracts.ModuleRunSpec diff --git a/internal/framework/runner/runner_test.go b/internal/framework/runner/runner_test.go index fac330c..1d0ce2a 100644 --- a/internal/framework/runner/runner_test.go +++ b/internal/framework/runner/runner_test.go @@ -162,6 +162,74 @@ func TestRunnerModulesRunSequentially(t *testing.T) { } } +func TestRunnerKeepsSectionMembershipAfterTextGrowth(t *testing.T) { + cfg := runnerTestConfig() + cfg.Chunking.MaxSectionTokens = 3 + cfg.Chunking.MinSectionTokens = 0 + transcript := &schema.Transcript{Segments: []schema.Segment{ + {ID: 1, Text: "a"}, {ID: 2, Text: "b"}, {ID: 3, Text: "c"}, {ID: 4, Text: "d"}, + }} + var mu sync.Mutex + seen := make(map[int][]int) + updatedText := "a one two three four five" + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{ + "grow": fakeModule{key: "grow", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + if req.Section.Index != 0 { + return nil, nil + } + return []proposals.CorrectionProposal{{TargetSegmentID: 1, OriginalText: "a", CorrectedText: updatedText, Confidence: 1}}, nil + }}, + "inspect": fakeModule{key: "inspect", policy: proposals.ReplacementPolicyRequireUnique, proposeF: func(req contracts.ProposalRequest) ([]proposals.CorrectionProposal, error) { + ids := make([]int, len(req.WorkingTranscript.Segments)) + for i, segment := range req.WorkingTranscript.Segments { + ids[i] = segment.ID + } + mu.Lock() + seen[req.Section.Index] = ids + mu.Unlock() + if req.Section.Index == 0 && req.WorkingTranscript.Segments[0].Text != updatedText { + return nil, errors.New("later module did not see updated section text") + } + return nil, nil + }}, + }}) + out, err := r.Run(context.Background(), RunInput{Config: &cfg, Transcript: transcript, ModuleSpecs: []contracts.ModuleRunSpec{{ModuleKey: "grow", InstanceName: "grow"}, {ModuleKey: "inspect", InstanceName: "inspect"}}}) + if err != nil { + t.Fatal(err) + } + if len(out.Sections) != 2 || out.Sections[0].StartSegmentID != 1 || out.Sections[0].EndSegmentID != 2 || out.Sections[0].EstimatedTokens != 2 || len(out.WorkUnits) != 4 { + t.Fatalf("initial inventory changed: sections=%+v units=%+v", out.Sections, out.WorkUnits) + } + if len(seen) != 2 || len(seen[0]) != 2 || seen[0][0] != 1 || seen[0][1] != 2 || len(seen[1]) != 2 || seen[1][0] != 3 || seen[1][1] != 4 { + t.Fatalf("later module saw changed section membership: %+v", seen) + } + if len(out.FinalTranscript.Segments) != 4 || out.FinalTranscript.Segments[0].Text != updatedText || out.FinalTranscript.Segments[3].ID != 4 { + t.Fatalf("recombined transcript: %+v", out.FinalTranscript) + } + if out.WorkUnits[0].CorrectionsApplied != 1 || out.WorkUnits[1].CorrectionsApplied != 0 { + t.Fatalf("section application counts: %+v", out.WorkUnits) + } +} + +func TestRunnerInventoriesEmptyWorkAndRejectsImpossibleChunking(t *testing.T) { + r := New(fakeFactory{modules: map[string]contracts.TranscriptModule{}}) + empty, err := r.Run(context.Background(), RunInput{Transcript: &schema.Transcript{}}) + if err != nil || len(empty.Sections) != 0 || len(empty.WorkUnits) != 0 || len(empty.FinalTranscript.Segments) != 0 { + t.Fatalf("empty transcript and pipeline: %+v, %v", empty, err) + } + input := &schema.Transcript{Segments: []schema.Segment{{ID: 1, Text: "one"}, {ID: 2, Text: "two"}}} + noModules, err := r.Run(context.Background(), RunInput{Transcript: input}) + if err != nil || len(noModules.Sections) != 1 || len(noModules.WorkUnits) != 0 || len(noModules.FinalTranscript.Segments) != 2 { + t.Fatalf("empty pipeline inventory: %+v, %v", noModules, err) + } + cfg := runnerTestConfig() + target := 3 + cfg.Chunking.TargetSections = &target + if _, err := r.Run(context.Background(), RunInput{Config: &cfg, Transcript: input}); err == nil { + t.Fatal("impossible section target was accepted") + } +} + func TestRunnerProposalsExecutePerChunkSection(t *testing.T) { cfg := runnerTestConfig() cfg.Chunking.MaxSectionTokens = 3