Integrate transcription checkpoints across lifecycle

2026-09-22 00:00:51 +00:00
parent 2ae63dbc2b
commit d872edca94
7 changed files with 581 additions and 2 deletions
--- a/docs/roadmap/implementation.md
+++ b/docs/roadmap/implementation.md
@@ -1,6 +1,6 @@
 # Implementation Plan: Speaker Checkpoints and Replayable Uploads

-Status: Implementation in progress; Stages 1-3 complete.
+Status: Implementation in progress; Stages 1-4 complete.

 ## Scope and Instructions

@@ -241,7 +241,18 @@ ordinary, concurrent, canceled, and forced execution.

 ## Stage 4: Verify Runner, Merge, and Durable Lifecycle Integration

-Status: Pending.
+Status: Complete.
+
+Completion marker (2026-09-22): Added real-persistence application coverage for
+cross-invocation retry, interrupted running state, failed aggregate saves,
+force/reset persistence, downstream invalidation, and one-time legacy upgrade.
+Versioned merge input discovery now requires aggregate success and an exact
+match between current speaker evidence and declared aggregate outputs, while
+legacy/manual directory fallback remains available. Restore coverage preserves
+session-relative speaker paths, leaves unpublished raw transcripts absent, and
+canonical transcripts remain reusable after run-directory cleanup. Verified
+with `go test ./internal/app ./internal/stage ./internal/manifest`,
+`go test -race ./internal/app`, and `git diff --check`.

 Preparation: `internal/app/runner.go`, `runner_runtime.go`,
 `analyze_checkpoint_recovery_test.go`, `runner_terminal_failure_test.go`,
--- a/internal/app/restore_execution_test.go
+++ b/internal/app/restore_execution_test.go
@@ -82,6 +82,28 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi
 	remoteManifest.RunID = "20260519T010203Z-a1b2c3d4"
 	producerRoot := "/prior/workspace/work/sample-campaign/2026-05-03"
 	remoteManifest.LocalWorkDir = filepath.Join(producerRoot, "runs", remoteManifest.RunID)
+	rawRelative := "transcripts/raw/alice.json"
+	rawOutput := manifest.ArtifactRecord{
+		Kind: "transcript_raw", LocalPath: filepath.Join(producerRoot, filepath.FromSlash(rawRelative)),
+		ProducerRunID: remoteManifest.RunID, Checksum: strings.Repeat("b", 64),
+	}
+	remoteManifest.MarkStageSucceeded("transcribe", time.Now().UTC(), []manifest.ArtifactRecord{rawOutput})
+	remoteManifest.Stages["transcribe"].TranscribeStateVersion = manifest.TranscribeStateContractVersion
+	remoteManifest.Stages["transcribe"].TranscribeSpeakers = map[string]manifest.TranscribeSpeakerRecord{
+		"alice": {
+			FingerprintVersion: manifest.TranscribeFingerprintContractVersion,
+			Fingerprint:        strings.Repeat("c", 64),
+			AudioPath:          "audio/alice.flac",
+			AudioSHA256:        strings.Repeat("a", 64),
+			AudioSize:          12,
+			Output: manifest.ArtifactRecord{
+				Kind: "transcript_raw", LocalPath: rawRelative,
+				ProducerRunID: remoteManifest.RunID, Checksum: strings.Repeat("b", 64),
+			},
+			OutputSize:    15,
+			ProducerRunID: remoteManifest.RunID,
+		},
+	}
 	remoteManifest.Stages["extract"] = &manifest.StageRecord{
 		Name: "extract", Status: manifest.StatusSucceeded,
 		Outputs: []manifest.ArtifactRecord{{
@@ -121,6 +143,13 @@ func TestExecuteRestoreRoundTripsPublishedExtractionAndManifestMetadata(t *testi
 	if want := filepath.Join(sessionRoot, "artifacts", "encounters.json"); lane.LocalPath != want {
 		t.Fatalf("rebased extraction path = %q, want %q", lane.LocalPath, want)
 	}
+	speaker := restored.Stages["transcribe"].TranscribeSpeakers["alice"]
+	if speaker.AudioPath != "audio/alice.flac" || speaker.Output.LocalPath != rawRelative {
+		t.Fatalf("restored speaker paths = %#v, want session-relative paths", speaker)
+	}
+	if _, err := os.Stat(filepath.Join(sessionRoot, filepath.FromSlash(rawRelative))); !os.IsNotExist(err) {
+		t.Fatalf("unpublished raw transcript should not be restored; stat err=%v", err)
+	}
 }

 func TestExecuteRestoreIncludeAudioRestoresAudio(t *testing.T) {
--- a/internal/app/runner_test_helpers_test.go
+++ b/internal/app/runner_test_helpers_test.go
@@ -6,6 +6,7 @@ import (
 	"encoding/hex"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"testing"

@@ -126,6 +127,14 @@ func seedCurrentTranscribeEvidence(
 	record := m.Stages["transcribe"]
 	record.TranscribeStateVersion = manifest.TranscribeStateContractVersion
 	record.TranscribeSpeakers = records
+	if len(record.Outputs) == 0 {
+		for _, speakerRecord := range records {
+			record.Outputs = append(record.Outputs, speakerRecord.Output)
+		}
+		slices.SortFunc(record.Outputs, func(a, b manifest.ArtifactRecord) int {
+			return strings.Compare(a.LocalPath, b.LocalPath)
+		})
+	}
 }

 func loadConfigForSemanticEvidence(t *testing.T, pipelinePath, campaignPath, sessionPath string) *config.Config {
--- a/internal/app/transcribe_checkpoint_recovery_test.go
+++ b/internal/app/transcribe_checkpoint_recovery_test.go
@@ -0,0 +1,338 @@
+package app
+
+import (
+	"context"
+	"errors"
+	"os"
+	"path/filepath"
+	"slices"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/adapters/whisperx"
+	"gitea.maximumdirect.net/eric/narratio/internal/artifacts"
+	"gitea.maximumdirect.net/eric/narratio/internal/config"
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+	"gitea.maximumdirect.net/eric/narratio/internal/stage"
+)
+
+func TestExecuteStagesRetriesOnlyUncommittedTranscribeSpeakers(t *testing.T) {
+	cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"a", "b", "c", "d"})
+	concurrency := 1
+	cfg.Pipeline.WhisperX.Concurrency = &concurrency
+	cause := errors.New("speaker c failed")
+	failing := &whisperx.FakeClient{TranscribeFn: func(
+		_ context.Context,
+		req whisperx.TranscribeRequest,
+	) (whisperx.TranscribeResult, error) {
+		if req.SpeakerID == "c" {
+			return whisperx.TranscribeResult{}, cause
+		}
+		if req.SpeakerID == "d" {
+			t.Fatalf("speaker d was dispatched after c failed")
+		}
+		return writeTranscribeRecoveryResult(req)
+	}}
+	transcribe := mustSelectStage(t, "transcribe")
+
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: baseStore, WhisperX: failing},
+	}); !errors.Is(err, cause) {
+		t.Fatalf("failed executeStages() error = %v, want %v", err, cause)
+	}
+	failed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
+	if failed.Stages["transcribe"].Status != manifest.StatusFailed {
+		t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status)
+	}
+	if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"a", "b"}) {
+		t.Fatalf("checkpointed speakers = %#v, want a and b", got)
+	}
+	if failed.Stages["prepare"].Status != manifest.StatusSucceeded {
+		t.Fatalf("prepare status = %q, want unchanged success", failed.Stages["prepare"].Status)
+	}
+	if failed.Stages["merge"].Status != manifest.StatusStale {
+		t.Fatalf("merge status = %q, want stale after transcribe failure", failed.Stages["merge"].Status)
+	}
+
+	retry := &whisperx.FakeClient{}
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: baseStore, WhisperX: retry},
+	}); err != nil {
+		t.Fatalf("retry executeStages() error = %v", err)
+	}
+	if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"c", "d"}) {
+		t.Fatalf("retry speakers = %#v, want c and d", got)
+	}
+	completed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
+	if completed.Stages["transcribe"].Status != manifest.StatusSucceeded {
+		t.Fatalf("transcribe status = %q, want succeeded", completed.Stages["transcribe"].Status)
+	}
+	if got := transcribeSpeakerKeys(completed); !slices.Equal(got, []string{"a", "b", "c", "d"}) {
+		t.Fatalf("completed speakers = %#v", got)
+	}
+}
+
+func TestExecuteStagesRebuildsTranscribeAggregateAfterFinalSaveFailure(t *testing.T) {
+	cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
+	store := &transcribeAggregateSaveFailingStore{
+		Store: baseStore,
+		Err:   errors.New("aggregate save failed"),
+	}
+	fake := &whisperx.FakeClient{}
+	transcribe := mustSelectStage(t, "transcribe")
+
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: store, WhisperX: fake},
+	}); !errors.Is(err, store.Err) {
+		t.Fatalf("executeStages() error = %v, want aggregate save failure", err)
+	}
+	failed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
+	if failed.Stages["transcribe"].Status != manifest.StatusFailed {
+		t.Fatalf("transcribe status = %q, want failed", failed.Stages["transcribe"].Status)
+	}
+	if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice", "bob"}) {
+		t.Fatalf("checkpointed speakers = %#v, want both speakers", got)
+	}
+
+	retry := &whisperx.FakeClient{}
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: baseStore, WhisperX: retry},
+	}); err != nil {
+		t.Fatalf("retry executeStages() error = %v", err)
+	}
+	if got := len(retry.RequestsSnapshot()); got != 0 {
+		t.Fatalf("retry adapter requests = %d, want aggregate-only rebuild", got)
+	}
+	completed := loadTranscribeRecoveryManifest(t, baseStore, cfg)
+	if completed.Stages["transcribe"].Status != manifest.StatusSucceeded || len(completed.Stages["transcribe"].Outputs) != 2 {
+		t.Fatalf("rebuilt aggregate = %#v", completed.Stages["transcribe"])
+	}
+}
+
+func TestExecuteStagesResumesInterruptedRunningTranscribeState(t *testing.T) {
+	cfg, baseStore := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
+	concurrency := 1
+	cfg.Pipeline.WhisperX.Concurrency = &concurrency
+	stageErr := errors.New("speaker b failed")
+	terminalErr := errors.New("terminal save interrupted")
+	store := &interruptAfterTranscribeCheckpointStore{Store: baseStore, Err: terminalErr}
+	fake := &whisperx.FakeClient{TranscribeFn: func(
+		_ context.Context,
+		req whisperx.TranscribeRequest,
+	) (whisperx.TranscribeResult, error) {
+		if req.SpeakerID == "bob" {
+			return whisperx.TranscribeResult{}, stageErr
+		}
+		return writeTranscribeRecoveryResult(req)
+	}}
+	transcribe := mustSelectStage(t, "transcribe")
+
+	_, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: store, WhisperX: fake},
+	})
+	if !errors.Is(err, stageErr) || !errors.Is(err, terminalErr) {
+		t.Fatalf("executeStages() error = %v, want stage and terminal failures", err)
+	}
+	interrupted := loadTranscribeRecoveryManifest(t, baseStore, cfg)
+	if interrupted.Stages["transcribe"].Status != manifest.StatusRunning {
+		t.Fatalf("transcribe status = %q, want durable running state", interrupted.Stages["transcribe"].Status)
+	}
+	if got := transcribeSpeakerKeys(interrupted); !slices.Equal(got, []string{"alice"}) {
+		t.Fatalf("checkpointed speakers = %#v, want alice", got)
+	}
+
+	retry := &whisperx.FakeClient{}
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: baseStore, WhisperX: retry},
+	}); err != nil {
+		t.Fatalf("retry executeStages() error = %v", err)
+	}
+	if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) {
+		t.Fatalf("retry speakers = %#v, want bob", got)
+	}
+}
+
+func TestExecuteStagesPersistsForcedTranscribeResetAcrossRetry(t *testing.T) {
+	cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice", "bob"})
+	concurrency := 1
+	cfg.Pipeline.WhisperX.Concurrency = &concurrency
+	transcribe := mustSelectStage(t, "transcribe")
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: store, WhisperX: &whisperx.FakeClient{}},
+	}); err != nil {
+		t.Fatalf("baseline executeStages() error = %v", err)
+	}
+
+	cause := errors.New("forced bob failed")
+	forced := &whisperx.FakeClient{TranscribeFn: func(
+		_ context.Context,
+		req whisperx.TranscribeRequest,
+	) (whisperx.TranscribeResult, error) {
+		if req.SpeakerID == "bob" {
+			return whisperx.TranscribeResult{}, cause
+		}
+		return writeTranscribeRecoveryResult(req)
+	}}
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Force: true,
+		Env:   &Env{ManifestStore: store, WhisperX: forced},
+	}); !errors.Is(err, cause) {
+		t.Fatalf("forced executeStages() error = %v, want %v", err, cause)
+	}
+	failed := loadTranscribeRecoveryManifest(t, store, cfg)
+	if got := transcribeSpeakerKeys(failed); !slices.Equal(got, []string{"alice"}) {
+		t.Fatalf("speakers after forced failure = %#v, want only replacement alice", got)
+	}
+
+	retry := &whisperx.FakeClient{}
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: store, WhisperX: retry},
+	}); err != nil {
+		t.Fatalf("retry executeStages() error = %v", err)
+	}
+	if got := requestedTranscribeSpeakers(retry); !slices.Equal(got, []string{"bob"}) {
+		t.Fatalf("retry speakers = %#v, want bob", got)
+	}
+}
+
+func TestExecuteStagesUpgradesLegacyTranscribeSuccessOnce(t *testing.T) {
+	cfg, store := setupTranscribeRecoveryFixture(t, []string{"alice"})
+	m := loadTranscribeRecoveryManifest(t, store, cfg)
+	m.MarkStageSucceeded("transcribe", time.Now().UTC(), []manifest.ArtifactRecord{{
+		Kind: "transcript_raw", LocalPath: "transcripts/raw/alice.json",
+	}})
+	if err := store.Save(context.Background(), manifestPathFor(cfg), m); err != nil {
+		t.Fatalf("save legacy transcribe success: %v", err)
+	}
+	transcribe := mustSelectStage(t, "transcribe")
+	upgrade := &whisperx.FakeClient{}
+	if _, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: store, WhisperX: upgrade},
+	}); err != nil {
+		t.Fatalf("upgrade executeStages() error = %v", err)
+	}
+	if got := requestedTranscribeSpeakers(upgrade); !slices.Equal(got, []string{"alice"}) {
+		t.Fatalf("upgrade speakers = %#v, want alice", got)
+	}
+	upgraded := loadTranscribeRecoveryManifest(t, store, cfg)
+	if upgraded.Stages["transcribe"].TranscribeStateVersion != manifest.TranscribeStateContractVersion {
+		t.Fatalf("transcribe state version = %d, want current", upgraded.Stages["transcribe"].TranscribeStateVersion)
+	}
+
+	reused := &whisperx.FakeClient{}
+	summary, err := executeStages(context.Background(), cfg, []stage.Stage{transcribe}, RunOptions{
+		Env: &Env{ManifestStore: store, WhisperX: reused},
+	})
+	if err != nil {
+		t.Fatalf("reuse executeStages() error = %v", err)
+	}
+	if len(summary.Skipped) != 1 || len(reused.RequestsSnapshot()) != 0 {
+		t.Fatalf("reuse summary = %#v, requests = %#v, want skipped without HTTP", summary, reused.RequestsSnapshot())
+	}
+}
+
+type transcribeAggregateSaveFailingStore struct {
+	manifest.Store
+	Err    error
+	Failed bool
+}
+
+func (store *transcribeAggregateSaveFailingStore) Save(
+	ctx context.Context,
+	path string,
+	m *manifest.Manifest,
+) error {
+	record := m.Stages["transcribe"]
+	if !store.Failed && record != nil && record.Status == manifest.StatusSucceeded &&
+		record.TranscribeStateVersion == manifest.TranscribeStateContractVersion {
+		store.Failed = true
+		return store.Err
+	}
+	return store.Store.Save(ctx, path, m)
+}
+
+type interruptAfterTranscribeCheckpointStore struct {
+	manifest.Store
+	Err         error
+	Interrupted bool
+}
+
+func (store *interruptAfterTranscribeCheckpointStore) Save(
+	ctx context.Context,
+	path string,
+	m *manifest.Manifest,
+) error {
+	if store.Interrupted {
+		return store.Err
+	}
+	record := m.Stages["transcribe"]
+	err := store.Store.Save(ctx, path, m)
+	if err == nil && record != nil && record.Status == manifest.StatusRunning && len(record.TranscribeSpeakers) == 1 {
+		store.Interrupted = true
+	}
+	return err
+}
+
+func setupTranscribeRecoveryFixture(t *testing.T, speakers []string) (*config.Config, *manifest.LocalStore) {
+	t.Helper()
+	cfg := testConfig(t)
+	store := &manifest.LocalStore{}
+	artifactStore := artifacts.NewLocalStore(cfg.Pipeline.Workspace.Root)
+	paths, err := artifactStore.EnsureLayoutFor(cfg.Session.Campaign, cfg.Session.SessionID)
+	if err != nil {
+		t.Fatalf("prepare session layout: %v", err)
+	}
+	m := manifest.New(cfg.Session.SessionID, time.Now().UTC())
+	m.Campaign = cfg.Session.Campaign
+	m.Inputs = make([]manifest.InputRecord, 0, len(speakers))
+	for _, speaker := range speakers {
+		relative := filepath.ToSlash(filepath.Join("audio", speaker+".flac"))
+		if err := os.WriteFile(filepath.Join(paths.AudioDir, speaker+".flac"), []byte("audio-"+speaker), 0o644); err != nil {
+			t.Fatalf("write audio fixture: %v", err)
+		}
+		m.Inputs = append(m.Inputs, manifest.InputRecord{Kind: "audio", Path: relative})
+	}
+	m.MarkStageSucceeded("prepare", time.Now().UTC(), nil)
+	m.MarkStageSucceeded("merge", time.Now().UTC(), nil)
+	if err := store.Save(context.Background(), paths.ManifestPath, m); err != nil {
+		t.Fatalf("save recovery fixture: %v", err)
+	}
+	return cfg, store
+}
+
+func writeTranscribeRecoveryResult(req whisperx.TranscribeRequest) (whisperx.TranscribeResult, error) {
+	if err := os.MkdirAll(filepath.Dir(req.OutputRawTranscriptPath), 0o755); err != nil {
+		return whisperx.TranscribeResult{}, err
+	}
+	if err := os.WriteFile(req.OutputRawTranscriptPath, []byte(`{"segments":[]}`), 0o644); err != nil {
+		return whisperx.TranscribeResult{}, err
+	}
+	return whisperx.TranscribeResult{OutputRawTranscriptPath: req.OutputRawTranscriptPath}, nil
+}
+
+func loadTranscribeRecoveryManifest(t *testing.T, store manifest.Store, cfg *config.Config) *manifest.Manifest {
+	t.Helper()
+	m, err := store.Load(context.Background(), manifestPathFor(cfg))
+	if err != nil {
+		t.Fatalf("load transcribe recovery manifest: %v", err)
+	}
+	return m
+}
+
+func transcribeSpeakerKeys(m *manifest.Manifest) []string {
+	keys := make([]string, 0, len(m.Stages["transcribe"].TranscribeSpeakers))
+	for key := range m.Stages["transcribe"].TranscribeSpeakers {
+		keys = append(keys, key)
+	}
+	slices.Sort(keys)
+	return keys
+}
+
+func requestedTranscribeSpeakers(fake *whisperx.FakeClient) []string {
+	requests := fake.RequestsSnapshot()
+	speakers := make([]string, 0, len(requests))
+	for _, request := range requests {
+		speakers = append(speakers, request.SpeakerID)
+	}
+	return speakers
+}
--- a/internal/stage/merge.go
+++ b/internal/stage/merge.go
@@ -306,6 +306,12 @@ func normalizeMergeInputs(
 }

 func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths) ([]string, error) {
+	if record, versioned, err := transcribeManifestState(m); err != nil {
+		return nil, err
+	} else if versioned {
+		return discoverVersionedRawTranscripts(record, paths)
+	}
+
 	fromManifest := make([]string, 0)
 	if m != nil && m.Stages != nil {
 		if tr := m.Stages["transcribe"]; tr != nil {
@@ -350,6 +356,64 @@ func discoverRawTranscripts(m *manifest.Manifest, paths artifacts.SessionPaths)
 	return dedupeAndSortPaths(out), nil
 }

+func discoverVersionedRawTranscripts(
+	record *manifest.StageRecord,
+	paths artifacts.SessionPaths,
+) ([]string, error) {
+	if record.Status != manifest.StatusSucceeded {
+		return nil, fmt.Errorf("versioned transcribe state is not aggregate-successful (status %q)", record.Status)
+	}
+	if err := manifest.ValidateTranscribeSpeakerCollection(
+		record.TranscribeStateVersion,
+		record.TranscribeSpeakers,
+	); err != nil {
+		return nil, fmt.Errorf("validate versioned transcribe state: %w", err)
+	}
+	if len(record.TranscribeSpeakers) == 0 {
+		return nil, fmt.Errorf("versioned transcribe state has no completed speakers")
+	}
+
+	expected := make(map[string]string, len(record.TranscribeSpeakers))
+	for speaker, speakerRecord := range record.TranscribeSpeakers {
+		path := filepath.Clean(artifacts.ResolveSessionLocalPathForRead(paths, speakerRecord.Output.LocalPath))
+		if other, exists := expected[path]; exists {
+			return nil, fmt.Errorf("versioned transcribe speakers %q and %q declare the same output", other, speaker)
+		}
+		expected[path] = speaker
+	}
+	inputs := make([]string, 0, len(expected))
+	for _, output := range record.Outputs {
+		if output.Kind != "transcript_raw" {
+			continue
+		}
+		path := strings.TrimSpace(output.LocalPath)
+		if path == "" {
+			return nil, fmt.Errorf("versioned transcribe aggregate output has an empty local path")
+		}
+		path = filepath.Clean(artifacts.ResolveSessionLocalPathForRead(paths, path))
+		speaker, declared := expected[path]
+		if !declared {
+			return nil, fmt.Errorf("versioned transcribe aggregate output %q is not declared by current speaker state", output.LocalPath)
+		}
+		if speaker == "" {
+			return nil, fmt.Errorf("versioned transcribe aggregate output %q is duplicated", output.LocalPath)
+		}
+		expected[path] = ""
+		inputs = append(inputs, path)
+	}
+	missing := make([]string, 0)
+	for _, speaker := range expected {
+		if speaker != "" {
+			missing = append(missing, speaker)
+		}
+	}
+	if len(missing) > 0 {
+		sort.Strings(missing)
+		return nil, fmt.Errorf("versioned transcribe aggregate outputs are incomplete for speakers: %s", strings.Join(missing, ", "))
+	}
+	return dedupeAndSortPaths(inputs), nil
+}
+
 func dedupeAndSortPaths(paths []string) []string {
 	if len(paths) == 0 {
 		return nil
--- a/internal/stage/merge_test.go
+++ b/internal/stage/merge_test.go
@@ -5,6 +5,7 @@ import (
 	"fmt"
 	"os"
 	"path/filepath"
+	"slices"
 	"strings"
 	"testing"
 	"time"
@@ -191,6 +192,70 @@ func TestMergeStageFallsBackToRawDirectoryWhenTranscribeOutputsMissing(t *testin
 	}
 }

+func TestMergeStageRejectsPartialVersionedTranscribeState(t *testing.T) {
+	env, m := setupMergeEnv(t)
+	paths := sessionPathsForEnv(env, m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.TranscriptsRawDir, "uncommitted.json"), `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
+	writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
+	setVersionedTranscribeMergeState(t, m, manifest.StatusFailed, map[string]string{
+		"alice": "transcripts/raw/alice.json",
+	})
+	fake := &seriatim.FakeRunner{}
+	env.Seriatim = fake
+
+	_, err := (mergeStage{}).Run(context.Background(), env, m)
+	if err == nil || !strings.Contains(err.Error(), "not aggregate-successful") {
+		t.Fatalf("merge.Run() error = %v, want incomplete aggregate rejection", err)
+	}
+	if len(fake.Requests) != 0 || len(fake.NormalizeRequests) != 0 {
+		t.Fatalf("adapter requests = %#v / %#v, want none", fake.Requests, fake.NormalizeRequests)
+	}
+}
+
+func TestMergeStageUsesOnlyCurrentVersionedAggregateOutputs(t *testing.T) {
+	env, m := setupMergeEnv(t)
+	paths := sessionPathsForEnv(env, m.SessionID)
+	alice := filepath.Join(paths.TranscriptsRawDir, "alice.json")
+	writeFile(t, alice, `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.TranscriptsRawDir, "removed.json"), `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
+	writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
+	setVersionedTranscribeMergeState(t, m, manifest.StatusSucceeded, map[string]string{
+		"alice": "transcripts/raw/alice.json",
+	})
+	fake := &seriatim.FakeRunner{}
+	env.Seriatim = fake
+
+	if _, err := (mergeStage{}).Run(context.Background(), env, m); err != nil {
+		t.Fatalf("merge.Run() error = %v", err)
+	}
+	if len(fake.NormalizeRequests) != 1 || fake.NormalizeRequests[0].InputTranscriptPath != alice {
+		t.Fatalf("normalize requests = %#v, want only current alice", fake.NormalizeRequests)
+	}
+}
+
+func TestMergeStageRejectsIncompleteVersionedAggregateOutputs(t *testing.T) {
+	env, m := setupMergeEnv(t)
+	paths := sessionPathsForEnv(env, m.SessionID)
+	writeFile(t, filepath.Join(paths.TranscriptsRawDir, "alice.json"), `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.TranscriptsRawDir, "bob.json"), `{"segments":[]}`)
+	writeFile(t, filepath.Join(paths.InputsDir, "speakers.yml"), "match: []\n")
+	writeFile(t, filepath.Join(paths.InputsDir, "autocorrect.yml"), "rules: []\n")
+	setVersionedTranscribeMergeState(t, m, manifest.StatusSucceeded, map[string]string{
+		"alice": "transcripts/raw/alice.json",
+		"bob":   "transcripts/raw/bob.json",
+	})
+	m.Stages["transcribe"].Outputs = m.Stages["transcribe"].Outputs[:1]
+	env.Seriatim = &seriatim.FakeRunner{}
+
+	_, err := (mergeStage{}).Run(context.Background(), env, m)
+	if err == nil || !strings.Contains(err.Error(), "incomplete for speakers: bob") {
+		t.Fatalf("merge.Run() error = %v, want incomplete aggregate output rejection", err)
+	}
+}
+
 func TestMergeStageResolvesWorkspaceQualifiedManifestOutputsWithoutDuplication(t *testing.T) {
 	env, m := setupMergeEnvWithRelativeWorkspaceRoot(t)
 	paths := sessionPathsForEnv(env, m.SessionID)
@@ -414,3 +479,44 @@ func setupMergeEnvWithWorkspace(t *testing.T, workspace string) (*Env, *manifest
 		ArtifactStore: store,
 	}, manifest.New("2026-05-03", time.Now().UTC())
 }
+
+func setVersionedTranscribeMergeState(
+	t *testing.T,
+	m *manifest.Manifest,
+	status manifest.StageStatus,
+	outputs map[string]string,
+) {
+	t.Helper()
+	if m.RunID == "" {
+		m.RunID = "20260503T010203Z-a1b2c3d4"
+	}
+	aggregate := make([]manifest.ArtifactRecord, 0, len(outputs))
+	records := make(map[string]manifest.TranscribeSpeakerRecord, len(outputs))
+	for speaker, outputPath := range outputs {
+		output := manifest.ArtifactRecord{
+			Kind: "transcript_raw", LocalPath: outputPath,
+			ProducerRunID: m.RunID, Checksum: strings.Repeat("b", 64),
+		}
+		aggregate = append(aggregate, output)
+		records[speaker] = manifest.TranscribeSpeakerRecord{
+			FingerprintVersion: manifest.TranscribeFingerprintContractVersion,
+			Fingerprint:        strings.Repeat("c", 64),
+			AudioPath:          "audio/" + speaker + ".flac",
+			AudioSHA256:        strings.Repeat("a", 64),
+			AudioSize:          1,
+			Output:             output,
+			OutputSize:         1,
+			ProducerRunID:      m.RunID,
+		}
+	}
+	slices.SortFunc(aggregate, func(a, b manifest.ArtifactRecord) int {
+		return strings.Compare(a.LocalPath, b.LocalPath)
+	})
+	m.MarkStageSucceeded("transcribe", time.Now().UTC(), aggregate)
+	record := m.Stages["transcribe"]
+	record.TranscribeStateVersion = manifest.TranscribeStateContractVersion
+	record.TranscribeSpeakers = records
+	if status != manifest.StatusSucceeded {
+		m.MarkStageFailed("transcribe", time.Now().UTC(), "incomplete")
+	}
+}
--- a/internal/stage/transcribe_checkpoint_test.go
+++ b/internal/stage/transcribe_checkpoint_test.go
@@ -271,6 +271,28 @@ func TestTranscribeRebuildsAggregateFromCurrentSpeakersWithoutRequests(t *testin
 	}
 }

+func TestTranscribeReuseDoesNotDependOnRunDirectory(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)
+	}
+	if _, err := (transcribeStage{}).Run(context.Background(), env, m); err != nil {
+		t.Fatalf("Run() error = %v", err)
+	}
+	paths := sessionPathsForEnv(env, m.SessionID)
+	if err := os.RemoveAll(paths.RunsDir); err != nil {
+		t.Fatalf("remove run directory: %v", err)
+	}
+	if _, err := os.Stat(paths.TranscriptsRawDir + "/alice.json"); err != nil {
+		t.Fatalf("canonical transcript missing after run cleanup: %v", err)
+	}
+	validation, err := (transcribeStage{}).ValidateResume(context.Background(), env, m)
+	if err != nil || !validation.Resumable {
+		t.Fatalf("ValidateResume() = %#v, %v, want canonical reuse", validation, err)
+	}
+}
+
 type failingTranscribeManifestStore struct {
 	manifest.Store
 	Saves  atomic.Int64