From 7147e952521b06021a75f08039ddd62b9b299d35 Mon Sep 17 00:00:00 2001 From: Eric Rakestraw Date: Tue, 22 Sep 2026 15:05:25 +0000 Subject: [PATCH] Add shared stage checkpoint persistence --- docs/roadmap/implementation.md | 13 +- internal/stage/checkpoint.go | 99 +++++++++ internal/stage/checkpoint_test.go | 327 ++++++++++++++++++++++++++++++ 3 files changed, 437 insertions(+), 2 deletions(-) create mode 100644 internal/stage/checkpoint.go create mode 100644 internal/stage/checkpoint_test.go diff --git a/docs/roadmap/implementation.md b/docs/roadmap/implementation.md index a75eba9..51edbd9 100644 --- a/docs/roadmap/implementation.md +++ b/docs/roadmap/implementation.md @@ -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`, diff --git a/internal/stage/checkpoint.go b/internal/stage/checkpoint.go new file mode 100644 index 0000000..5d5d932 --- /dev/null +++ b/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 +} diff --git a/internal/stage/checkpoint_test.go b/internal/stage/checkpoint_test.go new file mode 100644 index 0000000..82b8b52 --- /dev/null +++ b/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) + } +}