Keep transcript sections fixed throughout processing
This commit is contained in:
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user