6 Commits

Author SHA1 Message Date
70f1dbfb59 Prepare the v2.2.1 release
All checks were successful
ci/woodpecker/push/verify Pipeline was successful
ci/woodpecker/tag/release Pipeline was successful
2026-09-22 18:18:45 +00:00
f1575effc2 Document shared checkpoint persistence 2026-09-22 15:13:46 +00:00
7e39a1d51f Delegate transcription checkpoints to shared persistence 2026-09-22 15:12:11 +00:00
3722e572df Delegate analyze checkpoints to shared persistence 2026-09-22 15:09:29 +00:00
7147e95252 Add shared stage checkpoint persistence 2026-09-22 15:05:25 +00:00
9f2cfddce6 Plan shared stage checkpoint mechanics 2026-09-22 14:54:34 +00:00
11 changed files with 793 additions and 438 deletions

View File

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

View File

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

View File

@@ -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
state are serialized. Expensive hashing and HTTP work remain outside that
critical section.
- candidate collection construction, the shared
[checkpoint persistence boundary](manifest.md#checkpoint-persistence-boundary),
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.

35
docs/releases/v2.2.1.md Normal file
View File

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

View File

@@ -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
`docs/policy/`, the roadmap, and the stage-specific preparation below. Use the
maintained Go 1.26 toolchain. Preserve operator choices, speaker filenames,
bounded concurrency, streaming, and existing credentials/settings. Do not add
new commands, dependencies, service requirements, arbitrary speaker limits,
strict transcript schemas, or a generic checkpoint framework. No release or
upstream service work is part of this plan.
Before each stage, read `AGENTS.md`, `docs/development.md`, and all three
documents in `docs/policy/`: architecture, documentation, and testing. Use
maintained Go 1.26. Prefer graph tools for code discovery; inspect current
files if their index is stale. Read the stage-specific preparation below,
rather than unrelated historical roadmaps.
## 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`
to `manifest.StageRecord`. The latter is a map keyed by the existing derived
speaker ID, containing completed results only. Absence means that work must be
done; a separate pending/running/failed state machine is unnecessary. Aggregate
stage status/error remains authoritative for invocation failure. A versioned
empty collection is meaningful: no earlier completions are reusable.
One private helper in `internal/stage` owns candidate preparation, calling the
existing store, and promoting the selected stage record only after successful
save. The existing typed analyze/transcribe checkpoint methods become thin
callers. This is execution plumbing: keep it beside the callers rather than
moving orchestration into the storage package or exporting a reusable API.
Use a typed `TranscribeSpeakerRecord` with fingerprint version/digest,
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 these responsibilities stage-owned:
Keep the session-manifest schema identifier unchanged for this optional additive
state; validate supported per-stage versions and retain strict decoding. New
Narratio reads old manifests. Old binaries may reject the added fields under
their existing strict decoding; document this downgrade limitation.
- Transcribe's completed-only speaker map, evidence checks, bounded concurrency,
mutex, checkpoint-failure latch, cancellation, pre-dispatch force reset,
output installation, and aggregate repair.
- 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
once when transcribe is next selected, even if the aggregate status succeeded.
Old records do not bind individual transcript results to the audio bytes they
used. Do not guess that association from current files, old `per_file` metadata,
or abandoned run directories. This is the conservative upgrade path, not a new
general migration system. No other stage or unselected range is forcibly run.
Do not introduce a common subtask model, scheduler, transaction manager, global
mutex, automatic retry, additional checkpoint, or batching policy. Do not
migrate publish/restore, planning clones, cleanup, or ordinary runner saves:
their operations are not the demonstrated duplication. Leave HTTP replay and
all adapters unchanged.
### Reuse and Persistence
## Selected Design
Compute a versioned per-speaker fingerprint from speaker ID, session-relative
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.
### Shared Save-and-Promote Operation
Validate reusable output identity, canonical location, JSON, checksum, and size.
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.
Add `internal/stage/checkpoint.go` with a private helper shaped as:
One stage-owned mutex serializes candidate collection construction, manifest
save, and installation of committed state. Save a detached candidate and only
then update live authority, following analyze's pattern. Workers must not mutate
the live manifest outside that serialized section. Keep expensive HTTP work
outside the mutex. Stop dispatch/acceptance on checkpoint failure, join workers
before return, and preserve the last successfully committed collection.
```go
func checkpointStageCollection(
ctx context.Context,
store manifest.Store,
path string,
live *manifest.Manifest,
stageName string,
update func(*manifest.StageRecord) error,
) error
```
Before launching requests, durably checkpoint the reconciled collection. For
force, that collection is empty: this prevents later ordinary retries from
resurrecting pre-force outputs. If this initial save fails, do not start requests
or overwrite outputs. Force again clears the current collection again; ordinary
retry retains replacements already checkpointed after the previous force.
Names are implementation choices, not public contracts. Use a function and
small copying helpers, not a factory/options/interface framework. The two
existing typed checkpoint methods are the only production callers. They pass
the already-resolved manifest path; the helper does not discover paths or
inspect configuration.
Save each validated result at its canonical raw-transcript location before
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.
The helper performs this sequence:
### 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
body backed by the audio reader. Use `mime/multipart.Writer` to encode a bounded
metadata prefix and suffix around the file payload: capture the prefix after
`CreateFormFile`, then encode the language field and closing boundary as the
suffix. Generate this framing once per application attempt. Each body combines
fresh readers for prefix, freshly opened audio, and suffix using `io.MultiReader`.
No full-audio buffer or temporary multipart spool is needed.
Callers serialize the whole operation. Transcribe's current mutex does so;
analyze executes artifacts sequentially. Do not add a lock to the helper or
promise support for concurrent independent manifest writers. Atomic replacement
visibility does not become power-loss durability or a transaction covering
transcript files.
The initial body and `Request.GetBody` use the same factory/framing. Each owns
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.
### Bounded Candidate Isolation
Remove the old producer `Wait` dependency. Track body-local source errors with
synchronization so a swallowed upload read error cannot become a successful
transcript, and a deliberately closed original body cannot poison a successful
replay. Keep the outer retry policy and per-attempt context. Do not add replay
counters to the user retry budget or depend on Go's exact internal retry count.
Preserve ordinary HTTP redirect policy; body replay can also enable Go's
body-preserving redirects. Do not introduce a new redirect restriction.
Factor `cloneManifestForTranscribeCheckpoint` into a shared checkpoint-scoped
copy helper in the new file. Use ordinary typed copies and existing
`manifest.CloneAnalyzeArtifactCollection` and
`manifest.CloneTranscribeSpeakerCollection`. Do not introduce reflection
cloning, JSON round trips, or a public general-purpose clone API.
## 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
per-speaker completion records, strict load/save validation, isolated collection
cloning, legacy aggregate-only compatibility, and lifecycle-preservation
coverage. Verified with `go test ./internal/manifest` and `git diff --check`.
Completion marker (2026-09-22): Added the private stage-level checkpoint
save-and-promote helper and bounded candidate copying without migrating either
production caller. Focused real-store coverage proves live isolation through
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`,
their tests, and `docs/internal/manifest.md`.
Preparation: `internal/manifest/store.go`, `manifest.go`, `analyze_state.go`,
`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,
and deep-copy helper described above in a focused transcribe state file.
2. Validate ownership by the transcribe stage, supported versions, required
completed-record evidence, output kind, safe relative paths, SHA-256 values,
sizes, and producer identity consistency. Use existing speaker IDs as-is;
do not apply analysis-artifact key regexes or copy unrelated analysis limits.
3. Integrate validation with both load and save. Preserve speaker state through
running/failed/stale transitions and aggregate result clearing. Generic stage
invalidation does not discard checkpoints; reconciliation checks their data.
4. Test round-trip populated and versioned-empty collections, old manifests
without fields, malformed evidence, deep-copy isolation, and lifecycle
preservation. Avoid imposing JSON formatting or map order as contracts.
1. Add the private helper and bounded candidate copying described above.
Leave existing production checkpoint paths unchanged until their migration
stages. Do not duplicate manifest record-validation rules.
2. Add focused `internal/stage/checkpoint_test.go` coverage with real
`manifest.LocalStore` and temporary directories for successful save/reload.
Use a small Store wrapper only to observe/fail the persistence boundary.
3. Prove live progress remains old during Save, matches saved progress after
success, and does not change on update, validation, or save failure. Include
both typed collections and a transcribe checkpoint alongside analyze
records subject to save-time normalization. A failure test must exercise
candidate timestamp/normalization mutation before returning an error, not
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`.
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
Verify: `go test ./internal/manifest ./internal/stage` and
`go test -race ./internal/stage`.
Done when incremental progress and aggregate success work together safely under
ordinary, concurrent, canceled, and forced execution.
Done when the shared integrity boundary is tested without changing existing
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
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`.
Completion marker (2026-09-22): Replaced analyze's local candidate-copy,
save, and promotion choreography with the shared checkpoint operation while
retaining its existing-record requirement, typed validation, path resolution,
and post-save working and committed collection refresh. Existing recovery,
selection, invalidation, and rollback coverage required no duplication.
Verified with `go test ./internal/stage ./internal/app ./internal/manifest`,
`go test -race ./internal/stage ./internal/app`, and `git diff --check`.
Preparation: `internal/app/runner.go`, `runner_runtime.go`,
`analyze_checkpoint_recovery_test.go`, `runner_terminal_failure_test.go`,
`restore_manifest_paths.go`, `internal/stage/merge.go`, and existing publish,
restore, and cleanup tests routed from `docs/internal/overview.md`.
Preparation: `internal/stage/analyze_execution_state.go`,
`analyze_checkpoint_test.go`, `analyze_failure_test.go`,
`internal/app/analyze_checkpoint_recovery_test.go`, and
`docs/internal/stage-analyze.md`.
1. Add application-level tests with real manifest persistence and fake WhisperX:
failed stage across separate invocations; all speakers committed before a
failed aggregate save; interrupted running state; force/reset persistence;
and unchanged upstream/succeeded downstream invalidation behavior. Inject
errors at durable boundaries, not every private method call.
2. Close `discoverRawTranscripts`' directory-scan fallback for versioned
transcribe state: incomplete aggregate state is not usable, and completed
state consumes only declared current aggregate outputs. Keep legacy/manual
fallback behavior for manifests without the new state rather than banning
previously allowed standalone workflows. Test that partial files and removed
speaker files cannot leak into a versioned merge, including direct stage calls.
3. Audit manifest copies/projections, remote manifest publication, restored path
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.
1. Replace manual candidate copying, saving, and live promotion in analyze
with the shared helper. Retain typed validation, required-record checking,
the typed update callback, and committed/working collection refresh.
2. Preserve starting-checkpoint elision, selected-artifact behavior, catalog
sequencing, dependent invalidation on changed output, and rollback after
checkpoint failure. Do not change the runner's terminal failure save or
joined errors.
3. Run existing tests before adding coverage. Retain tests proving earlier
artifacts survive failure/cancellation, unsaved output bytes gain no
authority, selected/forced work preserves appropriate unrelated artifacts,
and interrupted runs recover across store reloads. Add only missing
integration coverage; do not duplicate Stage 1's isolation matrix.
Verify: `go test ./internal/app ./internal/stage ./internal/manifest` and
`go test -race ./internal/app`.
Verify: `go test ./internal/stage ./internal/app ./internal/manifest` and
`go test -race ./internal/stage ./internal/app`.
Done when the new authority survives runner and storage lifecycles and cannot
expose incomplete transcripts downstream.
Done when analyze delegates persistence choreography while progress, artifact
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
with bounded standard-library multipart framing and independently opened,
pull-based streaming bodies shared by the initial request and `GetBody`.
Bodies are tracked, idempotently closed, canceled with their attempt context,
and retain genuine read/close failures without treating an intentionally
abandoned original as a replay failure. Added focused coverage for identical
framing, partial-read restart, simultaneous handles, source failures, and
blocked-reader release. Verified with `go test ./internal/adapters/whisperx`,
`go test -race ./internal/adapters/whisperx` (including ten repeated runs), and
`git diff --check`.
Completion marker (2026-09-22): Delegated transcription's locked checkpoint
operation to the shared helper while retaining typed validation, direct-call
record initialization, stage timestamp updates, post-save working collection
refresh, and the existing caller-owned mutex and failure latch. Removed the
duplicate production manifest clone; the read-only resume assertion now uses
the shared bounded snapshot. Existing real-store, concurrency, force/retry,
interruption, aggregate-repair, and zero-sized-audio coverage passed unchanged.
Verified with `go test ./internal/stage ./internal/app ./internal/manifest`,
`go test -race ./internal/stage ./internal/app`, and `git diff --check`.
Preparation: `internal/adapters/whisperx/http.go`, `http_test.go`,
`docs/integrations/whisperx.md`, and the installed Go net/http Request/GetBody
contract. Keep this change inside the WhisperX adapter.
Preparation: `internal/stage/transcribe_execution_state.go`, `transcribe.go`,
`transcribe_resume.go`, `transcribe_checkpoint_test.go`,
`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
multipart encoding rather than hand-escaped MIME headers. Add no new payload
cap or upload-size setting. Preserve file-before-language wire semantics and
the existing form names.
2. Install the initial body and GetBody with the same per-attempt boundary.
Track and close all opened bodies even if a custom transport returns early;
make cleanup/context callbacks safe against overlapping reads and closes.
A GetBody failure must be returned, not converted into an empty upload.
3. Replace pipe-specific Wait/error paths in doTranscribeAttempt with source-error
handling appropriate to pull bodies. Genuine source errors remain visible;
intentional closure during transport retry is not itself a source error.
Keep response bounds, JSON validation, output installation, retry
classification/delay, cancellation, and application attempt reporting intact.
4. Remove obsolete producer-only helpers rather than retaining two upload paths.
Adapt tests tied to producer goroutines to assert streaming and resource
release, which are the actual contracts.
5. Add focused body tests: original/replayed bytes parse identically; replay
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.
1. Delegate `checkpointCollectionLocked` to the shared helper, retaining
typed initialization/update and successful working-collection refresh.
Keep the existing mutex across candidate creation, save, and promotion.
Do not move HTTP/hashing work into that critical section or add a lock.
2. Preserve failure-latch checks and worker joining, the parent versus worker
context distinction, force's durable empty starting collection, and
ordinary retry's reuse of committed forced replacements.
3. Remove `cloneManifestForTranscribeCheckpoint`. Its current use by a
read-only resume test is not a reason to retain duplicate production code.
Use an independent test snapshot or the shared bounded copy where sufficient,
without exporting a helper for tests.
4. Retain existing real-store and concurrency coverage: no lost completions;
pre-dispatch save failure prevents requests/canonical writes; later failure
retains prior authority; interrupted and aggregate-save recovery; failed
force followed by retry; run-directory independence; zero-sized audio;
removed/mismatched aggregate repair without additional WhisperX requests
or removed-speaker merge input. Do not weaken these regression assertions.
Verify: `go test ./internal/adapters/whisperx` and
`go test -race ./internal/adapters/whisperx`.
Verify: `go test ./internal/stage ./internal/app ./internal/manifest` and
`go test -race ./internal/stage ./internal/app`.
Done when replay is available without whole-file buffering or producer-lifetime
ambiguity, and all existing HTTP behavior tests still pass.
Done when both stages use one save/promote implementation and one bounded
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
that abandons a partially consumed original request, obtains and parses a fresh
multipart replay from `GetBody`, and installs the returned transcript within one
application attempt. Covered replay open/read failures, early responses,
parent cancellation during replay, and disposal of every opened source, while
proving an abandoned original does not poison a valid replay. Added a real HTTP
retry-exhaustion integration that durably retains an earlier speaker and resumes
only the uncommitted speakers. This controlled boundary verifies Narratio's
replay contract, not recovery from every real proxy protocol failure. Verified
with `go test ./internal/adapters/whisperx ./internal/app ./internal/stage`,
`go test -race ./internal/adapters/whisperx ./internal/app ./internal/stage`,
ten repeated race-enabled adapter runs, and `git diff --check`.
Completion marker (2026-09-22): Documented the private shared checkpoint
persistence boundary and linked the stage guides back to it while retaining
their separate analyze and transcribe progress policies. The refactor removed
the two local candidate/save/promotion sequences and the transcription-only
bounded clone; the retained thin wrappers own typed validation, updates,
working-state refresh, and their existing serialization and failure behavior.
The collective test review retained focused helper integrity coverage plus
stage/application recovery tests, without duplicating the helper's isolation
matrix. A cumulative scope audit found no schema, validation-policy,
scheduling, force, concurrency, output-selection, stage-order, adapter-call,
configuration, or public-interface change.
Preparation: Existing injectable HTTPClient/RoundTripper seam and local-server
tests in `internal/adapters/whisperx/http_test.go`; app WhisperX wiring tests.
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`,
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,
reconciliation, durable-reset force behavior, serialized saves, and aggregate
gate. Remove obsolete all-speakers-before-canonical-write descriptions.
2. Update the WhisperX contract for replayable bounded streaming, application
attempts versus transport replay, resource ownership, and the lack of an
exactly-once server guarantee. Do not claim the remote protocol issue is fixed.
3. Update operations/troubleshooting with ordinary resume versus force, durable
partial success, old-manifest one-time retranscription on selecting transcribe,
downgrade limitations, and private-model-change guidance. No new CLI flags
or promises of automatic historical-file recovery. Keep detailed wire and
storage contracts in their canonical owners.
4. Review changes against the feature roadmap and the implementation-discipline
guidance. Do not elevate selected storage encoding, internal ordering, test
synchronization, or exact error wording into new product guarantees.
5. Run the full repository gates below; document results and update completion
statuses only after they pass. Leave release preparation for a separate task.
Preparation: `docs/internal/manifest.md`, `docs/internal/stage-transcribe.md`,
`docs/internal/stage-analyze.md`, documentation/testing policies, and the
implementation diff against this plan's starting commit.
1. Update manifest internals to identify the shared checkpoint owner and copy/
save/promote boundary. Stage guides retain stage-specific policy and link
to that explanation instead of duplicating shared implementation details.
Do not rewrite immutable release notes or announce new operator behavior.
2. Assess tests collectively. Consolidate genuinely duplicated persistence
mechanics coverage, but retain stage/application recovery tests proving
correct integration. Avoid test-count/LOC quotas. Record a brief summary
here of duplicated mechanisms removed and thin wrappers retained.
3. Confirm no schema, validation policy, scheduling, force, concurrency, output
selection, stage order, adapter call, or configuration change slipped into
the refactor. Additional restrictions require a separate decision, not
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;
this plan selects a completed-only manifest collection, conservative old-record
handling, and pull-based replay as implementation choices. No stronger remote
reliability guarantee, new operator restriction, or generic orchestration
surface is proposed.
None requiring a product decision. A private stage-level helper is selected
because persistence already has an owner and the duplication is execution
plumbing. Model unification, library extraction, generalized transactions, and
broader cloning cleanup remain out of scope.

View File

@@ -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
[implementation plan](implementation.md) owns ordered changes and concrete
storage and replay decisions.
service failures. This document records the target state and scope. The
completed implementation sequence is retained in Git history; current behavior
is documented in the canonical guides linked below.
## Pre-implementation Limitations

View File

@@ -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 {
return fmt.Errorf("save session manifest %q: %w", manifestPath, err)
if err := checkpointStageCollection(
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

View File

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

View File

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

View File

@@ -156,39 +156,30 @@ func (state *transcribeExecutionState) checkpointCollectionLocked(
return fmt.Errorf("validate transcribe speaker collection: %w", err)
}
candidate := cloneManifestForTranscribeCheckpoint(state.manifest)
if candidate.Stages == nil {
candidate.Stages = make(map[string]*manifest.StageRecord)
}
record := candidate.Stages["transcribe"]
if record == nil {
now := time.Now().UTC()
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 fmt.Errorf("save session manifest %q: %w", state.paths.ManifestPath, err)
missingRecord := transcribeStageRecord(state.manifest) == nil
if err := checkpointStageCollection(
ctx,
state.env.ManifestStore,
state.paths.ManifestPath,
state.manifest,
"transcribe",
func(record *manifest.StageRecord) error {
if missingRecord {
now := time.Now().UTC()
*record = manifest.StageRecord{
Name: "transcribe", Status: manifest.StatusRunning, CreatedAt: now, UpdatedAt: now,
}
}
record.TranscribeStateVersion = manifest.TranscribeStateContractVersion
record.TranscribeSpeakers = manifest.CloneTranscribeSpeakerCollection(records)
record.UpdatedAt = time.Now().UTC()
return nil
},
); 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

View File

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