Prepare the v2.2.1 release

docs/internal/manifest.md

@@ -193,6 +193,25 @@ establish a current immutable snapshot.
 - writes through a sibling temporary file and atomic replacement visibility;
 - updates `updated_at` on save.
 
+### Checkpoint persistence boundary
+
+Analyze and transcribe share one private checkpoint operation in
+`internal/stage`. Their typed callers still decide when to checkpoint, validate
+their collections, populate stage-owned fields, and serialize access. The
+shared operation copies the manifest and its checkpoint-relevant mutable
+collections, applies the typed update to the copied target record, and calls
+the existing `manifest.Store` exactly once. Only after that save succeeds does
+it promote the saved target record and manifest `updated_at` into live state;
+it preserves an existing target record's identity and leaves unrelated live
+stage records in place.
+
+The copy boundary includes stage records, input and artifact slices, aggregate
+output/log/configuration slices, and the typed analyze and transcribe
+collections that callbacks or `LocalStore.Save` may modify. Metadata and other
+pointer fields remain shared read-only because neither checkpoint callback nor
+the store mutates them. This is a bounded execution helper, not a general
+manifest clone or a transaction across transcript/artifact files.
+
 Manifest replacement provides atomic visibility, not a crash- or power-loss
 durability guarantee. Callers reload after a returned save error before retrying.
 A transcribe or analyze checkpoint error leaves the prior manifest collection

docs/internal/stage-analyze.md

@@ -115,7 +115,11 @@ Supported source families:
 - before the first adapter call, checkpoints the reconciled starting collection
   when it removed configuration records or cleared obsolete authority. After
   each successful artifact, checkpoints the complete collection before making
-  that artifact available to the next scheduled dependent.
+  that artifact available to the next scheduled dependent. Analyze supplies
+  its typed update to the shared
+  [checkpoint persistence boundary](manifest.md#checkpoint-persistence-boundary)
+  and refreshes its working and committed collections only after the save
+  succeeds.
 - keeps adapter output in the invocation's run-local analyze directory until
   it is a safe, non-empty, bounded regular file with a calculated checksum and
   complete output contract. Canonical replacement uses the shared atomic file

docs/internal/stage-transcribe.md

@@ -35,9 +35,10 @@ Generate raw per-speaker transcripts from prepared audio using WhisperX.
 - output path returned by adapter must match requested output path.
 - an empty adapter result path means the requested path; adapters cannot select
   an alternate destination.
-- candidate collection construction, manifest save, and installation into live
+- candidate collection construction, the shared
-  state are serialized. Expensive hashing and HTTP work remain outside that
+  [checkpoint persistence boundary](manifest.md#checkpoint-persistence-boundary),
-  critical section.
+  and installation into live state are serialized by transcribe's mutex.
+  Expensive hashing and HTTP work remain outside that critical section.
 - a canonical file has no completion authority until its manifest checkpoint
   succeeds. A save failure preserves the last successfully committed
   collection for ordinary retry.

docs/releases/v2.2.1.md

@@ -0,0 +1,35 @@
+# Narratio v2.2.1
+
+This maintenance release consolidates checkpoint persistence for transcription
+and analysis without changing their operator-facing behavior.
+
+## Summary
+
+Transcribe and analyze now share one internal operation for preparing isolated
+checkpoint candidates, saving them, and promoting successfully saved progress
+into live state. Each stage retains its own progress model and recovery policy.
+
+## Compatibility
+
+CLI commands, configuration, manifest schemas, artifact formats, and external
+integration contracts are unchanged. Speaker concurrency, artifact selection,
+force behavior, and ordinary resume semantics are preserved. No dependency
+upgrade or new external-tool version requirement is introduced.
+
+## Upgrade
+
+No special upgrade action or configuration migration is required when upgrading
+from v2.2.0. See the [v2.2.0 release notes](v2.2.0.md) when upgrading from an
+earlier version for the original speaker-checkpoint upgrade considerations.
+
+## Changes
+
+- Replace duplicate analyze/transcribe save-and-promote sequences with a
+  shared, private checkpoint helper.
+- Isolate checkpoint collections from live state, including analyze records
+  normalized during a transcription checkpoint save.
+- Preserve transcription's caller-owned locking and failure handling, and
+  analysis's working/committed collections and rollback behavior.
+- Add focused checkpoint integrity tests while retaining stage and application
+  recovery coverage.
+- Document the shared [checkpoint persistence boundary](../internal/manifest.md#checkpoint-persistence-boundary).

docs/roadmap/implementation.md

@@ -1,416 +1,316 @@
-# Implementation Plan: Speaker Checkpoints and Replayable Uploads
+# Implementation Plan: Shared Stage Checkpoint Mechanics
 
-Status: Implementation complete; Stages 1-7 complete.
+Status: Complete.
 
-## Scope and Instructions
+## Scope and Authority
+
+Implement the agreed follow-up to v2.2.0: consolidate the thin checkpoint
+mechanism shared by transcribe and analyze without combining their progress
+models or changing operator behavior. There is no separate consolidation
+roadmap in this checkout. This plan uses the accompanying discussion as its
+scope: persistence is already centralized; candidate isolation and
+save-before-promotion are the duplication to remove. The completed
+[transcription roadmap](transcribe-checkpoints-and-replay.md) provides background,
+not authorization to reimplement transcription or HTTP replay. The preceding
+completed implementation plan remains in Git history.
 
-Implement [the feature roadmap](transcribe-checkpoints-and-replay.md). This plan
-replaces the completed Notarius-profile plan; that work remains in Git history.
 Complete the numbered stages in order, one bounded implementation prompt per
 stage for gpt-5.6-terra. Record completion and actual verification results here.
-Intermediate stages are not a finished feature and must not be released alone.
+This planning pass changes documentation only; the steps below are for the
+subsequent implementation sprint.
 
-Before working, read `AGENTS.md`, `docs/development.md`, all three policies in
+Before each stage, read `AGENTS.md`, `docs/development.md`, and all three
-`docs/policy/`, the roadmap, and the stage-specific preparation below. Use the
+documents in `docs/policy/`: architecture, documentation, and testing. Use
-maintained Go 1.26 toolchain. Preserve operator choices, speaker filenames,
+maintained Go 1.26. Prefer graph tools for code discovery; inspect current
-bounded concurrency, streaming, and existing credentials/settings. Do not add
+files if their index is stale. Read the stage-specific preparation below,
-new commands, dependencies, service requirements, arbitrary speaker limits,
+rather than unrelated historical roadmaps.
-strict transcript schemas, or a generic checkpoint framework. No release or
-upstream service work is part of this plan.
 
-## Settled Design Decisions
+## Target and Scope Boundaries
 
-### Progress Representation and Recovery
+The existing `internal/manifest.Store` and `LocalStore` remain the session
+manifest persistence interface and implementation. Encoding, schemas,
+validation, normalization, and atomic file replacement remain unchanged.
+There is no new store, journal, file format, dependency, command, configuration,
+or migration.
 
-Add optional `transcribe_state_version` (version 1) and `transcribe_speakers`
+One private helper in `internal/stage` owns candidate preparation, calling the
-to `manifest.StageRecord`. The latter is a map keyed by the existing derived
+existing store, and promoting the selected stage record only after successful
-speaker ID, containing completed results only. Absence means that work must be
+save. The existing typed analyze/transcribe checkpoint methods become thin
-done; a separate pending/running/failed state machine is unnecessary. Aggregate
+callers. This is execution plumbing: keep it beside the callers rather than
-stage status/error remains authoritative for invocation failure. A versioned
+moving orchestration into the storage package or exporting a reusable API.
-empty collection is meaningful: no earlier completions are reusable.
 
-Use a typed `TranscribeSpeakerRecord` with fingerprint version/digest,
+Keep these responsibilities stage-owned:
-session-relative audio path, audio SHA-256 and size, one `ArtifactRecord` output
-of kind `transcript_raw`, output size, and producer run ID. The output carries
-its session-relative canonical path and SHA-256. No new transcript source IDs,
-synthetic schema contracts, timestamps, or historical attempt collections are
-needed. Existing aggregate diagnostics can retain attempt/status/duration data
-for work performed in the current invocation; do not fabricate that information
-for reused outputs.
 
-Keep the session-manifest schema identifier unchanged for this optional additive
+- Transcribe's completed-only speaker map, evidence checks, bounded concurrency,
-state; validate supported per-stage versions and retain strict decoding. New
+  mutex, checkpoint-failure latch, cancellation, pre-dispatch force reset,
-Narratio reads old manifests. Old binaries may reject the added fields under
+  output installation, and aggregate repair.
-their existing strict decoding; document this downgrade limitation.
+- Analyze's dispositions, selection, dependencies, catalog updates, working and
+  committed collections, failure reconciliation, invalidation, and outputs.
+- Each stage's decision whether/when to checkpoint, typed validation, contextual
+  errors, and post-save working-state updates.
+- Runner aggregate transitions, session locking, terminal failure saves, and
+  fixed dependent-stage invalidation.
 
-For pre-feature manifests with no speaker evidence, regenerate transcription
+Do not introduce a common subtask model, scheduler, transaction manager, global
-once when transcribe is next selected, even if the aggregate status succeeded.
+mutex, automatic retry, additional checkpoint, or batching policy. Do not
-Old records do not bind individual transcript results to the audio bytes they
+migrate publish/restore, planning clones, cleanup, or ordinary runner saves:
-used. Do not guess that association from current files, old `per_file` metadata,
+their operations are not the demonstrated duplication. Leave HTTP replay and
-or abandoned run directories. This is the conservative upgrade path, not a new
+all adapters unchanged.
-general migration system. No other stage or unselected range is forcibly run.
 
-### Reuse and Persistence
+## Selected Design
 
-Compute a versioned per-speaker fingerprint from speaker ID, session-relative
+### Shared Save-and-Promote Operation
-audio identity, actual audio SHA-256/size, and the existing transcribe semantic
-configuration fingerprint (URL and language). Do not include invocation ID,
-absolute workspace root, concurrency, retry settings, timeout, or credentials.
-Stream input hashing with cancellation; do not trust stored input checksums
-alone. Recheck input bytes before accepting a new result to catch changes
-observed during transcription; fail that job without checkpointing mismatched
-evidence. This is not a hostile-filesystem snapshot or concurrent-editor
-transaction guarantee and requires no extra audio copies or file locks.
 
-Validate reusable output identity, canonical location, JSON, checksum, and size.
+Add `internal/stage/checkpoint.go` with a private helper shaped as:
-Use existing confinement/file primitives. Missing or corrupt regular outputs
-become work; unsafe paths/non-regular conflicts follow existing safety errors,
-not recursive deletion. Missing required audio is an input error, not permission
-to reuse an unverifiable transcript. Removed inputs are removed from the current
-collection and aggregate outputs without deleting arbitrary files.
 
-One stage-owned mutex serializes candidate collection construction, manifest
+```go
-save, and installation of committed state. Save a detached candidate and only
+func checkpointStageCollection(
-then update live authority, following analyze's pattern. Workers must not mutate
+    ctx context.Context,
-the live manifest outside that serialized section. Keep expensive HTTP work
+    store manifest.Store,
-outside the mutex. Stop dispatch/acceptance on checkpoint failure, join workers
+    path string,
-before return, and preserve the last successfully committed collection.
+    live *manifest.Manifest,
+    stageName string,
+    update func(*manifest.StageRecord) error,
+) error
+```
 
-Before launching requests, durably checkpoint the reconciled collection. For
+Names are implementation choices, not public contracts. Use a function and
-force, that collection is empty: this prevents later ordinary retries from
+small copying helpers, not a factory/options/interface framework. The two
-resurrecting pre-force outputs. If this initial save fails, do not start requests
+existing typed checkpoint methods are the only production callers. They pass
-or overwrite outputs. Force again clears the current collection again; ordinary
+the already-resolved manifest path; the helper does not discover paths or
-retry retains replacements already checkpointed after the previous force.
+inspect configuration.
 
-Save each validated result at its canonical raw-transcript location before
+The helper performs this sequence:
-checkpointing it. Run-local files remain staging/diagnostics, not reuse
-dependencies. Bytes left by an unsuccessful checkpoint have no completion
-authority. A worker may checkpoint a completed result using the parent stage
-context even if a sibling canceled the work context; parent cancellation or a
-checkpoint failure may prevent that save. Guarantee retention of committed
-work, not all responses that happened to finish during cancellation.
 
-### Multipart Replay Mechanism
+1. Check required live/store inputs without mutation. Copy a checkpoint
+   candidate and select a copied target record. If absent, provide an empty
+   candidate record to the caller, without inferring its name, status, or time.
+2. Apply the supplied update to that candidate record only. On update error,
+   return without saving or promoting. The typed caller validates and fills
+   its version, collection, and stage-owned initialization/timestamps.
+3. Invoke `Store.Save` once with the supplied context. Preserve error identity
+   through `%w` and useful path context. On error, do not promote, retry,
+   reload, or synthesize aggregate failure.
+4. After success, install the saved target record and candidate manifest
+   `UpdatedAt` into live state. Preserve an existing target record's pointer
+   identity by assigning through it. Allocate missing live maps/records only
+   after success. Do not replace the entire live manifest or overwrite other
+   stage records. Copy mutable checkpoint collections on promotion so saved
+   candidates and caller working collections cannot mutate live progress.
 
-Replace the pipe/producer upload mechanism with a small pull-based multipart
+Callers serialize the whole operation. Transcribe's current mutex does so;
-body backed by the audio reader. Use `mime/multipart.Writer` to encode a bounded
+analyze executes artifacts sequentially. Do not add a lock to the helper or
-metadata prefix and suffix around the file payload: capture the prefix after
+promise support for concurrent independent manifest writers. Atomic replacement
-`CreateFormFile`, then encode the language field and closing boundary as the
+visibility does not become power-loss durability or a transaction covering
-suffix. Generate this framing once per application attempt. Each body combines
+transcript files.
-fresh readers for prefix, freshly opened audio, and suffix using `io.MultiReader`.
-No full-audio buffer or temporary multipart spool is needed.
 
-The initial body and `Request.GetBody` use the same factory/framing. Each owns
+### Bounded Candidate Isolation
-an independent audio handle. Give bodies idempotent Close and context-triggered
-closure (for example `context.AfterFunc`), without holding a mutex across a
-blocking Read. Read/Close coordination must release a blocked source and retain
-genuine source-read/close errors while not treating deliberate early closure as
-source failure. The attempt owns cleanup of all created bodies, including
-replays; reject new body creation after its context expires.
 
-Remove the old producer `Wait` dependency. Track body-local source errors with
+Factor `cloneManifestForTranscribeCheckpoint` into a shared checkpoint-scoped
-synchronization so a swallowed upload read error cannot become a successful
+copy helper in the new file. Use ordinary typed copies and existing
-transcript, and a deliberately closed original body cannot poison a successful
+`manifest.CloneAnalyzeArtifactCollection` and
-replay. Keep the outer retry policy and per-attempt context. Do not add replay
+`manifest.CloneTranscribeSpeakerCollection`. Do not introduce reflection
-counters to the user retry budget or depend on Go's exact internal retry count.
+cloning, JSON round trips, or a public general-purpose clone API.
-Preserve ordinary HTTP redirect policy; body replay can also enable Go's
-body-preserving redirects. Do not introduce a new redirect restriction.
 
-## Stage 1: Add Durable Speaker Completion State
+Copy the manifest value, stages map, stage records, and mutable collections
+touched by these updates or `LocalStore.Save`. In particular, Save normalizes
+analyze collections throughout the candidate, including on a transcribe
+checkpoint. Clone those collections for every stage so normalization cannot
+leak into live state on failure. Preserve the existing bounded copies of
+input/artifact and output/log/config slices. Retain nil entries for existing
+validation to reject, rather than silently repairing invalid manifests.
+
+This is not arbitrary deep isolation of metadata. Neither the two update
+callbacks nor the current Store mutates `map[string]any` metadata, so it may
+remain shared read-only. Document that boundary and inspect every current
+normalization write. Use existing typed clones for nested checkpoint evidence;
+do not restrict legal metadata values or alter their Go/JSON representations
+to simplify copying.
+
+### Differences to Preserve
+
+- Analyze still requires an existing stage record. Its callback changes only
+  analyze version/collection, not transcribe-style initialization or timestamp
+  policy. It refreshes working and committed collections only after success.
+- Transcribe retains missing-record initialization for direct stage calls:
+  its typed caller supplies name, running status, and initial timestamps when
+  needed. Its callback retains the existing stage `UpdatedAt` update. Copying
+  an existing record must not clear aggregate fields or invoke lifecycle
+  transitions.
+- Existing transcribe live-record initialization is incomplete on the unusual
+  missing-record path; promoting the actual successfully saved candidate makes
+  its identity agree with disk. This does not add a new missing-record
+  rejection or change ordinary runner status transitions.
+- Analyze rollback/failure handling and transcribe failure latching remain in
+  their execution owners. The shared operation never invents failed items.
+- Preserve starting-checkpoint elision, force/reset semantics, empty versioned
+  collections, evidence and producer identity, legacy transcription handling,
+  supported zero-sized audio, and aggregate repair without HTTP calls.
+- Do not freeze exact error wording, wall-clock values, copy choreography, or
+  iteration order into new contracts. No new schema, limits, filename rules,
+  statuses, or compatibility guarantees are authorized.
+
+## Stage 1: Add the Shared Checkpoint Operation
 
 Status: Complete.
 
-Completion marker (2026-09-21): Added optional versioned transcribe-owned
+Completion marker (2026-09-22): Added the private stage-level checkpoint
-per-speaker completion records, strict load/save validation, isolated collection
+save-and-promote helper and bounded candidate copying without migrating either
-cloning, legacy aggregate-only compatibility, and lifecycle-preservation
+production caller. Focused real-store coverage proves live isolation through
-coverage. Verified with `go test ./internal/manifest` and `git diff --check`.
+updates, validation, save-time normalization and timestamps, save failures,
+missing-target creation, pointer-preserving promotion, unrelated aggregate
+preservation, caller-copy isolation, context/error identity, and exact one-save
+behavior. Verified with `go test ./internal/manifest ./internal/stage`,
+`go test -race ./internal/stage`, and `git diff --check`.
 
-Preparation: `internal/manifest/manifest.go`, `store.go`, `analyze_state.go`,
+Preparation: `internal/manifest/store.go`, `manifest.go`, `analyze_state.go`,
-their tests, and `docs/internal/manifest.md`.
+`transcribe_state.go`; `internal/stage/analyze_execution_state.go`,
+`transcribe_execution_state.go`; `docs/internal/manifest.md`.
 
-1. Implement the additive fields, record type, version constants, validation,
+1. Add the private helper and bounded candidate copying described above.
-   and deep-copy helper described above in a focused transcribe state file.
+   Leave existing production checkpoint paths unchanged until their migration
-2. Validate ownership by the transcribe stage, supported versions, required
+   stages. Do not duplicate manifest record-validation rules.
-   completed-record evidence, output kind, safe relative paths, SHA-256 values,
+2. Add focused `internal/stage/checkpoint_test.go` coverage with real
-   sizes, and producer identity consistency. Use existing speaker IDs as-is;
+   `manifest.LocalStore` and temporary directories for successful save/reload.
-   do not apply analysis-artifact key regexes or copy unrelated analysis limits.
+   Use a small Store wrapper only to observe/fail the persistence boundary.
-3. Integrate validation with both load and save. Preserve speaker state through
+3. Prove live progress remains old during Save, matches saved progress after
-   running/failed/stale transitions and aggregate result clearing. Generic stage
+   success, and does not change on update, validation, or save failure. Include
-   invalidation does not discard checkpoints; reconciliation checks their data.
+   both typed collections and a transcribe checkpoint alongside analyze
-4. Test round-trip populated and versioned-empty collections, old manifests
+   records subject to save-time normalization. A failure test must exercise
-   without fields, malformed evidence, deep-copy isolation, and lifecycle
+   candidate timestamp/normalization mutation before returning an error, not
-   preservation. Avoid imposing JSON formatting or map order as contracts.
+   merely fail before any work.
+4. Prove unrelated stage/aggregate data survives, existing target references
+   observe updates, missing-target creation is invisible on failure, and later
+   caller collection edits do not mutate committed records. Preserve context
+   and error identity. Test the supported copy/write boundary, not arbitrary
+   malicious Store mutation.
 
-Verify: `go test ./internal/manifest`.
+Verify: `go test ./internal/manifest ./internal/stage` and
-
-Done when durable storage can represent partial completion independently of
-aggregate status, without activating new stage behavior yet.
-
-## Stage 2: Implement Read-Only Speaker Reconciliation
-
-Status: Complete.
-
-Completion marker (2026-09-21): Extracted shared deterministic transcribe job
-construction and added read-only, cancellation-aware per-speaker input and
-output reconciliation. Covered unchanged reuse, semantic and operational
-settings, changed/added/removed inputs, legacy and forced work, missing/corrupt
-outputs, unsafe evidence, and missing audio without connecting runtime skip or
-checkpoint behavior. Verified with
-`go test ./internal/stage ./internal/manifest` and `git diff --check`.
-
-Preparation: `internal/stage/transcribe.go`, `semantic_contracts_initial.go`,
-`analyze_reconciliation.go`, `extract_resume.go`, their tests, and the artifact
-path/integrity helpers they use.
-
-1. Extract the existing audio discovery/job construction into a shared,
-   transcribe-local reconciliation path. Preserve current input selection and
-   speaker identity behavior. Build sorted jobs and classify completed results
-   against the current input set and evidence defined above.
-2. Add pure fingerprint construction and streamed input/output inspection.
-   Resolve manifest input paths with existing session-relative/absolute reading
-   helpers; do not change input precedence or assume the current working directory
-   is the session root. Newly written checkpoint paths are always relative.
-3. Return current records and work-to-do without writing files or manifest state.
-   Force selects all work and an empty starting collection; old unversioned
-   transcribe state supplies no reusable records. Do not connect aggregate skip
-   validation until execution can produce the new evidence in Stage 3.
-4. Test unchanged reuse; one changed audio among unchanged speakers; URL/language
-   changes; operational-setting exclusions; added/removed speakers; missing and
-   corrupt regular outputs; unsafe record paths; missing audio; and force.
-   Prefer compact tables and real temporary files over mocked hashing.
-
-Verify: `go test ./internal/stage ./internal/manifest`.
-
-Done when one read-only decision path describes exactly which speakers need
-work, without altering existing runtime behavior prematurely.
-
-## Stage 3: Execute and Checkpoint Speakers Incrementally
-
-Status: Complete.
-
-Completion marker (2026-09-21): Connected reconciliation to bounded execution,
-durably checkpointed the starting and per-speaker collections through detached
-manifest candidates, preserved committed work across failures and cancellation,
-rebuilt deterministic aggregate outputs from reusable evidence, and added
-speaker-aware resume validation. Updated production-style fixtures and covered
-serialized concurrent saves, force/retry replacement, input rechecks, save
-failures, partial recovery, and aggregate reconstruction. Verified with
-`go test ./internal/stage ./internal/app`, `go test -race ./internal/stage`, and
-`git diff --check`.
-
-Preparation: `internal/stage/transcribe.go`, `analyze_execution_state.go`,
-`run_local.go`,
-`internal/app/runner.go`, and transcribe/checkpoint test fixtures.
-
-1. Wire reconciliation into Run, require the existing ManifestStore collaborator,
-   and checkpoint the starting collection before launching HTTP work. Implement
-   detached candidate saves and serialized live-state updates. Audit Store.Save's
-   normalization side effects so a failed candidate save cannot mutate shared
-   authority; use detached records/collections wherever it may write.
-2. Retain bounded errgroup execution and fail-fast cancellation. Check the work
-   context before beginning each queued request. After success, validate adapter
-   output path/JSON, recheck audio evidence, materialize canonical bytes, compute
-   output evidence, and checkpoint the new record. Check a checkpoint-failure
-   latch under the serialization lock before accepting further completions.
-3. Preserve committed successes on all error/cancellation exits, including
-   starting-save, canonical-write, per-speaker-save, and aggregate-save failures.
-   Return meaningful speaker/error context and let the runner retain its existing
-   terminal failure behavior. Do not report partial aggregate outputs as success.
-4. Assemble final outputs in deterministic speaker order from the reconciled
-   completed records, including reused speakers, with original producer IDs and
-   checksums. If all records are current but aggregate status is failed/running,
-   finish without WhisperX calls and rebuild the aggregate success.
-5. Implement `transcribeStage.ValidateResume` using the same read-only
-   reconciliation, so a succeeded stage cannot skip changed audio or invalid
-   transcripts. Retain the existing URL/language semantic check and runner
-   invalidation behavior; no whole-config digest invalidation.
-6. Adapt existing tests that require no canonical partial files: the new contract
-   permits checkpointed files but forbids aggregate success. Update fixtures to
-   provide real manifest stores and records rather than a production fallback
-   that silently disables checkpointing.
-7. Test A/B completion followed by C failure and D unattempted (concurrency 1),
-   reload and retry only C/D; concurrently finishing speakers without lost saves;
-   failure and cancellation after a checkpoint; save failure retaining earlier
-   authority; and force failure followed by ordinary retry without old-result reuse.
-
-Verify: `go test ./internal/stage ./internal/app` and
 `go test -race ./internal/stage`.
 
-Done when incremental progress and aggregate success work together safely under
+Done when the shared integrity boundary is tested without changing existing
-ordinary, concurrent, canceled, and forced execution.
+stage behavior. Temporary copying overlap is removed in Stage 3.
 
-## Stage 4: Verify Runner, Merge, and Durable Lifecycle Integration
+## Stage 2: Migrate Analyze
 
 Status: Complete.
 
-Completion marker (2026-09-22): Added real-persistence application coverage for
+Completion marker (2026-09-22): Replaced analyze's local candidate-copy,
-cross-invocation retry, interrupted running state, failed aggregate saves,
+save, and promotion choreography with the shared checkpoint operation while
-force/reset persistence, downstream invalidation, and one-time legacy upgrade.
+retaining its existing-record requirement, typed validation, path resolution,
-Versioned merge input discovery now requires aggregate success and an exact
+and post-save working and committed collection refresh. Existing recovery,
-match between current speaker evidence and declared aggregate outputs, while
+selection, invalidation, and rollback coverage required no duplication.
-legacy/manual directory fallback remains available. Restore coverage preserves
+Verified with `go test ./internal/stage ./internal/app ./internal/manifest`,
-session-relative speaker paths, leaves unpublished raw transcripts absent, and
+`go test -race ./internal/stage ./internal/app`, and `git diff --check`.
-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`,
+Preparation: `internal/stage/analyze_execution_state.go`,
-`analyze_checkpoint_recovery_test.go`, `runner_terminal_failure_test.go`,
+`analyze_checkpoint_test.go`, `analyze_failure_test.go`,
-`restore_manifest_paths.go`, `internal/stage/merge.go`, and existing publish,
+`internal/app/analyze_checkpoint_recovery_test.go`, and
-restore, and cleanup tests routed from `docs/internal/overview.md`.
+`docs/internal/stage-analyze.md`.
 
-1. Add application-level tests with real manifest persistence and fake WhisperX:
+1. Replace manual candidate copying, saving, and live promotion in analyze
-   failed stage across separate invocations; all speakers committed before a
+   with the shared helper. Retain typed validation, required-record checking,
-   failed aggregate save; interrupted running state; force/reset persistence;
+   the typed update callback, and committed/working collection refresh.
-   and unchanged upstream/succeeded downstream invalidation behavior. Inject
+2. Preserve starting-checkpoint elision, selected-artifact behavior, catalog
-   errors at durable boundaries, not every private method call.
+   sequencing, dependent invalidation on changed output, and rollback after
-2. Close `discoverRawTranscripts`' directory-scan fallback for versioned
+   checkpoint failure. Do not change the runner's terminal failure save or
-   transcribe state: incomplete aggregate state is not usable, and completed
+   joined errors.
-   state consumes only declared current aggregate outputs. Keep legacy/manual
+3. Run existing tests before adding coverage. Retain tests proving earlier
-   fallback behavior for manifests without the new state rather than banning
+   artifacts survive failure/cancellation, unsaved output bytes gain no
-   previously allowed standalone workflows. Test that partial files and removed
+   authority, selected/forced work preserves appropriate unrelated artifacts,
-   speaker files cannot leak into a versioned merge, including direct stage calls.
+   and interrupted runs recover across store reloads. Add only missing
-3. Audit manifest copies/projections, remote manifest publication, restored path
+   integration coverage; do not duplicate Stage 1's isolation matrix.
-   handling, and cleanup. Preserve relative speaker paths on restore and verify
-   they resolve under the destination session, not the producer machine. Reuse
-   existing confinement validation, not a second path-policy layer.
-4. Add focused round-trip coverage through existing publish/restore fixtures for
-   speaker records. Restored outputs absent from the published selection are
-   non-reusable locally; do not silently expand remote publication to every raw
-   speaker or upload audio as a new default. Removing run directories must not
-   remove canonical checkpointed transcripts or make reuse require run files.
-5. Verify old succeeded manifests are readable and rerun once when transcribe is
-   selected, then reuse normally. Keep old manifests usable for other commands
-   and bounded ranges. Do not automatically import old unrecorded JSON files.
 
-Verify: `go test ./internal/app ./internal/stage ./internal/manifest` and
+Verify: `go test ./internal/stage ./internal/app ./internal/manifest` and
-`go test -race ./internal/app`.
+`go test -race ./internal/stage ./internal/app`.
 
-Done when the new authority survives runner and storage lifecycles and cannot
+Done when analyze delegates persistence choreography while progress, artifact
-expose incomplete transcripts downstream.
+selection, and failure behavior remain unchanged.
 
-## Stage 5: Implement Replayable Streaming Multipart Bodies
+## Stage 3: Migrate Transcribe and Remove Duplicate Machinery
 
 Status: Complete.
 
-Completion marker (2026-09-22): Replaced the producer-goroutine/pipe upload
+Completion marker (2026-09-22): Delegated transcription's locked checkpoint
-with bounded standard-library multipart framing and independently opened,
+operation to the shared helper while retaining typed validation, direct-call
-pull-based streaming bodies shared by the initial request and `GetBody`.
+record initialization, stage timestamp updates, post-save working collection
-Bodies are tracked, idempotently closed, canceled with their attempt context,
+refresh, and the existing caller-owned mutex and failure latch. Removed the
-and retain genuine read/close failures without treating an intentionally
+duplicate production manifest clone; the read-only resume assertion now uses
-abandoned original as a replay failure. Added focused coverage for identical
+the shared bounded snapshot. Existing real-store, concurrency, force/retry,
-framing, partial-read restart, simultaneous handles, source failures, and
+interruption, aggregate-repair, and zero-sized-audio coverage passed unchanged.
-blocked-reader release. Verified with `go test ./internal/adapters/whisperx`,
+Verified with `go test ./internal/stage ./internal/app ./internal/manifest`,
-`go test -race ./internal/adapters/whisperx` (including ten repeated runs), and
+`go test -race ./internal/stage ./internal/app`, and `git diff --check`.
-`git diff --check`.
 
-Preparation: `internal/adapters/whisperx/http.go`, `http_test.go`,
+Preparation: `internal/stage/transcribe_execution_state.go`, `transcribe.go`,
-`docs/integrations/whisperx.md`, and the installed Go net/http Request/GetBody
+`transcribe_resume.go`, `transcribe_checkpoint_test.go`,
-contract. Keep this change inside the WhisperX adapter.
+`transcribe_reconciliation_test.go`,
+`internal/app/transcribe_checkpoint_recovery_test.go`, and
+`docs/internal/stage-transcribe.md`.
 
-1. Implement the pull-based framing/body factory described above, using standard
+1. Delegate `checkpointCollectionLocked` to the shared helper, retaining
-   multipart encoding rather than hand-escaped MIME headers. Add no new payload
+   typed initialization/update and successful working-collection refresh.
-   cap or upload-size setting. Preserve file-before-language wire semantics and
+   Keep the existing mutex across candidate creation, save, and promotion.
-   the existing form names.
+   Do not move HTTP/hashing work into that critical section or add a lock.
-2. Install the initial body and GetBody with the same per-attempt boundary.
+2. Preserve failure-latch checks and worker joining, the parent versus worker
-   Track and close all opened bodies even if a custom transport returns early;
+   context distinction, force's durable empty starting collection, and
-   make cleanup/context callbacks safe against overlapping reads and closes.
+   ordinary retry's reuse of committed forced replacements.
-   A GetBody failure must be returned, not converted into an empty upload.
+3. Remove `cloneManifestForTranscribeCheckpoint`. Its current use by a
-3. Replace pipe-specific Wait/error paths in doTranscribeAttempt with source-error
+   read-only resume test is not a reason to retain duplicate production code.
-   handling appropriate to pull bodies. Genuine source errors remain visible;
+   Use an independent test snapshot or the shared bounded copy where sufficient,
-   intentional closure during transport retry is not itself a source error.
+   without exporting a helper for tests.
-   Keep response bounds, JSON validation, output installation, retry
+4. Retain existing real-store and concurrency coverage: no lost completions;
-   classification/delay, cancellation, and application attempt reporting intact.
+   pre-dispatch save failure prevents requests/canonical writes; later failure
-4. Remove obsolete producer-only helpers rather than retaining two upload paths.
+   retains prior authority; interrupted and aggregate-save recovery; failed
-   Adapt tests tied to producer goroutines to assert streaming and resource
+   force followed by retry; run-directory independence; zero-sized audio;
-   release, which are the actual contracts.
+   removed/mismatched aggregate repair without additional WhisperX requests
-5. Add focused body tests: original/replayed bytes parse identically; replay
+   or removed-speaker merge input. Do not weaken these regression assertions.
-   after partial consumption starts from the beginning; simultaneous bodies use
-   independent handles; source failures propagate; Close/cancel releases a
-   blocked reader without deadlock. Use synthetic audio and gated readers.
 
-Verify: `go test ./internal/adapters/whisperx` and
+Verify: `go test ./internal/stage ./internal/app ./internal/manifest` and
-`go test -race ./internal/adapters/whisperx`.
+`go test -race ./internal/stage ./internal/app`.
 
-Done when replay is available without whole-file buffering or producer-lifetime
+Done when both stages use one save/promote implementation and one bounded
-ambiguity, and all existing HTTP behavior tests still pass.
+copy implementation, with their different models and synchronization intact.
 
-## Stage 6: Verify Replay Through the HTTP Adapter Boundary
+## Stage 4: Document, Review Scope, and Validate
 
 Status: Complete.
 
-Completion marker (2026-09-22): Added a deterministic transport regression
+Completion marker (2026-09-22): Documented the private shared checkpoint
-that abandons a partially consumed original request, obtains and parses a fresh
+persistence boundary and linked the stage guides back to it while retaining
-multipart replay from `GetBody`, and installs the returned transcript within one
+their separate analyze and transcribe progress policies. The refactor removed
-application attempt. Covered replay open/read failures, early responses,
+the two local candidate/save/promotion sequences and the transcription-only
-parent cancellation during replay, and disposal of every opened source, while
+bounded clone; the retained thin wrappers own typed validation, updates,
-proving an abandoned original does not poison a valid replay. Added a real HTTP
+working-state refresh, and their existing serialization and failure behavior.
-retry-exhaustion integration that durably retains an earlier speaker and resumes
+The collective test review retained focused helper integrity coverage plus
-only the uncommitted speakers. This controlled boundary verifies Narratio's
+stage/application recovery tests, without duplicating the helper's isolation
-replay contract, not recovery from every real proxy protocol failure. Verified
+matrix. A cumulative scope audit found no schema, validation-policy,
-with `go test ./internal/adapters/whisperx ./internal/app ./internal/stage`,
+scheduling, force, concurrency, output-selection, stage-order, adapter-call,
-`go test -race ./internal/adapters/whisperx ./internal/app ./internal/stage`,
+configuration, or public-interface change.
-ten repeated race-enabled adapter runs, and `git diff --check`.
 
-Preparation: Existing injectable HTTPClient/RoundTripper seam and local-server
+Verified with `go test ./...`, `go test -race ./...`, `go vet ./...`,
-tests in `internal/adapters/whisperx/http_test.go`; app WhisperX wiring tests.
+`go build ./...`, `go test ./internal/doccheck`,
-
-1. Add a deterministic transport-boundary regression: consume part of the
-   original request, close it as a transport would, call GetBody, consume and
-   parse the replay using the request Content-Type, then return valid JSON.
-   Verify exact audio/language, installed output, and one application attempt.
-   This must fail before replay support; merely checking non-nil GetBody is
-   insufficient. Do not assert Go's internal retry count or require a flaky
-   handcrafted HTTP/2 peer in the default suite.
-2. Cover replay-open/read failure, early response, parent cancellation during
-   replay, and disposal of all body handles. Reuse Stage 5 body coverage rather
-   than repeat every low-level case. Prove an intentionally abandoned original
-   body does not turn a valid replay response into failure.
-3. Retain local HTTP tests for application retries, non-retryable responses,
-   response limits, invalid JSON, and bounded streaming; exercise that an
-   exhausted HTTP failure still leaves earlier speaker checkpoints reusable
-   using the existing stage/app seams.
-4. Record the verification boundary honestly: controlled transport replay proves
-   Narratio's body contract, not recovery from every real proxy PROTOCOL_ERROR.
-
-Verify: `go test ./internal/adapters/whisperx ./internal/app ./internal/stage`
-and race tests for changed packages.
-
-Done when the replay defect is protected at the adapter boundary and the
-application retry/checkpoint behavior remains intact.
-
-## Stage 7: Document and Validate the Completed Feature
-
-Status: Complete.
-
-Completion marker (2026-09-22): Updated the canonical transcribe, merge,
-manifest, WhisperX, operations, and troubleshooting documentation for durable
-speaker reconciliation, serialized checkpoints, aggregate gating, durable
-force reset, legacy upgrade and downgrade limits, replayable streaming bodies,
-resource ownership, and the absence of an exactly-once remote guarantee.
-Reviewed the completed implementation against the feature roadmap and
-implementation-discipline guidance. Verified with `go test ./...`,
-`go test -race ./...`, `go vet ./...`, `go build ./...`,
-`go test ./internal/doccheck`,
 `go test ./internal/config -run '^TestExamplesLoadAndValidate$'`, and
-`git diff --check`.
+`git diff --check`; all passed.
 
-1. Update canonical transcribe and manifest internals for the collection,
+Preparation: `docs/internal/manifest.md`, `docs/internal/stage-transcribe.md`,
-   reconciliation, durable-reset force behavior, serialized saves, and aggregate
+`docs/internal/stage-analyze.md`, documentation/testing policies, and the
-   gate. Remove obsolete all-speakers-before-canonical-write descriptions.
+implementation diff against this plan's starting commit.
-2. Update the WhisperX contract for replayable bounded streaming, application
+
-   attempts versus transport replay, resource ownership, and the lack of an
+1. Update manifest internals to identify the shared checkpoint owner and copy/
-   exactly-once server guarantee. Do not claim the remote protocol issue is fixed.
+   save/promote boundary. Stage guides retain stage-specific policy and link
-3. Update operations/troubleshooting with ordinary resume versus force, durable
+   to that explanation instead of duplicating shared implementation details.
-   partial success, old-manifest one-time retranscription on selecting transcribe,
+   Do not rewrite immutable release notes or announce new operator behavior.
-   downgrade limitations, and private-model-change guidance. No new CLI flags
+2. Assess tests collectively. Consolidate genuinely duplicated persistence
-   or promises of automatic historical-file recovery. Keep detailed wire and
+   mechanics coverage, but retain stage/application recovery tests proving
-   storage contracts in their canonical owners.
+   correct integration. Avoid test-count/LOC quotas. Record a brief summary
-4. Review changes against the feature roadmap and the implementation-discipline
+   here of duplicated mechanisms removed and thin wrappers retained.
-   guidance. Do not elevate selected storage encoding, internal ordering, test
+3. Confirm no schema, validation policy, scheduling, force, concurrency, output
-   synchronization, or exact error wording into new product guarantees.
+   selection, stage order, adapter call, or configuration change slipped into
-5. Run the full repository gates below; document results and update completion
+   the refactor. Additional restrictions require a separate decision, not
-   statuses only after they pass. Leave release preparation for a separate task.
+   tests that retrospectively declare new policy.
+4. Run all gates below and record actual results. Mark this plan complete only
+   when both migrations and verification pass. Do not release, tag, push, delete
+   other roadmaps, or expand into a repository-wide cloning refactor.
 
 ```sh
 go test ./...
@@ -422,10 +322,12 @@ go test ./internal/config -run '^TestExamplesLoadAndValidate$'
 git diff --check
 ```
 
+Done when duplication is removed, recovery remains covered, documentation
+matches the implementation, and all gates pass.
+
 ## Open Questions
 
-None requiring user input. The agreed roadmap supplies the product decisions;
+None requiring a product decision. A private stage-level helper is selected
-this plan selects a completed-only manifest collection, conservative old-record
+because persistence already has an owner and the duplication is execution
-handling, and pull-based replay as implementation choices. No stronger remote
+plumbing. Model unification, library extraction, generalized transactions, and
-reliability guarantee, new operator restriction, or generic orchestration
+broader cloning cleanup remain out of scope.
-surface is proposed.

docs/roadmap/transcribe-checkpoints-and-replay.md

@@ -15,9 +15,9 @@ peer `PROTOCOL_ERROR` and an inability to replay the request body without
 `Request.GetBody`.
 
 These are related reliability improvements, not a promise to eliminate all
-service failures. This document owns the target state and scope; the
+service failures. This document records the target state and scope. The
-[implementation plan](implementation.md) owns ordered changes and concrete
+completed implementation sequence is retained in Git history; current behavior
-storage and replay decisions.
+is documented in the canonical guides linked below.
 
 ## Pre-implementation Limitations
 

internal/stage/analyze_execution_state.go

@@ -136,28 +136,26 @@ func (state *analyzeExecutionState) checkpointCollection(
 		return fmt.Errorf("validate analyze artifact collection: %w", err)
 	}
 
-	candidate := *m
-	candidate.Stages = make(map[string]*manifest.StageRecord, len(m.Stages))
-	for name, record := range m.Stages {
-		candidate.Stages[name] = record
-	}
-	analyzeRecord := *m.Stages["analyze"]
-	analyzeRecord.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
-	analyzeRecord.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(records)
-	candidate.Stages["analyze"] = &analyzeRecord
-
 	manifestPath := artifacts.SessionManifestPathForCampaign(
 		state.execution.Paths.WorkspaceRoot,
 		m.Campaign,
 		m.SessionID,
 	)
-	if err := state.execution.Env.ManifestStore.Save(ctx, manifestPath, &candidate); err != nil {
+	if err := checkpointStageCollection(
-		return fmt.Errorf("save session manifest %q: %w", manifestPath, err)
+		ctx,
+		state.execution.Env.ManifestStore,
+		manifestPath,
+		m,
+		"analyze",
+		func(record *manifest.StageRecord) error {
+			record.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
+			record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(records)
+			return nil
+		},
+	); err != nil {
+		return err
 	}
 	liveRecord := m.Stages["analyze"]
-	liveRecord.AnalyzeStateVersion = candidate.Stages["analyze"].AnalyzeStateVersion
-	liveRecord.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(candidate.Stages["analyze"].AnalyzeArtifacts)
-	m.UpdatedAt = candidate.UpdatedAt
 	state.records = manifest.CloneAnalyzeArtifactCollection(liveRecord.AnalyzeArtifacts)
 	state.committedRecords = manifest.CloneAnalyzeArtifactCollection(liveRecord.AnalyzeArtifacts)
 	return nil

internal/stage/checkpoint.go

@@ -0,0 +1,99 @@
+package stage
+
+import (
+	"context"
+	"fmt"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+// checkpointStageCollection isolates one stage-owned collection update from
+// live authority until the existing manifest store reports a successful save.
+// Callers own serialization and all stage-specific validation and policy.
+func checkpointStageCollection(
+	ctx context.Context,
+	store manifest.Store,
+	path string,
+	live *manifest.Manifest,
+	stageName string,
+	update func(*manifest.StageRecord) error,
+) error {
+	if live == nil {
+		return fmt.Errorf("session manifest is required")
+	}
+	if store == nil {
+		return fmt.Errorf("manifest store is required")
+	}
+	if update == nil {
+		return fmt.Errorf("checkpoint update is required")
+	}
+
+	candidate := cloneManifestForCheckpoint(live)
+	if candidate.Stages == nil {
+		candidate.Stages = make(map[string]*manifest.StageRecord)
+	}
+	target, exists := candidate.Stages[stageName]
+	if !exists {
+		target = &manifest.StageRecord{}
+		candidate.Stages[stageName] = target
+	} else if target == nil {
+		// Retain the invalid nil entry for Store.Save to reject rather than
+		// silently repairing manifest state. The detached record only gives the
+		// typed update a safe value to operate on before that validation.
+		target = &manifest.StageRecord{}
+	}
+	if err := update(target); err != nil {
+		return fmt.Errorf("update stage %q checkpoint: %w", stageName, err)
+	}
+	if err := store.Save(ctx, path, candidate); err != nil {
+		return fmt.Errorf("save session manifest %q: %w", path, err)
+	}
+
+	savedRecord := cloneStageRecordForCheckpoint(candidate.Stages[stageName])
+	if savedRecord == nil {
+		return fmt.Errorf("saved stage %q checkpoint record is nil", stageName)
+	}
+	if live.Stages != nil {
+		if current := live.Stages[stageName]; current != nil {
+			*current = *savedRecord
+		} else {
+			live.Stages[stageName] = savedRecord
+		}
+	} else {
+		live.Stages = map[string]*manifest.StageRecord{stageName: savedRecord}
+	}
+	live.UpdatedAt = candidate.UpdatedAt
+	return nil
+}
+
+// cloneManifestForCheckpoint copies the bounded mutable state touched by the
+// two checkpoint callbacks or by manifest.LocalStore.Save. Metadata maps and
+// other pointer fields remain shared read-only because neither boundary mutates
+// them.
+func cloneManifestForCheckpoint(source *manifest.Manifest) *manifest.Manifest {
+	clone := *source
+	clone.Inputs = append([]manifest.InputRecord(nil), source.Inputs...)
+	clone.Artifacts = append([]manifest.ArtifactRecord(nil), source.Artifacts...)
+	if source.Stages == nil {
+		clone.Stages = nil
+		return &clone
+	}
+	clone.Stages = make(map[string]*manifest.StageRecord, len(source.Stages))
+	for name, sourceRecord := range source.Stages {
+		clone.Stages[name] = cloneStageRecordForCheckpoint(sourceRecord)
+	}
+	return &clone
+}
+
+func cloneStageRecordForCheckpoint(source *manifest.StageRecord) *manifest.StageRecord {
+	if source == nil {
+		return nil
+	}
+	clone := *source
+	clone.Outputs = append([]manifest.ArtifactRecord(nil), source.Outputs...)
+	clone.Logs = append([]string(nil), source.Logs...)
+	clone.GeneratedConfigs = append([]string(nil), source.GeneratedConfigs...)
+	clone.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(source.TranscribeSpeakers)
+	clone.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(source.AnalyzeArtifacts)
+	return &clone
+}

internal/stage/checkpoint_test.go

@@ -0,0 +1,327 @@
+package stage
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"path/filepath"
+	"reflect"
+	"slices"
+	"strings"
+	"testing"
+	"time"
+
+	"gitea.maximumdirect.net/eric/narratio/internal/manifest"
+)
+
+func TestCheckpointStageCollectionSavesThenPromotesTarget(t *testing.T) {
+	now := time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC)
+	live := checkpointTestManifest(now)
+	originalAnalyze := live.Stages["analyze"]
+	originalTranscribe := live.Stages["transcribe"]
+	originalMerge := live.Stages["merge"]
+	replacement := map[string]manifest.AnalyzeArtifactRecord{
+		"result": checkpointAnalyzeRecord(now.Add(time.Minute), []string{"zeta", "alpha"}),
+	}
+	path := filepath.Join(t.TempDir(), "manifest.json")
+	base := &manifest.LocalStore{}
+	store := &checkpointBoundaryStore{
+		Store: base,
+		BeforeSave: func(candidate *manifest.Manifest) {
+			if live.Stages["analyze"] != originalAnalyze || len(originalAnalyze.AnalyzeArtifacts) != 1 {
+				t.Fatal("live analyze authority changed before save")
+			}
+			if candidate.Stages["analyze"] == originalAnalyze || candidate.Stages["transcribe"] == originalTranscribe {
+				t.Fatal("checkpoint candidate shares stage records with live manifest")
+			}
+		},
+	}
+
+	err := checkpointStageCollection(context.Background(), store, path, live, "analyze", func(record *manifest.StageRecord) error {
+		record.AnalyzeStateVersion = manifest.AnalyzeStateContractVersion
+		record.AnalyzeArtifacts = replacement
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("checkpointStageCollection() error = %v", err)
+	}
+	if store.Calls != 1 {
+		t.Fatalf("Save calls = %d, want 1", store.Calls)
+	}
+	if live.Stages["analyze"] != originalAnalyze {
+		t.Fatal("existing target record pointer identity changed")
+	}
+	if live.Stages["transcribe"] != originalTranscribe || live.Stages["merge"] != originalMerge {
+		t.Fatal("unrelated stage record identity changed")
+	}
+	if originalAnalyze.Status != manifest.StatusRunning || originalAnalyze.Outputs[0].LocalPath != "artifacts/aggregate.md" {
+		t.Fatalf("aggregate analyze data changed = %#v", originalAnalyze)
+	}
+	if got := originalAnalyze.AnalyzeArtifacts["result"].Dependencies; !slices.Equal(got, []string{"alpha", "zeta"}) {
+		t.Fatalf("promoted dependencies = %#v, want normalized saved state", got)
+	}
+	if !live.UpdatedAt.After(now) {
+		t.Fatalf("live UpdatedAt = %v, want save timestamp after %v", live.UpdatedAt, now)
+	}
+
+	durable, err := base.Load(context.Background(), path)
+	if err != nil {
+		t.Fatalf("Load() error = %v", err)
+	}
+	if !reflect.DeepEqual(durable.Stages["analyze"], live.Stages["analyze"]) {
+		t.Fatalf("durable analyze = %#v, live = %#v", durable.Stages["analyze"], live.Stages["analyze"])
+	}
+	if !reflect.DeepEqual(durable.Stages["transcribe"], originalTranscribe) || !reflect.DeepEqual(durable.Stages["merge"], originalMerge) {
+		t.Fatal("unrelated durable stage data changed")
+	}
+
+	changed := replacement["result"]
+	changed.Status = manifest.AnalyzeArtifactFailed
+	changed.Dependencies[0] = "changed"
+	replacement["result"] = changed
+	if got := live.Stages["analyze"].AnalyzeArtifacts["result"]; got.Status != manifest.AnalyzeArtifactMissing || got.Dependencies[0] != "alpha" {
+		t.Fatalf("caller edits mutated committed record = %#v", got)
+	}
+}
+
+func TestCheckpointStageCollectionFailureLeavesLiveAuthorityUnchanged(t *testing.T) {
+	t.Run("update", func(t *testing.T) {
+		live := checkpointTestManifest(time.Now().UTC())
+		before := checkpointManifestJSON(t, live)
+		cause := errors.New("update failed")
+		store := &checkpointBoundaryStore{Store: &manifest.LocalStore{}}
+		err := checkpointStageCollection(context.Background(), store, filepath.Join(t.TempDir(), "manifest.json"), live, "analyze", func(record *manifest.StageRecord) error {
+			record.AnalyzeArtifacts = map[string]manifest.AnalyzeArtifactRecord{}
+			return cause
+		})
+		if !errors.Is(err, cause) {
+			t.Fatalf("error = %v, want update cause", err)
+		}
+		if store.Calls != 0 {
+			t.Fatalf("Save calls = %d, want 0", store.Calls)
+		}
+		assertCheckpointManifestJSON(t, live, before)
+	})
+
+	t.Run("validation", func(t *testing.T) {
+		live := checkpointTestManifest(time.Now().UTC())
+		before := checkpointManifestJSON(t, live)
+		store := &checkpointBoundaryStore{Store: &manifest.LocalStore{}}
+		path := filepath.Join(t.TempDir(), "manifest.json")
+		err := checkpointStageCollection(context.Background(), store, path, live, "analyze", func(record *manifest.StageRecord) error {
+			record.Name = "wrong-stage"
+			return nil
+		})
+		if err == nil {
+			t.Fatal("error = nil, want manifest validation failure")
+		}
+		if store.Calls != 1 {
+			t.Fatalf("Save calls = %d, want 1", store.Calls)
+		}
+		assertCheckpointManifestJSON(t, live, before)
+	})
+
+	t.Run("save after candidate mutation", func(t *testing.T) {
+		now := time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC)
+		live := checkpointTestManifest(now)
+		analyze := live.Stages["analyze"].AnalyzeArtifacts["result"]
+		analyze.Dependencies = []string{"zeta", "alpha"}
+		live.Stages["analyze"].AnalyzeArtifacts["result"] = analyze
+		before := checkpointManifestJSON(t, live)
+		cause := errors.New("save failed after mutation")
+		store := &checkpointBoundaryStore{Store: &manifest.LocalStore{}, ReturnErr: cause}
+		path := filepath.Join(t.TempDir(), "manifest.json")
+		replacement := checkpointTranscribeCollection("bob", "run-2")
+		err := checkpointStageCollection(context.Background(), store, path, live, "transcribe", func(record *manifest.StageRecord) error {
+			record.TranscribeStateVersion = manifest.TranscribeStateContractVersion
+			record.TranscribeSpeakers = replacement
+			return nil
+		})
+		if !errors.Is(err, cause) {
+			t.Fatalf("error = %v, want save cause", err)
+		}
+		if !strings.Contains(err.Error(), path) {
+			t.Fatalf("error = %q, want manifest path context", err)
+		}
+		if store.Calls != 1 || store.SavedCandidate == nil {
+			t.Fatalf("Save calls/candidate = %d/%p, want one observed candidate", store.Calls, store.SavedCandidate)
+		}
+		if !store.SavedCandidate.UpdatedAt.After(now) {
+			t.Fatalf("candidate UpdatedAt = %v, want Store.Save mutation", store.SavedCandidate.UpdatedAt)
+		}
+		if got := store.SavedCandidate.Stages["analyze"].AnalyzeArtifacts["result"].Dependencies; !slices.Equal(got, []string{"alpha", "zeta"}) {
+			t.Fatalf("candidate dependencies = %#v, want save-time normalization", got)
+		}
+		if got := live.Stages["analyze"].AnalyzeArtifacts["result"].Dependencies; !slices.Equal(got, []string{"zeta", "alpha"}) {
+			t.Fatalf("live dependencies = %#v, save-time normalization leaked", got)
+		}
+		assertCheckpointManifestJSON(t, live, before)
+
+		changed := replacement["bob"]
+		changed.AudioPath = "audio/changed.flac"
+		replacement["bob"] = changed
+		if got := live.Stages["transcribe"].TranscribeSpeakers; len(got) != 1 || got["alice"].AudioPath != "audio/alice.flac" {
+			t.Fatalf("failed save changed live transcribe collection = %#v", got)
+		}
+	})
+}
+
+func TestCheckpointStageCollectionMissingTargetAppearsOnlyAfterSuccess(t *testing.T) {
+	now := time.Date(2026, 9, 22, 12, 0, 0, 0, time.UTC)
+	live := checkpointTestManifest(now)
+	delete(live.Stages, "transcribe")
+	path := filepath.Join(t.TempDir(), "manifest.json")
+	cause := errors.New("save failed")
+	failing := &checkpointBoundaryStore{Store: &manifest.LocalStore{}, ReturnErr: cause}
+	update := func(record *manifest.StageRecord) error {
+		record.Name = "transcribe"
+		record.Status = manifest.StatusRunning
+		record.CreatedAt = now
+		record.UpdatedAt = now
+		record.TranscribeStateVersion = manifest.TranscribeStateContractVersion
+		record.TranscribeSpeakers = checkpointTranscribeCollection("alice", "run-1")
+		return nil
+	}
+
+	if err := checkpointStageCollection(context.Background(), failing, path, live, "transcribe", update); !errors.Is(err, cause) {
+		t.Fatalf("failed checkpoint error = %v, want %v", err, cause)
+	}
+	if _, exists := live.Stages["transcribe"]; exists {
+		t.Fatal("missing live target was allocated after failed save")
+	}
+
+	successPath := filepath.Join(t.TempDir(), "manifest.json")
+	if err := checkpointStageCollection(context.Background(), &manifest.LocalStore{}, successPath, live, "transcribe", update); err != nil {
+		t.Fatalf("successful checkpoint error = %v", err)
+	}
+	created := live.Stages["transcribe"]
+	if created == nil || created.Name != "transcribe" || created.Status != manifest.StatusRunning {
+		t.Fatalf("promoted missing target = %#v", created)
+	}
+	durable, err := (&manifest.LocalStore{}).Load(context.Background(), successPath)
+	if err != nil {
+		t.Fatalf("Load() error = %v", err)
+	}
+	if !reflect.DeepEqual(durable.Stages["transcribe"], created) {
+		t.Fatalf("durable target = %#v, live = %#v", durable.Stages["transcribe"], created)
+	}
+}
+
+func TestCheckpointStageCollectionPreservesContextAndRequiredInputs(t *testing.T) {
+	now := time.Now().UTC()
+	live := checkpointTestManifest(now)
+	path := filepath.Join(t.TempDir(), "manifest.json")
+	update := func(*manifest.StageRecord) error { return nil }
+	store := &checkpointBoundaryStore{Store: &manifest.LocalStore{}}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	cancel()
+	before := checkpointManifestJSON(t, live)
+	err := checkpointStageCollection(ctx, store, path, live, "analyze", update)
+	if !errors.Is(err, context.Canceled) {
+		t.Fatalf("canceled error = %v, want context.Canceled", err)
+	}
+	if !strings.Contains(err.Error(), path) {
+		t.Fatalf("canceled error = %q, want manifest path context", err)
+	}
+	assertCheckpointManifestJSON(t, live, before)
+
+	if err := checkpointStageCollection(context.Background(), store, path, nil, "analyze", update); err == nil {
+		t.Fatal("nil live manifest error = nil")
+	}
+	if err := checkpointStageCollection(context.Background(), nil, path, live, "analyze", update); err == nil {
+		t.Fatal("nil store error = nil")
+	}
+	if err := checkpointStageCollection(context.Background(), store, path, live, "analyze", nil); err == nil {
+		t.Fatal("nil update error = nil")
+	}
+}
+
+type checkpointBoundaryStore struct {
+	manifest.Store
+	Calls          int
+	BeforeSave     func(*manifest.Manifest)
+	ReturnErr      error
+	SavedCandidate *manifest.Manifest
+}
+
+func (store *checkpointBoundaryStore) Save(ctx context.Context, path string, candidate *manifest.Manifest) error {
+	store.Calls++
+	if store.BeforeSave != nil {
+		store.BeforeSave(candidate)
+	}
+	if err := store.Store.Save(ctx, path, candidate); err != nil {
+		return err
+	}
+	store.SavedCandidate = candidate
+	return store.ReturnErr
+}
+
+func checkpointTestManifest(now time.Time) *manifest.Manifest {
+	live := manifest.New("session-1", now)
+	live.Campaign = "campaign-1"
+	live.RunID = "run-1"
+	live.Inputs = []manifest.InputRecord{{Kind: "audio", Path: "audio/alice.flac"}}
+	live.Artifacts = []manifest.ArtifactRecord{{Kind: "existing", LocalPath: "artifacts/existing.json"}}
+	live.Stages["analyze"] = &manifest.StageRecord{
+		Name: "analyze", Status: manifest.StatusRunning, CreatedAt: now, UpdatedAt: now,
+		Outputs:             []manifest.ArtifactRecord{{Kind: "analysis", LocalPath: "artifacts/aggregate.md"}},
+		Logs:                []string{"runs/run-1/analyze.log"},
+		GeneratedConfigs:    []string{"runs/run-1/analyze.yml"},
+		Metadata:            map[string]any{"shared_read_only": true},
+		AnalyzeStateVersion: manifest.AnalyzeStateContractVersion,
+		AnalyzeArtifacts: map[string]manifest.AnalyzeArtifactRecord{
+			"result": checkpointAnalyzeRecord(now, []string{"old_dependency"}),
+		},
+	}
+	live.Stages["transcribe"] = &manifest.StageRecord{
+		Name: "transcribe", Status: manifest.StatusFailed, CreatedAt: now, UpdatedAt: now,
+		TranscribeStateVersion: manifest.TranscribeStateContractVersion,
+		TranscribeSpeakers:     checkpointTranscribeCollection("alice", "run-1"),
+	}
+	live.Stages["merge"] = &manifest.StageRecord{
+		Name: "merge", Status: manifest.StatusSucceeded, CreatedAt: now, UpdatedAt: now,
+		Outputs: []manifest.ArtifactRecord{{Kind: "transcript_base", LocalPath: "transcripts/base.json"}},
+	}
+	return live
+}
+
+func checkpointAnalyzeRecord(now time.Time, dependencies []string) manifest.AnalyzeArtifactRecord {
+	return manifest.AnalyzeArtifactRecord{
+		Key: "result", Status: manifest.AnalyzeArtifactMissing,
+		Dependencies: dependencies, ProducerRunID: "run-1", UpdatedAt: now,
+	}
+}
+
+func checkpointTranscribeCollection(speaker, runID string) map[string]manifest.TranscribeSpeakerRecord {
+	return map[string]manifest.TranscribeSpeakerRecord{
+		speaker: {
+			FingerprintVersion: manifest.TranscribeFingerprintContractVersion,
+			Fingerprint:        strings.Repeat("b", 64),
+			AudioPath:          "audio/" + speaker + ".flac",
+			AudioSHA256:        strings.Repeat("a", 64),
+			AudioSize:          0,
+			Output: manifest.ArtifactRecord{
+				Kind: "transcript_raw", LocalPath: "transcripts/raw/" + speaker + ".json",
+				Checksum: strings.Repeat("c", 64), ProducerRunID: runID,
+			},
+			OutputSize: 1, ProducerRunID: runID,
+		},
+	}
+}
+
+func checkpointManifestJSON(t *testing.T, value *manifest.Manifest) []byte {
+	t.Helper()
+	data, err := json.Marshal(value)
+	if err != nil {
+		t.Fatalf("Marshal() error = %v", err)
+	}
+	return data
+}
+
+func assertCheckpointManifestJSON(t *testing.T, value *manifest.Manifest, want []byte) {
+	t.Helper()
+	if got := checkpointManifestJSON(t, value); !slices.Equal(got, want) {
+		t.Fatalf("live manifest changed on failure\ngot:  %s\nwant: %s", got, want)
+	}
+}

internal/stage/transcribe_execution_state.go

@@ -156,39 +156,30 @@ func (state *transcribeExecutionState) checkpointCollectionLocked(
 		return fmt.Errorf("validate transcribe speaker collection: %w", err)
 	}
 
-	candidate := cloneManifestForTranscribeCheckpoint(state.manifest)
+	missingRecord := transcribeStageRecord(state.manifest) == nil
-	if candidate.Stages == nil {
+	if err := checkpointStageCollection(
-		candidate.Stages = make(map[string]*manifest.StageRecord)
+		ctx,
-	}
+		state.env.ManifestStore,
-	record := candidate.Stages["transcribe"]
+		state.paths.ManifestPath,
-	if record == nil {
+		state.manifest,
+		"transcribe",
+		func(record *manifest.StageRecord) error {
+			if missingRecord {
 				now := time.Now().UTC()
-		record = &manifest.StageRecord{
+				*record = manifest.StageRecord{
 					Name: "transcribe", Status: manifest.StatusRunning, CreatedAt: now, UpdatedAt: now,
 				}
-		candidate.Stages["transcribe"] = record
 			}
 			record.TranscribeStateVersion = manifest.TranscribeStateContractVersion
 			record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(records)
 			record.UpdatedAt = time.Now().UTC()
-	if err := state.env.ManifestStore.Save(ctx, state.paths.ManifestPath, candidate); err != nil {
+			return nil
-		return fmt.Errorf("save session manifest %q: %w", state.paths.ManifestPath, err)
+		},
+	); err != nil {
+		return err
 	}
 
 	live := transcribeStageRecord(state.manifest)
-	if live == nil {
-		live = &manifest.StageRecord{}
-		if state.manifest.Stages == nil {
-			state.manifest.Stages = make(map[string]*manifest.StageRecord)
-		}
-		state.manifest.Stages["transcribe"] = live
-	}
-	live.TranscribeStateVersion = candidate.Stages["transcribe"].TranscribeStateVersion
-	live.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(
-		candidate.Stages["transcribe"].TranscribeSpeakers,
-	)
-	live.UpdatedAt = candidate.Stages["transcribe"].UpdatedAt
-	state.manifest.UpdatedAt = candidate.UpdatedAt
 	state.records = manifest.CloneTranscribeSpeakerCollection(live.TranscribeSpeakers)
 	return nil
 }
@@ -259,27 +250,6 @@ func transcribeStageRecord(m *manifest.Manifest) *manifest.StageRecord {
 	return m.Stages["transcribe"]
 }
 
-func cloneManifestForTranscribeCheckpoint(source *manifest.Manifest) *manifest.Manifest {
-	clone := *source
-	clone.Inputs = append([]manifest.InputRecord(nil), source.Inputs...)
-	clone.Artifacts = append([]manifest.ArtifactRecord(nil), source.Artifacts...)
-	clone.Stages = make(map[string]*manifest.StageRecord, len(source.Stages))
-	for name, sourceRecord := range source.Stages {
-		if sourceRecord == nil {
-			clone.Stages[name] = nil
-			continue
-		}
-		record := *sourceRecord
-		record.Outputs = append([]manifest.ArtifactRecord(nil), sourceRecord.Outputs...)
-		record.Logs = append([]string(nil), sourceRecord.Logs...)
-		record.GeneratedConfigs = append([]string(nil), sourceRecord.GeneratedConfigs...)
-		record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(sourceRecord.TranscribeSpeakers)
-		record.AnalyzeArtifacts = manifest.CloneAnalyzeArtifactCollection(sourceRecord.AnalyzeArtifacts)
-		clone.Stages[name] = &record
-	}
-	return &clone
-}
-
 func cloneStringAnyMap(source map[string]any) map[string]any {
 	if source == nil {
 		return nil

internal/stage/transcribe_reconciliation_test.go

@@ -58,7 +58,7 @@ func TestTranscribeResumeAcceptsEquivalentOutputsAndInspectsRemovalReadOnly(t *t
 		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)
+	before := cloneManifestForCheckpoint(m)
 	validation, err = (transcribeStage{}).ValidateResume(context.Background(), env, m)
 	if err != nil || validation.Resumable {
 		t.Fatalf("removed speaker: %#v, %v", validation, err)