Add shared stage checkpoint persistence
This commit is contained in:
@@ -1,6 +1,6 @@
|
||||
# Implementation Plan: Shared Stage Checkpoint Mechanics
|
||||
|
||||
Status: Not started.
|
||||
Status: Implementation in progress; Stage 1 complete.
|
||||
|
||||
## Scope and Authority
|
||||
|
||||
@@ -152,7 +152,16 @@ to simplify copying.
|
||||
|
||||
## Stage 1: Add the Shared Checkpoint Operation
|
||||
|
||||
Status: Pending.
|
||||
Status: Complete.
|
||||
|
||||
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/store.go`, `manifest.go`, `analyze_state.go`,
|
||||
`transcribe_state.go`; `internal/stage/analyze_execution_state.go`,
|
||||
|
||||
99
internal/stage/checkpoint.go
Normal file
99
internal/stage/checkpoint.go
Normal 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
|
||||
}
|
||||
327
internal/stage/checkpoint_test.go
Normal file
327
internal/stage/checkpoint_test.go
Normal 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)
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user