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

Repair stale transcription outputs when resuming

Browse Source
This commit is contained in:
Eric Rakestraw
2026-09-22 00:50:51 +00:00
parent 96b4e73f53
commit 54d335c820
6 changed files with 245 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

83
internal/app/transcribe_checkpoint_recovery_test.go
View File

@@ -6,9 +6,11 @@ import (
"os"
"path/filepath"
"slices"
"strings"
"testing"
"time"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/seriatim"
"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/config"
@@ -109,6 +111,87 @@ func TestExecuteStagesRebuildsTranscribeAggregateAfterFinalSaveFailure(t *testin
}
}
func TestExecuteStagesRepairsSucceededTranscribeAggregateWithoutRequests(t *testing.T) {
for _, test := range []struct {
name string
mutate func(*manifest.Manifest)
removed bool
}{
{name: "removed speaker", removed: true, mutate: func(m *manifest.Manifest) {
m.Inputs = slices.DeleteFunc(m.Inputs, func(in manifest.InputRecord) bool { return filepath.Base(in.Path) == "bob.flac" })
}},
{name: "missing output", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs = m.Stages["transcribe"].Outputs[:1] }},
{name: "duplicate output", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[1] = m.Stages["transcribe"].Outputs[0] }},
{name: "wrong path", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].LocalPath = "transcripts/raw/other.json" }},
{name: "wrong checksum", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].Checksum = "obsolete" }},
{name: "wrong producer", mutate: func(m *manifest.Manifest) { m.Stages["transcribe"].Outputs[0].ProducerRunID = "other-run" }},
} {
t.Run(test.name, func(t *testing.T) {
cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
transcribe := mustSelectStage(t, "transcribe")
if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: &whisperx.FakeClient{}},
}); err != nil {
t.Fatal(err)
}
m := loadTranscribeRecoveryManifest(t, store, cfg)
test.mutate(m)
m.MarkStageSucceeded("merge", time.Now().UTC(), nil)
if err := store.Save(context.Background(), manifestPathFor(cfg), m); err != nil {
t.Fatal(err)
}
retry := &whisperx.FakeClient{}
summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
Env: &Env{ManifestStore: store, WhisperX: retry},
})
if err != nil {
t.Fatal(err)
}
if len(summary.Skipped) != 0 || len(retry.RequestsSnapshot()) != 0 {
t.Fatalf("want aggregate repair without HTTP; summary=%#v requests=%#v", summary, retry.RequestsSnapshot())
}
completed := loadTranscribeRecoveryManifest(t, store, cfg)
want := []string{"alice", "bob"}
if test.removed {
want = []string{"alice"}
}
record := completed.Stages["transcribe"]
if !slices.Equal(transcribeSpeakerKeys(completed), want) || len(record.Outputs) != len(want) {
t.Fatalf("unexpected repaired state: %#v", record)
}
for _, output := range record.Outputs {
speaker := record.TranscribeSpeakers[strings.TrimSuffix(filepath.Base(output.LocalPath), ".json")]
if output.Checksum != speaker.Output.Checksum || output.ProducerRunID != speaker.ProducerRunID {
t.Fatalf("aggregate evidence does not match checkpoint: %#v", output)
}
}
if completed.Stages["merge"].Status != manifest.StatusStale {
t.Fatal("aggregate repair did not stale merge")
}
if test.removed {
paths := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root).SessionPathsFor(cfg.Session.Campaign, cfg.Session.SessionID)
for _, name := range []string{"speakers.yml", "autocorrect.yml"} {
if err := os.WriteFile(filepath.Join(paths.InputsDir, name), []byte("{}"), 0o644); err != nil {
t.Fatal(err)
}
}
merge := &seriatim.FakeRunner{}
if _, err := executeStages(context.Background(), cfg, []stage.Stage{mustSelectStage(t, "merge")}, RunOptions{
Env: &Env{ManifestStore: store, Seriatim: merge},
}); err != nil {
t.Fatal(err)
}
if len(merge.NormalizeRequests) != 1 || filepath.Base(merge.NormalizeRequests[0].InputTranscriptPath) != "alice.json" {
t.Fatalf("removed speaker reached merge: %#v", merge.NormalizeRequests)
}
if _, err := os.Stat(filepath.Join(paths.TranscriptsRawDir, "bob.json")); err != nil {
t.Fatalf("incidental transcript was removed: %v", err)
}
}
})
}
}
func TestExecuteStagesResumesInterruptedRunningTranscribeState(t *testing.T) {
cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
concurrency := 1

9
internal/stage/transcribe_checkpoint_test.go
View File

@@ -222,6 +222,14 @@ func TestTranscribeResumeValidationUsesSpeakerEvidence(t *testing.T) {
t.Fatalf("Run() error = %v", err)
}
validation, err := (transcribeStage{}).ValidateResume(context.Background(), env, m)
if err != nil || validation.Resumable {
t.Fatalf("ValidateResume() = %#v, %v, want missing aggregate repaired", validation, err)
}
if len(m.Stages["transcribe"].Outputs) != 0 {
t.Fatal("resume inspection mutated aggregate outputs")
}
m.Stages["transcribe"].Outputs = []manifest.ArtifactRecord{m.Stages["transcribe"].TranscribeSpeakers["alice"].Output}
validation, err = (transcribeStage{}).ValidateResume(context.Background(), env, m)
if err != nil || !validation.Resumable {
t.Fatalf("ValidateResume() = %#v, %v, want resumable", validation, err)
}
@@ -280,6 +288,7 @@ func TestTranscribeReuseDoesNotDependOnRunDirectory(t *testing.T) {
if _, err := (transcribeStage{}).Run(context.Background(), env, m); err != nil {
t.Fatalf("Run() error = %v", err)
}
m.Stages["transcribe"].Outputs = []manifest.ArtifactRecord{m.Stages["transcribe"].TranscribeSpeakers["alice"].Output}
paths := sessionPathsForEnv(env, m.SessionID)
if err := os.RemoveAll(paths.RunsDir); err != nil {
t.Fatalf("remove run directory: %v", err)

24
internal/stage/transcribe_reconciliation_test.go
View File

@@ -5,6 +5,7 @@ import (
"errors"
"os"
"path/filepath"
"reflect"
"slices"
"strings"
"testing"
@@ -44,6 +45,29 @@ func TestReconcileTranscribeSpeakersReusesUnchangedCompletedEvidence(t *testing.
}
}
func TestTranscribeResumeAcceptsEquivalentOutputsAndInspectsRemovalReadOnly(t *testing.T) {
env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac", "bob.flac"})
record := m.Stages["transcribe"]
record.Status = manifest.StatusSucceeded
// Reverse speaker order and mix the supported local path representations.
bob := record.TranscribeSpeakers["bob"].Output
bob.LocalPath = filepath.Join(sessionPathsForEnv(env, m.SessionID).Root, filepath.FromSlash(bob.LocalPath))
record.Outputs = []manifest.ArtifactRecord{bob, record.TranscribeSpeakers["alice"].Output}
validation, err := (transcribeStage{}).ValidateResume(context.Background(), env, m)
if err != nil || !validation.Resumable {
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)
validation, err = (transcribeStage{}).ValidateResume(context.Background(), env, m)
if err != nil || validation.Resumable {
t.Fatalf("removed speaker: %#v, %v", validation, err)
}
if !reflect.DeepEqual(m, before) {
t.Fatal("resume inspection mutated manifest")
}
}
func TestReconcileTranscribeSpeakersReplacesOnlyChangedAudio(t *testing.T) {
env, m := setupCurrentTranscribeReconciliation(t, []string{"alice.flac", "bob.flac"})
paths := sessionPathsForEnv(env, m.SessionID)

28
internal/stage/transcribe_resume.go
View File

@@ -6,6 +6,7 @@ import (
"sort"
"strings"
"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
"gitea.maximumdirect.net/eric/narratio/internal/manifest"
)
@@ -19,6 +20,10 @@ func (transcribeStage) ValidateResume(
return ResumeValidation{}, fmt.Errorf("transcribe resume: reconcile speakers: %w", err)
}
if len(reconciliation.work) == 0 {
paths := sessionPathsForEnv(env, m.SessionID)
if !transcribeAggregateCurrent(m.Stages["transcribe"], reconciliation.current, paths) {
return NonResumable("speaker collection or aggregate outputs require reconciliation"), nil
}
return Resumable(), nil
}
speakers := make([]string, 0, len(reconciliation.work))
@@ -29,4 +34,27 @@ func (transcribeStage) ValidateResume(
return NonResumable("speaker transcripts require execution: " + strings.Join(speakers, ", ")), nil
}
// No HTTP work does not imply that the aggregate is current. The reconciled
// collection is a subset of the stored records, so its cardinality also detects
// removed speakers. Compare local output evidence independently of list order
// and of the supported absolute/session-relative path representations.
func transcribeAggregateCurrent(record *manifest.StageRecord, current map[string]manifest.TranscribeSpeakerRecord, paths artifacts.SessionPaths) bool {
if len(record.TranscribeSpeakers) != len(current) || len(record.Outputs) != len(current) {
return false
}
expected := make(map[string]manifest.ArtifactRecord, len(current))
for _, speaker := range current {
expected[artifacts.ResolveSessionLocalPathForRead(paths, speaker.Output.LocalPath)] = speaker.Output
}
for _, output := range record.Outputs {
path := artifacts.ResolveSessionLocalPathForRead(paths, output.LocalPath)
want, ok := expected[path]
if !ok || output.Kind != want.Kind || output.Checksum != want.Checksum || output.ProducerRunID != want.ProducerRunID {
return false
}
delete(expected, path)
}
return len(expected) == 0
}
var _ ResumeValidator = transcribeStage{}