diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 06c6d5a..324a1b7 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -1,6 +1,6 @@ # Implementation Plan: Shared Stage Checkpoint Mechanics -Status: Implementation in progress; Stages 1-2 complete. +Status: Implementation in progress; Stages 1-3 complete. ## Scope and Authority @@ -229,7 +229,17 @@ selection, and failure behavior remain unchanged. ## Stage 3: Migrate Transcribe and Remove Duplicate Machinery -Status: Pending. +Status: Complete. + +Completion marker (2026-09-22): Delegated transcription's locked checkpoint +operation to the shared helper while retaining typed validation, direct-call +record initialization, stage timestamp updates, post-save working collection +refresh, and the existing caller-owned mutex and failure latch. Removed the +duplicate production manifest clone; the read-only resume assertion now uses +the shared bounded snapshot. Existing real-store, concurrency, force/retry, +interruption, aggregate-repair, and zero-sized-audio coverage passed unchanged. +Verified with `go test ./internal/stage ./internal/app ./internal/manifest`, +`go test -race ./internal/stage ./internal/app`, and `git diff --check`. Preparation: `internal/stage/transcribe_execution_state.go`, `transcribe.go`, `transcribe_resume.go`, `transcribe_checkpoint_test.go`, diff --git a/internal/stage/transcribe_execution_state.go b/internal/stage/transcribe_execution_state.go index 640035c..702aa44 100644 --- a/internal/stage/transcribe_execution_state.go +++ b/internal/stage/transcribe_execution_state.go @@ -156,39 +156,30 @@ func (state *transcribeExecutionState) checkpointCollectionLocked( return fmt.Errorf("validate transcribe speaker collection: %w", err) } - candidate := cloneManifestForTranscribeCheckpoint(state.manifest) - if candidate.Stages == nil { - candidate.Stages = make(map[string]*manifest.StageRecord) - } - record := candidate.Stages["transcribe"] - if record == nil { - now := time.Now().UTC() - record = &manifest.StageRecord{ - Name: "transcribe", Status: manifest.StatusRunning, CreatedAt: now, UpdatedAt: now, - } - candidate.Stages["transcribe"] = record - } - record.TranscribeStateVersion = manifest.TranscribeStateContractVersion - record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(records) - record.UpdatedAt = time.Now().UTC() - if err := state.env.ManifestStore.Save(ctx, state.paths.ManifestPath, candidate); err != nil { - return fmt.Errorf("save session manifest %q: %w", state.paths.ManifestPath, err) + missingRecord := transcribeStageRecord(state.manifest) == nil + if err := checkpointStageCollection( + ctx, + state.env.ManifestStore, + state.paths.ManifestPath, + state.manifest, + "transcribe", + func(record *manifest.StageRecord) error { + if missingRecord { + now := time.Now().UTC() + *record = manifest.StageRecord{ + Name: "transcribe", Status: manifest.StatusRunning, CreatedAt: now, UpdatedAt: now, + } + } + record.TranscribeStateVersion = manifest.TranscribeStateContractVersion + record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(records) + record.UpdatedAt = time.Now().UTC() + return nil + }, + ); err != nil { + return err } live := transcribeStageRecord(state.manifest) - if live == nil { - live = &manifest.StageRecord{} - if state.manifest.Stages == nil { - state.manifest.Stages = make(map[string]*manifest.StageRecord) - } - state.manifest.Stages["transcribe"] = live - } - live.TranscribeStateVersion = candidate.Stages["transcribe"].TranscribeStateVersion - live.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection( - candidate.Stages["transcribe"].TranscribeSpeakers, - ) - live.UpdatedAt = candidate.Stages["transcribe"].UpdatedAt - state.manifest.UpdatedAt = candidate.UpdatedAt state.records = manifest.CloneTranscribeSpeakerCollection(live.TranscribeSpeakers) return nil } @@ -259,27 +250,6 @@ func transcribeStageRecord(m *manifest.Manifest) *manifest.StageRecord { return m.Stages["transcribe"] } -func cloneManifestForTranscribeCheckpoint(source *manifest.Manifest) *manifest.Manifest { - clone := *source - clone.Inputs = append([]manifest.InputRecord(nil), source.Inputs...) - clone.Artifacts = append([]manifest.ArtifactRecord(nil), source.Artifacts...) - clone.Stages = make(map[string]*manifest.StageRecord, len(source.Stages)) - for name, sourceRecord := range source.Stages { - if sourceRecord == nil { - clone.Stages[name] = nil - continue - } - record := *sourceRecord - record.Outputs = append([]manifest.ArtifactRecord(nil), sourceRecord.Outputs...) - record.Logs = append([]string(nil), sourceRecord.Logs...) - record.GeneratedConfigs = append([]string(nil), sourceRecord.GeneratedConfigs...) - record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(sourceRecord.TranscribeSpeakers) - record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(sourceRecord.AnalyzeArtifacts) - clone.Stages[name] = &record - } - return &clone -} - func cloneStringAnyMap(source map[string]any) map[string]any { if source == nil { return nil diff --git a/internal/stage/transcribe_reconciliation_test.go b/internal/stage/transcribe_reconciliation_test.go index 98d42be..56da6f1 100644 --- a/internal/stage/transcribe_reconciliation_test.go +++ b/internal/stage/transcribe_reconciliation_test.go @@ -58,7 +58,7 @@ func TestTranscribeResumeAcceptsEquivalentOutputsAndInspectsRemovalReadOnly(t *t t.Fatalf("equivalent aggregate: %#v, %v", validation, err) } m.Inputs = slices.DeleteFunc(m.Inputs, func(in manifest.InputRecord) bool { return filepath.Base(in.Path) == "bob.flac" }) - before := cloneManifestForTranscribeCheckpoint(m) + before := cloneManifestForCheckpoint(m) validation, err = (transcribeStage{}).ValidateResume(context.Background(), env, m) if err != nil || validation.Resumable { t.Fatalf("removed speaker: %#v, %v", validation, err)