diff --git a/docs/internal/stage-transcribe.md b/docs/internal/stage-transcribe.md index 6d69938..2b088ec 100644 --- a/docs/internal/stage-transcribe.md +++ b/docs/internal/stage-transcribe.md @@ -62,7 +62,11 @@ timeout, credentials, and diagnostic locations are operational and do not change this evidence. An ordinary retry reuses every valid speaker checkpoint even when aggregate -transcribe status is failed, running, or stale. `--force` first durably replaces +transcribe status is failed, running, or stale. A succeeded stage is skipped only +when its stored speaker set and aggregate output evidence also match the current +collection. Otherwise, execution repairs those records without retranscribing +reusable speakers and stales dependents through the normal runner path. +`--force` first durably replaces the collection with an empty versioned collection, so a later ordinary retry cannot resurrect pre-force results; speakers checkpointed by that forced attempt remain reusable. diff --git a/docs/roadmap/remediation.md b/docs/roadmap/remediation.md new file mode 100644 index 0000000..e5f6b55 --- /dev/null +++ b/docs/roadmap/remediation.md @@ -0,0 +1,96 @@ +# Transcription Checkpoint Review Remediation + +Status: Open. + +Reviewed commits `7b0bf99` through `96b4e73` against the speaker-checkpoint and +upload-replay roadmap and the original user request. No production changes were +made during review. The two failures below were reproduced with temporary tests +using the existing stage fixtures; those temporary tests were removed afterward. + +## 1. Reconcile Aggregate State Even When No Speaker Needs HTTP Work + +Priority: P2. Status: Complete. + +Implemented read-only aggregate identity/evidence checks before resume skips. +Regression tests cover removed inputs, missing/duplicate/mismatched outputs, +repair without HTTP requests, downstream invalidation, and exclusion of removed +speakers from merge without deleting their files. Equivalent path forms and +output ordering remain reusable. Focused app/stage tests and documentation +checks pass. + +Location: `internal/stage/transcribe_resume.go:21` and the reconciliation result +consumed there. + +`ValidateResume` returns reusable whenever `reconciliation.work` is empty. That +only establishes that the remaining speakers need no transcription; it does +not establish that the stored speaker collection and aggregate outputs match +the current input set. Removing a speaker while the remaining inputs are +unchanged produces an empty work list, so a succeeded record can be skipped +without pruning the removed speaker. Its old aggregate output can then remain +available to merge. Likewise, missing/inconsistent aggregate output references +are not repaired when all individual checkpoints remain valid. A prepare rerun +may already stale transcribe in some workflows, but the resume validator cannot +rely on that to satisfy its own current-input/output checks. + +Reproduction: create complete alice/bob checkpoint records and aggregate +outputs with the existing fixture, mark the stage succeeded, remove bob from +the manifest's audio inputs, and call `ValidateResume`. It returns reusable +despite still retaining bob in the stored collection and aggregate outputs. + +Required change: + +- Distinguish no HTTP work from no reconciliation work. Validate that the + committed speaker set and aggregate transcript references represent the + reconciled current set, including relevant identity/evidence, before skipping. +- When only collection/output repair is needed, return non-resumable and let + normal execution checkpoint the pruned collection and rebuild aggregate + outputs without retranscribing reusable speakers. Preserve existing dependent + invalidation; do not force otherwise-current speaker jobs. +- Keep resume inspection read-only. Do not fix this by deleting incidental + transcript files or changing stage selection rules. + +Verification: add removed-only input coverage (not just removal combined with +an added speaker), plus incomplete/inconsistent aggregate output coverage. +Exercise the runner skip path and assert zero additional WhisperX requests, +correct rebuilt outputs, and no removed speaker reaching merge. + +## 2. Remove the Unrequested Positive Audio-Size Requirement + +Priority: P2. Status: Pending. + +Location: `internal/manifest/transcribe_state.go:95`. + +The new record validator requires `AudioSize > 0`. Previously Narratio accepted +a regular `.flac` file and delegated transcription acceptance to WhisperX; +neither the user request nor the roadmap introduced minimum audio size or a new +audio-content validation policy. Zero-byte content still has a valid checksum +and can be represented by the checkpoint mechanism. The current check runs +after a successful response and output materialization, causing the entire +stage to fail instead of retaining the completed work. + +Reproduction: prepare one speaker, replace its prepared audio with an empty +regular file, and use the existing successful fake WhisperX client. Stage +execution fails at checkpoint validation with `audio_size must be positive`. +This demonstrates the extra Narratio restriction; it does not assert that a +particular production WhisperX deployment accepts empty audio. + +Required change: + +- Permit zero audio size and reject negative sizes. Keep byte hashing and + size equality checks. Let the downstream service decide whether the input + can be transcribed, as before. +- Do not remove the nonempty valid-JSON output requirement, which has a separate + existing justification. Do not add a new early empty-audio rejection instead. +- Update any new tests/documentation that promote strictly positive input size + to a contract. + +Verification: cover a zero-size record round-trip and successful adapter result +being checkpointed/reused for that input; retain negative-size rejection and +ordinary downstream-failure tests. + +## Verification After Remediation + +Run focused stage/manifest/application tests, changed-package race tests, and +the standard repository validation gates from `docs/development.md`. Preserve +the existing replay and cancellation regression tests; these findings do not +require redesigning the HTTP adapter. diff --git a/internal/app/transcribe_checkpoint_recovery_test.go b/internal/app/transcribe_checkpoint_recovery_test.go index 8a9cf9b..bd3060b 100644 --- a/internal/app/transcribe_checkpoint_recovery_test.go +++ b/internal/app/transcribe_checkpoint_recovery_test.go @@ -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 diff --git a/internal/stage/transcribe_checkpoint_test.go b/internal/stage/transcribe_checkpoint_test.go index 7e3df62..78e42c8 100644 --- a/internal/stage/transcribe_checkpoint_test.go +++ b/internal/stage/transcribe_checkpoint_test.go @@ -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) diff --git a/internal/stage/transcribe_reconciliation_test.go b/internal/stage/transcribe_reconciliation_test.go index 707ef87..98d42be 100644 --- a/internal/stage/transcribe_reconciliation_test.go +++ b/internal/stage/transcribe_reconciliation_test.go @@ -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) diff --git a/internal/stage/transcribe_resume.go b/internal/stage/transcribe_resume.go index f573b0d..fae484a 100644 --- a/internal/stage/transcribe_resume.go +++ b/internal/stage/transcribe_resume.go @@ -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{}