Delegate transcription checkpoints to shared persistence
This commit is contained in:
@@ -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`,
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)
|
||||
|
||||
Reference in New Issue
Block a user