From 702f622e189dea9a8035cff4d7e022138c062fc0 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Mon, 10 Aug 2026 21:53:54 +0000 Subject: [PATCH] Harden prepare and transcribe transitions --- docs/internal/stage-prepare.md | 7 +- docs/internal/stage-transcribe.md | 11 ++- docs/roadmap/implementation.md | 2 + internal/config/validation_duration_test.go | 2 +- internal/stage/prepare.go | 46 ++++++++--- internal/stage/prepare_test.go | 21 ++++- internal/stage/transcribe.go | 10 +++ internal/stage/transcribe_test.go | 88 +++++++++++++++++++++ 8 files changed, 165 insertions(+), 22 deletions(-) diff --git a/docs/internal/stage-prepare.md b/docs/internal/stage-prepare.md index 99a6175..faea272 100644 --- a/docs/internal/stage-prepare.md +++ b/docs/internal/stage-prepare.md @@ -30,10 +30,12 @@ Materialize canonical current-session inputs before processing stages. - validates required config/store state. - enforces local audio vs S3 audio mutual exclusivity. +- rejects duplicate explicit local audio sources after resolution. +- gives distinct local source paths with the same basename deterministic unique + prepared filenames so neither source is overwritten. - materializes S3 audio through spool/cache-aware logic. - scans enabled configured artifact inputs for `narratio.previous_session.artifact.*` requirements. -- when previous requirements exist: - - clears managed `previous/` state; +- clears managed `previous/` state on every invocation, then, when requirements exist: - resolves the pointer-selected previous source through the shared resolver; - downloads previous manifest/artifacts; - records previous inputs in `manifest.inputs`. @@ -47,6 +49,7 @@ mapping, while the isolated legacy reader rejects ambiguous fallback matches. - only `prepare` hydrates canonical `previous/` cache state. - managed previous artifacts are stored under `previous/artifacts/**` without duplicate `artifacts/artifacts/` nesting. +- managed `previous/` state represents only the current requirement set. - `manifest.inputs` ordering is deterministic (`kind`, `path`). ## Related Contracts And Tests diff --git a/docs/internal/stage-transcribe.md b/docs/internal/stage-transcribe.md index c896526..0539cdd 100644 --- a/docs/internal/stage-transcribe.md +++ b/docs/internal/stage-transcribe.md @@ -15,16 +15,19 @@ Generate raw per-speaker transcripts from prepared audio using WhisperX. ## Key Behavior - discovers prepared audio from manifest inputs or canonical audio directory. -- derives speaker ID from `.flac` basename. +- derives the transcript identity from the prepared `.flac` filename. - dispatches WhisperX requests through a bounded worker pool. - validates each output as JSON. -- writes run-local outputs then materializes canonical transcript outputs. +- writes run-local outputs then materializes canonical transcript outputs only + after every planned request succeeds. ## Invariants -- speaker basenames must be unique. +- prepared audio identities must be unique; prepare disambiguates distinct + source paths that share a basename. - output path returned by adapter must match requested output path. -- each successful output is validated before stage success. +- each successful output is validated before stage success, and cancellation or + incomplete dispatch cannot be reported as a successful result. ## Related Contracts And Tests diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index 1265270..11c3f32 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -840,6 +840,8 @@ cancellation at each phase, partial completion, duplicate source, same-basename distinct source, retry, and no-false-manifest tests. This is the primary TST-008 stage; Stage 25 adds output/resolver boundary cases. +**Status:** Completed. + ## Stage 25 — Enforce output-path authority and shared singleton resolution **Read first:** `audit-findings.md` lines 3306–3330 (ARC-006), 3515–3537 diff --git a/internal/config/validation_duration_test.go b/internal/config/validation_duration_test.go index 7d41a96..421be91 100644 --- a/internal/config/validation_duration_test.go +++ b/internal/config/validation_duration_test.go @@ -46,7 +46,7 @@ func TestValidateWhisperXRequiresAbsoluteHTTPSEndpoint(t *testing.T) { Language: "en", Timeout: "1s", Retries: &retries, - RetryDelay: "0s", + RetryDelay: "1ms", Concurrency: &concurrency, } diff --git a/internal/stage/prepare.go b/internal/stage/prepare.go index 31f5c41..9668da1 100644 --- a/internal/stage/prepare.go +++ b/internal/stage/prepare.go @@ -205,10 +205,10 @@ func (prepareStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) (*S previousRequirements := collectPreparePreviousRequirements(env.Config) var previousHydration *previousSessionHydrationResult + if err := clearManagedPreviousState(paths); err != nil { + return nil, fmt.Errorf("prepare: clear previous-session cache: %w", err) + } if len(previousRequirements) > 0 { - if err := clearManagedPreviousState(paths); err != nil { - return nil, fmt.Errorf("prepare: clear previous-session cache: %w", err) - } hydration, err := hydratePreviousSessionArtifacts(ctx, env, paths, previousRequirements) if err != nil { return nil, fmt.Errorf("prepare: hydrate previous-session artifacts: %w", err) @@ -302,6 +302,7 @@ func resolveAudioInputs(sessionDir string, inputs config.SessionInputsConfig) ([ func resolveLocalAudioFiles(sessionDir string, inputs config.SessionInputsConfig) ([]string, error) { if len(inputs.AudioFiles) > 0 { out := make([]string, 0, len(inputs.AudioFiles)) + seen := make(map[string]struct{}, len(inputs.AudioFiles)) for _, p := range inputs.AudioFiles { resolved, err := resolvePath(sessionDir, p) if err != nil { @@ -313,6 +314,10 @@ func resolveLocalAudioFiles(sessionDir string, inputs config.SessionInputsConfig if err := requireFile(resolved, "audio file"); err != nil { return nil, err } + if _, exists := seen[resolved]; exists { + return nil, fmt.Errorf("duplicate audio source %q", resolved) + } + seen[resolved] = struct{}{} out = append(out, resolved) } sort.Strings(out) @@ -352,24 +357,41 @@ func resolveLocalAudioFiles(sessionDir string, inputs config.SessionInputsConfig } func materializeLocalAudioInputs(env *Env, paths artifacts.SessionPaths, resolvedAudio []string, registerInput func(kind, path, checksum string)) error { - copiedByDest := map[string]string{} + destinations := localAudioDestinations(resolvedAudio) for _, src := range resolvedAudio { - base := filepath.Base(src) - if prev, exists := copiedByDest[base]; exists && prev != src { - return fmt.Errorf("duplicate audio basename %q from %q and %q", base, prev, src) - } - copiedByDest[base] = src - - dst := filepath.Join(paths.AudioDir, base) + dst := filepath.Join(paths.AudioDir, destinations[src]) checksum, err := copyFileIfChanged(env.ArtifactStore, src, dst) if err != nil { - return fmt.Errorf("materialize audio %q: %w", base, err) + return fmt.Errorf("materialize audio %q: %w", src, err) } registerInput("audio", dst, checksum) } return nil } +func localAudioDestinations(audioPaths []string) map[string]string { + byBase := make(map[string][]string, len(audioPaths)) + for _, audioPath := range audioPaths { + base := filepath.Base(audioPath) + byBase[base] = append(byBase[base], audioPath) + } + + destinations := make(map[string]string, len(audioPaths)) + for base, sources := range byBase { + if len(sources) == 1 { + destinations[sources[0]] = base + continue + } + extension := filepath.Ext(base) + stem := strings.TrimSuffix(base, extension) + for _, source := range sources { + digest := sha256.Sum256([]byte(filepath.Clean(source))) + destinations[source] = fmt.Sprintf("%s-%s%s", stem, hex.EncodeToString(digest[:8]), extension) + } + } + return destinations +} + type s3AudioMaterializationStats struct { CacheHits int CacheMisses int diff --git a/internal/stage/prepare_test.go b/internal/stage/prepare_test.go index b0c3565..277c103 100644 --- a/internal/stage/prepare_test.go +++ b/internal/stage/prepare_test.go @@ -413,7 +413,7 @@ func TestPrepareStageAudioSourceConflictFails(t *testing.T) { } } -func TestPrepareStageWithoutPreviousRequirementsDoesNotTouchPreviousState(t *testing.T) { +func TestPrepareStageWithoutPreviousRequirementsClearsPreviousState(t *testing.T) { env, m := setupPrepareEnv(t) root := filepath.Dir(env.Config.SessionPath) writeFile(t, filepath.Join(root, "audio", "a.flac"), "a") @@ -427,8 +427,23 @@ func TestPrepareStageWithoutPreviousRequirementsDoesNotTouchPreviousState(t *tes if err != nil { t.Fatalf("prepare.Run() error = %v", err) } - if _, err := os.Stat(stalePath); err != nil { - t.Fatalf("expected previous stale file to remain untouched: %v", err) + if _, err := os.Stat(stalePath); !os.IsNotExist(err) { + t.Fatalf("expected stale previous file to be cleared, stat err = %v", err) + } +} + +func TestPrepareStageRejectsDuplicateExplicitAudioSources(t *testing.T) { + env, m := setupPrepareEnv(t) + root := filepath.Dir(env.Config.SessionPath) + writeFile(t, filepath.Join(root, "audio", "alice.flac"), "alice") + env.Config.Session.Inputs.AudioFiles = []string{"./audio/alice.flac", "./audio/../audio/alice.flac"} + + _, err := (prepareStage{}).Run(context.Background(), env, m) + if err == nil || !strings.Contains(err.Error(), "duplicate audio source") { + t.Fatalf("prepare.Run() error = %v, want duplicate audio source error", err) + } + if len(m.Inputs) != 0 { + t.Fatalf("manifest inputs = %#v, want no prepared inputs", m.Inputs) } } diff --git a/internal/stage/transcribe.go b/internal/stage/transcribe.go index dd8ad64..01702ba 100644 --- a/internal/stage/transcribe.go +++ b/internal/stage/transcribe.go @@ -112,6 +112,7 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) jobCh = make(chan job) mu sync.Mutex firstErr error + completed int perFile = map[string]map[string]any{} outputRef = map[string]artifacts.Ref{} ) @@ -175,6 +176,7 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) SessionID: sessionID, AbsolutePath: j.outPath, } + completed++ mu.Unlock() } } @@ -184,12 +186,14 @@ func (transcribeStage) Run(ctx context.Context, env *Env, m *manifest.Manifest) go worker() } + dispatched := 0 dispatch: for _, j := range jobs { select { case <-stageCtx.Done(): break dispatch case jobCh <- j: + dispatched++ } } close(jobCh) @@ -198,6 +202,12 @@ dispatch: if firstErr != nil { return nil, firstErr } + if err := ctx.Err(); err != nil { + return nil, fmt.Errorf("transcribe: canceled after dispatching %d of %d audio jobs and completing %d: %w", dispatched, len(jobs), completed, err) + } + if completed != len(jobs) { + return nil, fmt.Errorf("transcribe: completed %d of %d planned audio jobs after dispatching %d", completed, len(jobs), dispatched) + } speakers := make([]string, 0, len(perFile)) for speaker := range perFile { diff --git a/internal/stage/transcribe_test.go b/internal/stage/transcribe_test.go index f1ec2c8..1ba6d01 100644 --- a/internal/stage/transcribe_test.go +++ b/internal/stage/transcribe_test.go @@ -222,6 +222,94 @@ func TestTranscribeStageUsesRunLocalOutputAndMaterializesCanonical(t *testing.T) } } +func TestTranscribeStagePreservesDistinctSameBasenameAudio(t *testing.T) { + env, m := setupTranscribeEnv(t, nil) + root := filepath.Dir(env.Config.SessionPath) + writeFile(t, filepath.Join(root, "first", "alice.flac"), "first") + writeFile(t, filepath.Join(root, "second", "alice.flac"), "second") + env.Config.Session.Inputs.AudioDir = "" + env.Config.Session.Inputs.AudioFiles = []string{"./first/alice.flac", "./second/alice.flac"} + env.WhisperX = &whisperx.FakeClient{} + + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + result, err := (transcribeStage{}).Run(context.Background(), env, m) + if err != nil { + t.Fatalf("transcribe.Run() error = %v", err) + } + if len(result.Outputs) != 2 { + t.Fatalf("outputs = %#v, want two distinct outputs", result.Outputs) + } + if result.Outputs[0].AbsolutePath == result.Outputs[1].AbsolutePath { + t.Fatalf("output paths = %#v, want distinct identities", result.Outputs) + } + requests := env.WhisperX.(*whisperx.FakeClient).RequestsSnapshot() + if len(requests) != 2 || requests[0].AudioPath == requests[1].AudioPath || requests[0].OutputRawTranscriptPath == requests[1].OutputRawTranscriptPath { + t.Fatalf("requests = %#v, want distinct audio and output paths", requests) + } +} + +func TestTranscribeStageCancellationDoesNotMaterializePartialOutputs(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac", "bob.flac", "cara.flac"}) + m.RunID = "20260810T214607Z-a1b2c3d4" + fastComplete := make(chan struct{}) + env.WhisperX = &whisperx.FakeClient{ + TranscribeFn: func(ctx context.Context, req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) { + if req.SpeakerID == "alice" { + if err := writeJSONFile(req.OutputRawTranscriptPath, map[string]any{"speaker": req.SpeakerID}); err != nil { + return whisperx.TranscribeResult{}, err + } + close(fastComplete) + return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil + } + <-ctx.Done() + return whisperx.TranscribeResult{}, ctx.Err() + }, + } + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + + ctx, cancel := context.WithCancel(context.Background()) + done := make(chan error, 1) + go func() { + _, err := (transcribeStage{}).Run(ctx, env, m) + done <- err + }() + select { + case <-fastComplete: + cancel() + case <-time.After(time.Second): + cancel() + t.Fatal("no transcription completed before cancellation") + } + if err := <-done; err == nil || !errors.Is(err, context.Canceled) { + t.Fatalf("transcribe.Run() error = %v, want context cancellation", err) + } + + canonical := filepath.Join(sessionPathsForEnv(env, m.SessionID).TranscriptsRawDir, "alice.json") + if _, err := os.Stat(canonical); !os.IsNotExist(err) { + t.Fatalf("partial canonical output should not exist, stat err = %v", err) + } +} + +func TestTranscribeStageRejectsCanceledContext(t *testing.T) { + env, m := setupTranscribeEnv(t, []string{"alice.flac"}) + env.WhisperX = &whisperx.FakeClient{} + if _, err := (prepareStage{}).Run(context.Background(), env, m); err != nil { + t.Fatalf("prepare.Run() error = %v", err) + } + ctx, cancel := context.WithCancel(context.Background()) + cancel() + if _, err := (transcribeStage{}).Run(ctx, env, m); err == nil || !errors.Is(err, context.Canceled) { + t.Fatalf("transcribe.Run() error = %v, want context cancellation", err) + } + if got := len(env.WhisperX.(*whisperx.FakeClient).RequestsSnapshot()); got != 0 { + t.Fatalf("adapter calls = %d, want none", got) + } +} + func setupTranscribeEnv(t *testing.T, audioFiles []string) (*Env, *manifest.Manifest) { t.Helper()