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

Harden prepare and transcribe transitions

Browse Source
This commit is contained in:
Eric Rakestraw
2026-08-10 21:53:54 +00:00
parent 9da2c1e144
commit 702f622e18
8 changed files with 165 additions and 22 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
docs/internal/stage-prepare.md
View File

@@ -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

11
docs/internal/stage-transcribe.md
View File

@@ -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

2
docs/roadmap/implementation.md
View File

@@ -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

2
internal/config/validation_duration_test.go
View File

@@ -46,7 +46,7 @@ func TestValidateWhisperXRequiresAbsoluteHTTPSEndpoint(t *testing.T) {
Language: "en",
Timeout: "1s",
Retries: &retries,
RetryDelay: "0s",
RetryDelay: "1ms",
Concurrency: &concurrency,
}

46
internal/stage/prepare.go
View File

@@ -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

21
internal/stage/prepare_test.go
View File

@@ -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)
}
}

10
internal/stage/transcribe.go
View File

@@ -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 {

88
internal/stage/transcribe_test.go
View File

@@ -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()